oceanbase/unittest/sql/engine/test_exec_context.cpp

/**
 * Copyright (c) 2021 OceanBase
 * OceanBase CE is licensed under Mulan PubL v2.
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
 * You may obtain a copy of Mulan PubL v2 at:
 *          http://license.coscl.org.cn/MulanPubL-2.0
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PubL v2 for more details.
 */

#include <gtest/gtest.h>
#include "../optimizer/ob_mock_part_mgr.h"
#define protected public
#define private public
#include "sql/engine/test_engine_util.h"
#include "sql/engine/ob_exec_context.h"
#include "sql/engine/ob_phy_operator.h"
#include "sql/engine/ob_phy_operator_type.h"
#include "sql/plan_cache/ob_plan_cache_manager.h"
#include "sql/session/ob_sql_session_info.h"
#include "sql/engine/expr/ob_expr_in.h"
#include "observer/ob_server.h"
#include "observer/ob_server_struct.h"
#include "sql/executor/ob_interm_result_pool.h"
#include "share/ob_tenant_mgr.h"

using namespace oceanbase::sql;
using namespace oceanbase::common;
using namespace test;
using namespace oceanbase::observer;
using namespace oceanbase::share;

class ObExecContextTest : public ::testing::Test {
public:
  ObExecContextTest();
  virtual ~ObExecContextTest();
  virtual void SetUp();
  virtual void TearDown();

private:
  // disallow copy
  ObExecContextTest(const ObExecContextTest& other);
  ObExecContextTest& operator=(const ObExecContextTest& other);

private:
  // data members
};
ObExecContextTest::ObExecContextTest()
{}

ObExecContextTest::~ObExecContextTest()
{}

void ObExecContextTest::SetUp()
{}

void ObExecContextTest::TearDown()
{}

class ObPhyOperatorCtxFake : public ObPhyOperator::ObPhyOperatorCtx {
public:
  ObPhyOperatorCtxFake(ObExecContext& ctx);
  ~ObPhyOperatorCtxFake();
  virtual void destroy()
  {
    return ObPhyOperatorCtx::destroy_base();
  }
  int phy_op_calc();
  void set_phy_op_ctx_a(int a);
  void set_phy_op_ctx_b(int b);

private:
  int phy_op_ctx_a_;
  int phy_op_ctx_b_;
};

ObPhyOperatorCtxFake::ObPhyOperatorCtxFake(ObExecContext& ctx)
    : ObPhyOperatorCtx(ctx), phy_op_ctx_a_(0), phy_op_ctx_b_(0)
{}

ObPhyOperatorCtxFake::~ObPhyOperatorCtxFake()
{}

void ObPhyOperatorCtxFake::set_phy_op_ctx_a(int a)
{
  phy_op_ctx_a_ = a;
}

void ObPhyOperatorCtxFake::set_phy_op_ctx_b(int b)
{
  phy_op_ctx_b_ = b;
}

int ObPhyOperatorCtxFake::phy_op_calc()
{
  return phy_op_ctx_a_ + phy_op_ctx_b_;
}

class ObRootTransmitInput : public ObIPhyOperatorInput {
public:
  ObRootTransmitInput(ObExecContext& ctx) : input_param_a_(0)
  {
    UNUSED(ctx);
  }
  virtual void reset() override
  {
    input_param_a_ = 0;
  }
  void set_input_param_a(int64_t a)
  {
    input_param_a_ = a;
  }

