op_impl.h

/*
 * Copyright (c) 2015, 2024, Oracle and/or its affiliates.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License, version 2.0, as
 * published by the Free Software Foundation.
 *
 * This program is designed to work with certain software (including
 * but not limited to OpenSSL) that is licensed under separate terms, as
 * designated in a particular file or component or in included license
 * documentation. The authors of MySQL hereby grant you an additional
 * permission to link the program and your derivative works with the
 * separately licensed software that they have either included with
 * the program or referenced in the documentation.
 *
 * Without limiting anything contained in the foregoing, this file,
 * which is part of Connector/C++, is also subject to the
 * Universal FOSS Exception, version 1.0, a copy of which can be found at
 * https://oss.oracle.com/licenses/universal-foss-exception.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 * See the GNU General Public License, version 2.0, for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/

#ifndef MYSQLX_COMMON_OP_IMPL_H
#defineMYSQLX_COMMON_OP_IMPL_H

#include"common.h"
#include"session.h"
#include"result.h"
#include"db_object.h"

#include<mysql/cdk.h>
#include<mysqlx/common.h>
#include<mysqlx/common/op_if.h>

PUSH_SYS_WARNINGS
#include<bitset>
#include<list>
POP_SYS_WARNINGS


/*
 This file defines a hierarchy of classes which implement executable objects
 that execute various CRUD operations.
*/