  int64_t get_input_param_a() const
  {
    return input_param_a_;
  }
  int serialize(char* buf, int64_t buf_len, int64_t& pos) const
  {
    UNUSED(buf);
    UNUSED(buf_len);
    UNUSED(pos);
    return OB_NOT_IMPLEMENT;
  }
  int deserialize(const char* buf, int64_t data_len, int64_t& pos)
  {
    UNUSED(buf);
    UNUSED(data_len);
    UNUSED(pos);
    return OB_NOT_IMPLEMENT;
  }
  int64_t get_serialize_size() const
  {
    return 0;
  }
  int init(ObExecContext& ctx, ObTaskInfo& task_info, ObPhyOperator& op)
  {
    UNUSED(ctx);
    UNUSED(task_info);
    UNUSED(op);
    return OB_NOT_IMPLEMENT;
  }
  ObPhyOperatorType get_phy_op_type() const
  {
    return PHY_ROOT_TRANSMIT;
  }

private:
  int64_t input_param_a_;
};

const uint64_t phy_op_id = 0;
// test create physical operator normal
TEST_F(ObExecContextTest, create_phy_op_test)
{
  int ret = OB_SUCCESS;
  void* ptr = NULL;
  ObExecContext exec_context;
  ObIAllocator* allocator = NULL;
  ObPhyOperatorCtxFake* phy_op_ctx = NULL;
  ObRootTransmitInput* input_param = NULL;
  int64_t phy_op_size = 64;

  ASSERT_EQ(OB_SUCCESS, create_test_session(exec_context));
  allocator = &exec_context.get_allocator();
  ASSERT_EQ(OB_SUCCESS, exec_context.init_phy_op(phy_op_size));
  for (int i = 0; i < phy_op_size; i++) {
    ret = CREATE_PHY_OPERATOR_CTX(
        ObPhyOperatorCtxFake, exec_context, phy_op_id + i, static_cast<ObPhyOperatorType>(0), phy_op_ctx);
    ASSERT_FALSE(NULL == phy_op_ctx);
    phy_op_ctx->set_phy_op_ctx_a(i);
    phy_op_ctx->set_phy_op_ctx_b(i + 1);
    ASSERT_EQ(i * 2 + 1, phy_op_ctx->phy_op_calc());

    // create input param
    ret = CREATE_PHY_OP_INPUT(ObRootTransmitInput, exec_context, phy_op_id + i, PHY_ROOT_TRANSMIT, input_param);
    ASSERT_FALSE(NULL == input_param);
    input_param->set_input_param_a(i);
    ASSERT_EQ(i, input_param->get_input_param_a());
  }

  for (int i = 0; i < phy_op_size; i++) {
    phy_op_ctx = GET_PHY_OPERATOR_CTX(ObPhyOperatorCtxFake, exec_context, phy_op_id + i);
    ASSERT_FALSE(NULL == phy_op_ctx);
    ASSERT_EQ(i * 2 + 1, phy_op_ctx->phy_op_calc());
    input_param = GET_PHY_OP_INPUT(ObRootTransmitInput, exec_context, phy_op_id + i);
    ASSERT_FALSE(NULL == input_param);
    ASSERT_EQ(i, input_param->get_input_param_a());
  }

  ptr = allocator->alloc(10);
  ASSERT_FALSE(NULL == ptr);
  allocator->free(ptr);
  UNUSED(ret);
}

// excepted create physical operator fail, create the same physical operator twice
TEST_F(ObExecContextTest, create_phy_op_fail)
{
  ObExecContext exec_context;
  ASSERT_EQ(OB_SUCCESS, create_test_session(exec_context));
  int64_t phy_op_size = 64;
  ASSERT_EQ(OB_SUCCESS, exec_context.init_phy_op(phy_op_size));

  ObPhyOperatorCtxFake* phy_op_ctx = NULL;
  ASSERT_EQ(OB_SUCCESS,
      CREATE_PHY_OPERATOR_CTX(
          ObPhyOperatorCtxFake, exec_context, phy_op_id, static_cast<ObPhyOperatorType>(0), phy_op_ctx));
  ASSERT_FALSE(NULL == phy_op_ctx);
  phy_op_ctx = GET_PHY_OPERATOR_CTX(ObPhyOperatorCtxFake, exec_context, phy_op_id);
  ASSERT_FALSE(NULL == phy_op_ctx);

  // create phy_op_ctx twice
  ASSERT_EQ(OB_INIT_TWICE,
      CREATE_PHY_OPERATOR_CTX(
          ObPhyOperatorCtxFake, exec_context, phy_op_id, static_cast<ObPhyOperatorType>(0), phy_op_ctx));
  ASSERT_TRUE(NULL == phy_op_ctx);

  // test create input param failed
  ObRootTransmitInput* input_param = NULL;
  ASSERT_EQ(
      OB_SUCCESS, CREATE_PHY_OP_INPUT(ObRootTransmitInput, exec_context, phy_op_id, PHY_ROOT_TRANSMIT, input_param));
  ASSERT_FALSE(NULL == input_param);
  input_param = GET_PHY_OP_INPUT(ObRootTransmitInput, exec_context, phy_op_id);
  ASSERT_FALSE(NULL == input_param);
  // create input_param twice
  ASSERT_EQ(
      OB_INIT_TWICE, CREATE_PHY_OP_INPUT(ObRootTransmitInput, exec_context, phy_op_id, PHY_ROOT_TRANSMIT, input_param));
  ASSERT_TRUE(NULL == input_param);
}

TEST_F(ObExecContextTest, create_physical_plan_ctx)
{
  ObExecContext exec_context;
  ASSERT_EQ(OB_SUCCESS, create_test_session(exec_context));
  ObPhysicalPlanCtx* plan_ctx = NULL;
  ASSERT_EQ(OB_SUCCESS, exec_context.init_phy_op(64));
  // create normal
  ASSERT_EQ(OB_SUCCESS, exec_context.create_physical_plan_ctx());
  ASSERT_FALSE(NULL == exec_context.get_physical_plan_ctx());
  // get plan ctx normal
  plan_ctx = exec_context.get_physical_plan_ctx();
  ASSERT_FALSE(NULL == plan_ctx);
}

TEST_F(ObExecContextTest, create_physical_plan_ctx_fail)
{
  ObExecContext exec_context;
  ASSERT_EQ(OB_SUCCESS, create_test_session(exec_context));
  ObPhysicalPlanCtx* plan_ctx = NULL;

  ASSERT_EQ(OB_SUCCESS, exec_context.init_phy_op(64));
  // get physical plan but not be created
  plan_ctx = exec_context.get_physical_plan_ctx();
  ASSERT_TRUE(NULL == plan_ctx);
  // create physical plan twice
  ASSERT_EQ(OB_SUCCESS, exec_context.create_physical_plan_ctx());
  ASSERT_FALSE(NULL == exec_context.get_physical_plan_ctx());
  ASSERT_EQ(OB_INIT_TWICE, exec_context.create_physical_plan_ctx());
}

TEST_F(ObExecContextTest, invalid_argument)
{
  ObExecContext exec_context;
  ASSERT_EQ(OB_SUCCESS, create_test_session(exec_context));
  int64_t phy_op_size = 64;
  uint64_t phy_op_id = 100;

  ASSERT_EQ(OB_SUCCESS, exec_context.init_phy_op(phy_op_size));
  // phy_op_id larger than phy_op_size
  ObPhyOperatorCtxFake* phy_op_ctx = NULL;
  ASSERT_EQ(OB_INVALID_ARGUMENT,
      CREATE_PHY_OPERATOR_CTX(
          ObPhyOperatorCtxFake, exec_context, phy_op_id, static_cast<ObPhyOperatorType>(0), phy_op_ctx));
  ASSERT_TRUE(NULL == phy_op_ctx);
  // OB_INVALID_ID phy_op_id
  ASSERT_EQ(OB_INVALID_ARGUMENT,
      CREATE_PHY_OPERATOR_CTX(
          ObPhyOperatorCtxFake, exec_context, OB_INVALID_ID, static_cast<ObPhyOperatorType>(0), phy_op_ctx));
  ASSERT_EQ(NULL, phy_op_ctx);
  phy_op_ctx = GET_PHY_OPERATOR_CTX(ObPhyOperatorCtxFake, exec_context, phy_op_id);
  ASSERT_TRUE(NULL == phy_op_ctx);
}