namespacemysqlx {
namespaceimpl {
namespacecommon {

enumclassObject_type
{
 SCHEMA,
 COLLECTION,
 TABLE,
 VIEW
};


/*
 Base for CRUD operation implementation classes.

 This class handles the final execution of an operation, which
 is performed as follows (see method `wait`).

 1. First, appropriate CRUD operation is sent to the server using
 underlying CDK session. This produces a cdk::Reply object which
 is used for further processing. Sending the CRUD operation is
 performed by method `send_command` which should be overwritten by
 derived class. Derived class has access to the underlying CDK session
 with method `get_cdk_session()`.

 2. After getting cdk::Reply object implementation waits for it to
 receive server reply and then returns Result_base instance created
 from the cdk::Reply object.

 The Op_base template is parametrized by the implementation interface
 `IF` that derived class wants to implement (see executable.h for interface
 definitions). The Op_base template implements some of the interface methods,
 other templates and derived class should implement the rest.
*/

template <classIF>
classOp_base
 : public IF
 , protected Result_init
{
public:

enum Prepare_state
 {
 PS_EXECUTE,
 PS_PREPARE_EXECUTE,
 PS_EXECUTE_PREPARED
 };

protected:

using string = std::string;
using Shared_session_impl = shared_ptr<Session_impl>;
using Shared_stmt_id = shared_ptr<uint32_t>;

 Shared_session_impl m_sess;

/*
 Note: using cdk::scoped_ptr to be able to transfer ownership
 to a different object.
*/
 cdk::scoped_ptr<cdk::Reply> m_reply;

 Shared_stmt_id m_stmt_id;
 Prepare_state m_prepare_state = PS_EXECUTE;
bool m_inited = false;
bool m_completed = false;

public:

Op_base(const Shared_session_impl &sess)
 : m_sess(sess)
 {}

/*
 Note: When objects of classes exteding Op_base are copied, only definiton
 of the operation to be executed is copied. The execution state such as
 m_reply, m_inited etc. is not copied from the other instance, but is
 initialized so that the copied operation is ready to be executed.
*/

Op_base(const Op_base& other)
 : m_sess(other.m_sess)
 , m_stmt_id(other.m_stmt_id)
 , m_prepare_state(other.m_prepare_state)
 {}

virtual~Op_base() override
 {
try {
// Let's aquire lock so that any remaingin replies are consumed on ~Reply
// and this way avoid race conditions with Client::close()
auto lock = m_sess->lock();
release_stmt_id();
 m_reply.reset();
 }
catch (...)
 {}
 }

 cdk::Session& get_cdk_session()
 {
assert(m_sess);
if (!m_sess->m_sess->is_valid()){
THROW("Session is no longer valid");
 }
return *(m_sess->m_sess);
 }

uint32_tcreate_stmt_id()
 {
assert(m_sess);
if (m_stmt_id.use_count() != 1) {
uint32_t id = m_sess->create_stmt_id();
if(id != 0)
 m_stmt_id.reset(newuint32_t(id));
else
 m_stmt_id.reset();
 }
returnget_stmt_id();
 }

voidrelease_stmt_id()
 {
if (m_stmt_id.use_count() == 1)
 m_sess->release_stmt_id(*m_stmt_id);
 m_stmt_id.reset();
 }


/*
 Clears operation state and, if stmt_id != 0, informe session about error on
 PS
*/

voidreset_state()
 {
if (m_stmt_id.use_count()==1)
get_session()->error_stmt_id(*m_stmt_id);
 m_stmt_id.reset();
 m_prepare_state = PS_EXECUTE;
 m_reply.reset();
 m_inited = false;
 m_completed = false;
 }

uint32_tget_stmt_id()
 {
return m_stmt_id ? *m_stmt_id.get() : 0;
 }

 Prepare_state get_prepare_state()
 {
return m_prepare_state;
 }

voidset_prepare_state(Prepare_state x)
 {
 m_prepare_state = x;
 }

// Async execution

/*
 Initialize statement execution (if not already done) by sending command
 to the server. This initializes m_reply to point to a cdk::Reply object
 that waits for the server reply.
*/

voidinit()
 {
if (m_inited)
return;
 m_inited = true;

assert(m_sess);

/*
 Prepare session for sending a new command. This gives session a chance
 to do necessary cleanups, such as consuming pending reply to a previous
 command.

 TODO: It should be possible to send next command while previous results
 are still being consumed (rd/wr split). But this requires changes in CDK.
 At present calling send_command(), which creates a new CDK reply object,
 discards a previous reply (if any) and this confuses the overall result
 processing logic.
*/

 m_sess->prepare_for_cmd();
 m_reply.reset(send_command());
 }

boolis_completed()
 {
if (m_completed)
returntrue;

init();
 m_completed = (!m_reply) || m_reply->is_completed();
return m_completed;
 }


/*
 Drive statement execution operation. First call init() to initialize it
 if it was not done before. Then wait for the reply object to become ready.
*/

voidcont()
 {
if (m_completed)
return;
init();
if (m_reply)
 {
 m_reply->cont();
bool retry = false;
try {
check_errors();

 } catch (cdk::mysqlx::Server_prepare_error&)
 {
 retry = true;
 }
catch (cdk::mysqlx::Server_expectation_error&)
 {
 retry = true;
 }

if(retry)
 {
/*
 If we* are executing prepare + execute pipeline and prepare step
 failed, try again, this time executing without prepare.
 reset_state will set stmt_id= 0 which will trigger the direct execute
*/
reset_state();
cont();
 }

 }
 }

/*
 Drive statement execution until server reply is available.
*/

voidwait()
 {
init();
if (m_reply)
 {
 m_reply->wait();

bool retry = false;
try {
check_errors();

 } catch (cdk::mysqlx::Server_prepare_error&)
 {
 retry = true;
 }
catch (cdk::mysqlx::Server_expectation_error&)
 {
 retry = true;
 }

if(retry)
 {
/*
 If we are executing prepare + execute pipeline and prepare step
 failed, try again, this time executing without prepare.
 reset_state will set stmt_id= 0 which will trigger the direct execute
*/
reset_state();
wait();
 }


 }
 }


// Synchronous execution

/*
 If an error has happened and it was ignored, then the initializer will
 return NULL from get_reply().
*/

 Result_init& execute() override
 {
auto lock = m_sess->lock();
// Can not execute operation that is already completed.
assert(!m_completed);



execute_prepare();
wait();
execute_cleanup();

return *this;
 }

protected:

/*
 For PS operations, do_send_command should be overriden to send a command to
 the server and return a cdk object representing server's reply to that
 command. For non-PS, send_command should then be overriden. The Op_base
 instance takes ownership of the returned reply object.

 TODO: Currently send_command() allocates new cdk::Reply object on heap
 and then passes it to result object which takes ownership. Avoid dynamic
 allocation: return cdk reply initializer instead and use it to initialize
 an instance of cdk::Reply inside result object (in its implementation
 actually). This way only result implementation has to be allocated on heap.
*/

virtual cdk::Reply* send_command() = 0;

virtual cdk::Reply* do_send_command()
 {
assert(false);
returnnullptr;
 }

/*
 Either call do_send_command() to send (and possibly prepare) a command or,
 if there is an up-to-date prepared statement for the original command, send
 command that executes this prepared statement.
*/

 cdk::Reply* send_prepared_command(const cdk::Limit* limit,
const cdk::Param_source* param
 )
 {
if (use_prepared_statement())
 {
returnnewcdk::Reply(get_cdk_session().prepared_execute(
get_stmt_id(),
 limit,
 param
 ));
 }
returndo_send_command();
 }

 cdk::Reply* send_prepared_command(const cdk::Any_list* list)
 {
if (use_prepared_statement())
 {
returnnewcdk::Reply(get_cdk_session().prepared_execute(
get_stmt_id(),
 list
 ));
 }
returndo_send_command();
 }


/*
 Hooks that are called just before and after execution of the operation.

 TODO: Currently these hooks are not called when executing asynchronously.
*/

// LCOV_EXCL_START
virtualvoidexecute_prepare()
 {}

virtualvoidexecute_cleanup()
 {}
// LCOV_EXCL_STOP


/*
 Handling errors reported by server.

 Normally when server reports erros, operation throws error during execution.
 But a derived class can define a list of server errors which are ignored
 and let the operation execute successfully (with empty result).
*/

 std::set<cdk::error_code> m_skip_errors;

voidskip_error(const cdk::error_code &ec)
 {
 m_skip_errors.emplace(ec);
 }

voidclear_skip_errors()
 {
 m_skip_errors.clear();
 }

voidcheck_errors()
 {
if (0 == m_reply->entry_count())
return;

const cdk::Error &err = m_reply->get_error();

/*
 If error is found on the skip list, we do not throw it. But the CDK reply
 object is not valid anymore. We delete it so that later get_reply()
 returns NULL. This indicates an empty result after ignoring server error.
*/

if (m_skip_errors.end() != m_skip_errors.find(err.code()))
 m_reply.reset();
else
 err.rethrow();
 }


/*
 Result_init

 Op_base implements Result_init interface methods which are used to build
 a result object representing results of the operation. These results are
 obtained from server using the cdk reply object.
*/

 Shared_session_impl get_session() override
 {
return m_sess;
 }

 cdk::Reply* get_reply() override
 {
if (!is_completed())
THROW("Attempt to get result of incomplete operation");

/*
 Server reply to the command is now passed to the result instance.
 We reset m_inited and m_completed flag so that upon next execution the
 command will be sent to the server again and a new reply will be created.
*/

 m_inited = false;
 m_completed = false;

/*
 Note: caller takes ownership of the returned cdk::Reply object and for
 that reason we need to release() here to avoid double deletion of the
 object.
*/

return m_reply.release();
 }

private:

/*
 Returns true if the statement has been prepared before and that prepared
 statement should be used for execution, false if we need to execute and
 prepare the original statement (either for the first time, or because it has
 been modified).
*/
booluse_prepared_statement()
 {
auto prepare = get_prepare_state();

/*
 Upon first execution, prepare is on PS_EXECUTE state and get_stmt_id() is
 0. The new statement id is not allocated yet and function returns false,
 meaning that the statement will be executed directly without preparing.
 Also, prepare is set to PS_PREPARE_EXECUTE.

 On next execution, prepare is then on PS_PREPARE_EXECUTE. Then new
 PS id is allocated and function returns false, meaning that the statement
 gets prepared and executed. Also, the state is set to PS_EXECUTE_PREPARED.

 On 3rd and following executions, if state stays PS_EXECUTE_PREPARED, this
 function will return true meaning that the prepared statement should be
 used.
*/

if (prepare == PS_PREPARE_EXECUTE )
 {
create_stmt_id();
 }
elseif (prepare == PS_EXECUTE)
 {
release_stmt_id();
 }

switch(prepare)
 {
case PS_EXECUTE:
set_prepare_state(PS_PREPARE_EXECUTE);
break;
case PS_PREPARE_EXECUTE:
set_prepare_state(PS_EXECUTE_PREPARED);
break;
case PS_EXECUTE_PREPARED:
break;
 }

return prepare == PS_EXECUTE_PREPARED &&
get_stmt_id()!=0;
 }
};


/*
 This template adds to the given Base class implementations for Bind_if
 interface methods which handle storing values of named parameters. It
 works only for named parameters.

 Method get_params() returns stored parameter values in the form expected by
 CDK (cdk::Param_source). It returns NULL if no parameter values were defined.
*/

template <classBase>
classOp_bind
 : public Base
 , cdk::Param_source
{
protected:

using string = std::string;
using Shared_session_impl = typename Base::Shared_session_impl;

Op_bind(Shared_session_impl sess) : Base(sess)
 {}

typedef std::map<cdk::string, Value> param_map_t;
param_map_t m_map;


// Parameters

voidadd_param(const string &name, const Value &val) override
 {
auto el = m_map.emplace(name, val);
//substitute if exists
if (!el.second)
 {
 el.first->second = val;
 }
 }

voidadd_param(const string &name, const cdk::string &val)
 {
add_param(name, Value::Access::mk_str(val));
 }

voidadd_param(const string &name, const std::string &val)
 {
add_param(name, Value(val));
 }


voidadd_param(Value) override
 {
assert(false);
 }

voidadd_param(const cdk::string &val)
 {
add_param(Value::Access::mk_str(val));
 }

voidadd_param(const std::string &val)
 {
add_param(Value(val));
 }


voidclear_params() override
 {
 m_map.clear();
 }


// cdk::Param_source

voidprocess(Processor &prc) constoverride
 {
 prc.doc_begin();

for (auto it : m_map)
 {
 Value_scalar val(it.second);
 val.process_if(prc.key_val(it.first));
 }

 prc.doc_end();
 }

public:

 cdk::Param_source* get_params()
 {
return m_map.empty() ? nullptr : this;
 }

 cdk::Reply* send_command() override
 {
returnBase::send_prepared_command(nullptr, get_params());
 }
};


/*
 This tempate adds to the given Base class implementations for Limit_if
 interface methods which set limits on query results.

 Method get_limit() returns the limits in the form expected by CDK
 (cdk::Limit). It returns NULL if no limits were set.
*/

template <classBase>
classOp_limit
 : public Base
 , cdk::Limit
{
protected:

using Shared_session_impl = typename Base::Shared_session_impl;

Op_limit(Shared_session_impl sess) : Base(sess)
 {}

row_count_t m_limit = 0;
row_count_t m_offset = 0;
bool m_has_limit = false;
bool m_has_offset = false;

// Limit and offset

voidset_limit(unsigned lm) override
 {
/*
 Setting limit is not treated as changing the statement
 completely. Re-prepare is needed only if the statement
 was already prepared without any limits.
*/

if (nullptr == get_limit() &&
Base::get_prepare_state() == Base::PS_EXECUTE_PREPARED)
 {
Base::set_prepare_state(Base::PS_PREPARE_EXECUTE);
 }

 m_has_limit = true;
 m_limit = lm;
 }

voidclear_limit() override
 {
/*
 Clearing limit is not treated as changing the statement
 completely. Re-prepare is needed only if the statement
 was already prepared with limits and now it will have
 no limits (because no offset was set).
*/

if (nullptr != get_limit() && !m_has_offset &&
Base::get_prepare_state() == Base::PS_EXECUTE_PREPARED)
 {
Base::set_prepare_state(Base::PS_PREPARE_EXECUTE);
 }

 m_has_limit = false;
 }


voidset_offset(unsigned offset) override
 {
/*
 Setting offset is not treated as changing the statement
 completely. Re-prepare is needed only if the statement
 was already prepared without any limits.
*/

if (nullptr == get_limit() &&
Base::get_prepare_state() == Base::PS_EXECUTE_PREPARED)
 {
Base::set_prepare_state(Base::PS_PREPARE_EXECUTE);
 }

 m_has_offset = true;
 m_offset = offset;
 }

voidclear_offset() override
 {
/*
 Clearing offset is not treated as changing the statement
 completely. Re-prepare is needed only if the statement
 was already prepared with limits and now it will have
 no limits (because no row limit was set).
*/

if (nullptr != get_limit() && !m_has_limit &&
Base::get_prepare_state() == Base::PS_EXECUTE_PREPARED)
 {
Base::set_prepare_state(Base::PS_PREPARE_EXECUTE);
 }

 m_has_offset = false;
 }


 cdk::Limit* get_limit()
 {
return m_has_limit || m_has_offset ? this : nullptr;
 }


 cdk::Reply* send_command() override
 {
returnBase::send_prepared_command(get_limit(), Base::get_params());
 }


// cdk::Limit interface

row_count_tget_row_count() constoverride
 { return m_has_limit ? m_limit : std::numeric_limits<row_count_t>::max(); }

constrow_count_t* get_offset() constoverride
 {
return m_has_offset ? &m_offset : nullptr;
 }

};


/*
 This template adds to the given Base class implementations of Sort_if
 interface methods which specify sorting of a query results.

 Method get_order_by() returns sorting specifications in the form expected by
 CDK (cdk::Order_by). It returns NULL if no order specifications were given.

 The PM template parameter tells in which mode the expressions used in order
 specifications should be parsed.
*/

template <parser::Parser_mode::value PM, classBase>
classOp_sort
 : public Base
 , cdk::Order_by
{
protected:

using Shared_session_impl = typename Base::Shared_session_impl;
usingdirection_t = typename Base::direction_t;
using string = std::string;

structorder_item
 {
enum {
 ASC = cdk::api::Sort_direction::ASC,
 DESC = cdk::api::Sort_direction::DESC,
 PARSE = ASC + DESC + 1
 } m_dir;
 string m_expr;

order_item(const string &expr)
 : m_dir(PARSE), m_expr(expr)
 {}

order_item(const string &expr, direction_t dir)
 : m_dir(Base::ASC == dir ? ASC : DESC), m_expr(expr)
 {}
 };

 std::list<order_item> m_order;

voidadd_sort(const string &expr, direction_t dir) override
 {
Base::set_prepare_state(Base::PS_EXECUTE);
 m_order.emplace_back(expr, dir);
 }

voidadd_sort(const string &sort) override
 {
Base::set_prepare_state(Base::PS_EXECUTE);
 m_order.emplace_back(sort);
 }

voidclear_sort() override
 {
if (get_order_by())
Base::set_prepare_state(Base::PS_EXECUTE);
 m_order.clear();
 }

Op_sort(Shared_session_impl sess) : Base(sess)
 {}

public:

 cdk::Order_by* get_order_by()
 {
return m_order.empty() ? nullptr : this;
 }

protected:

// cdk::Order_by interface

voidprocess(Order_by::Processor& prc) constoverride
 {
 prc.list_begin();

for (const order_item &item : m_order)
 {
auto *el = prc.list_el();
if (!el)
continue;

switch (item.m_dir)
 {
case order_item::ASC:
case order_item::DESC:
 {
 parser::Expression_parser parser(PM, item.m_expr);
 parser.process_if(el->sort_key(
cdk::api::Sort_direction::value(item.m_dir)
 ));
 }
break;

case order_item::PARSE:
 {
 parser::Order_parser order_parser(PM, item.m_expr);
 order_parser.process_if(el);
 }
break;
 }
 }

 prc.list_end();
 }

using Base::process;

};

/*
 This template adds to the given Base class implementations of Having_if
 interface methods which specify grouping selection criteria.

 Method get_having() returns selection criteris as CDK expression.
 It returns NULL if no selection criteria was specified.

 The PM template parameter tells in which mode the selecting expression
 should be parsed.
*/

template <parser::Parser_mode::value PM, classBase>
classOp_having
 : public Base
 , cdk::Expression
{
protected:

using string = std::string;

 string m_having;

public:

using Shared_session_impl = typename Base::Shared_session_impl;

Op_having(Shared_session_impl sess) : Base(sess)
 {}

voidset_having(const string &having) override
 {
Base::set_prepare_state(Base::PS_EXECUTE);
 m_having = having;
 }

voidclear_having() override
 {
if (get_having())
Base::set_prepare_state(Base::PS_EXECUTE);
 m_having.clear();
 }

 cdk::Expression* get_having()
 {
return m_having.empty() ? nullptr : this;
 }

protected:

// cdk::Expression processor

voidprocess(cdk::Expression::Processor& prc) constoverride
 {
 parser::Expression_parser expr_parser(PM, m_having);
 expr_parser.process(prc);
 }

using Base::process;
};


/*
 This template adds to the given Base class implementations of Group_by_if
 interface methods which specify grouping of query results.

 Method get_group_by() returns grouping specification in the form expected
 by CDK (a list of expressions). It returns NULL if no grouping was specified.

 The PM template parameter tells in which mode the grouping expressions
 should be parsed.
*/

template <parser::Parser_mode::value PM, classBase>
classOp_group_by
 : public Base
 , cdk::Expr_list
{
using string = std::string;
 std::vector<string> m_group_by;

public:

using Shared_session_impl = typename Base::Shared_session_impl;

voidadd_group_by(const string &group_by) override
 {
Base::set_prepare_state(Base::PS_EXECUTE);
 m_group_by.push_back(group_by);
 }

voidclear_group_by() override
 {
if (get_group_by())
Base::set_prepare_state(Base::PS_EXECUTE);
 m_group_by.clear();
 }

Op_group_by(Shared_session_impl sess) : Base(sess)
 {}