TEST_F(ObExecContextTest, serialization)
{
  int ret = OB_SUCCESS;
  ObExecContext exec_context;
  ASSERT_EQ(OB_SUCCESS, create_test_session(exec_context));
  ObPhyOperatorCtxFake* phy_op_ctx = NULL;
  ObRootTransmitInput* input_param = NULL;
  int64_t phy_op_size = 64;
  ObPlanCacheManager plan_cache_mgr;

  ASSERT_EQ(OB_SUCCESS, ObPreProcessSysVars::init_sys_var());
  ASSERT_EQ(OB_SUCCESS, exec_context.init_phy_op(phy_op_size));
  ASSERT_EQ(OB_SUCCESS, create_test_session(exec_context));
  auto my_session = exec_context.get_my_session();
  ASSERT_FALSE(NULL == my_session);
  ObPhysicalPlan cur_phy_plan;
  ASSERT_EQ(OB_SUCCESS, my_session->set_cur_phy_plan(&cur_phy_plan));
  ASSERT_EQ(OB_SUCCESS, exec_context.create_physical_plan_ctx());
  ASSERT_EQ(OB_SUCCESS, my_session->load_default_sys_variable(false, true));
  exec_context.set_plan_cache_manager(&plan_cache_mgr);

  for (int i = 0; i < phy_op_size; i++) {
    ret = CREATE_PHY_OPERATOR_CTX(
        ObPhyOperatorCtxFake, exec_context, phy_op_id + i, static_cast<ObPhyOperatorType>(0), phy_op_ctx);
    ASSERT_FALSE(NULL == phy_op_ctx);
    phy_op_ctx->set_phy_op_ctx_a(i);
    phy_op_ctx->set_phy_op_ctx_b(i + 1);
    ASSERT_EQ(i * 2 + 1, phy_op_ctx->phy_op_calc());

    // create input param
    ret = CREATE_PHY_OP_INPUT(ObRootTransmitInput, exec_context, phy_op_id + i, PHY_ROOT_TRANSMIT, input_param);
    ASSERT_FALSE(NULL == input_param);
    input_param->set_input_param_a(i);
    ASSERT_EQ(i, input_param->get_input_param_a());
  }

  for (int i = 0; i < phy_op_size; i++) {
    phy_op_ctx = GET_PHY_OPERATOR_CTX(ObPhyOperatorCtxFake, exec_context, phy_op_id + i);
    ASSERT_FALSE(NULL == phy_op_ctx);
    ASSERT_EQ(i * 2 + 1, phy_op_ctx->phy_op_calc());
    input_param = GET_PHY_OP_INPUT(ObRootTransmitInput, exec_context, phy_op_id + i);
    ASSERT_FALSE(NULL == input_param);
    ASSERT_EQ(i, input_param->get_input_param_a());
  }

  int64_t pos = 0;
  const int64_t MAX_SERIALIZE_BUF_LEN = 10240;
  char buf[MAX_SERIALIZE_BUF_LEN] = {'\0'};
  ASSERT_EQ(OB_SUCCESS, exec_context.serialize(buf, MAX_SERIALIZE_BUF_LEN, pos));
  ASSERT_EQ(pos, exec_context.get_serialize_size());
  UNUSED(ret);
}

int main(int argc, char** argv)
{
  system("rm -rf test_exec_context.log");
  ::testing::InitGoogleTest(&argc, argv);
  OB_LOGGER.set_log_level("INFO");
  OB_LOGGER.set_file_name("test_exec_context.log", true);
  int ret = RUN_ALL_TESTS();
  OB_LOGGER.disable();
  return ret;
}