oceanbase/unittest/sql/plan_cache/test_pc_perf.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.
 */

#define private public
#include "rpc/frame/ob_req_transport.h"
#include "observer/ob_server.h"
#include "sql/optimizer/test_optimizer_utils.h"
#include "test_sql.h"

using namespace oceanbase;
using namespace common;
using namespace sql;
using namespace share;
using namespace observer;
using namespace rpc::frame;
static test::TestSQL *test_sql = NULL;
ObSql sql_engine;
ObRsMgr rs_mgr;
ObAddr addr;

std::vector<string> test_sql_array;
bool g_stop = false;
int64_t g_end_time = 0;

struct Config {
  int thread_num;
  const char *schema_file;
  const char *query_file;
  Config();

  bool is_valid();
};

bool Config::is_valid() {
  return thread_num != 0 && schema_file != NULL && query_file != NULL;
}

Config::Config() {
  thread_num = 0;
  schema_file = NULL;
  query_file = NULL;
}

struct ThreadArgs {
  const Config *config;
  int64_t exec_count; //out param_;
  int64_t pc_total_time;
  int64_t parse_total_time;
  int64_t select_count;
  int64_t update_count;
  int64_t insert_count;
  int64_t delete_count;

  ThreadArgs();
};

ThreadArgs::ThreadArgs() {
  config = NULL;
  exec_count = 0;
  pc_total_time = 0;
  parse_total_time = 0;
  select_count = 0;
  update_count = 0;
  insert_count = 0;
  delete_count = 0;
}

class ObTestPlanCachePerformance : public test::TestOptimizerUtils
{
  public:
    int test(const char * query_str);
    int init_engine(const char *schema_file);
    virtual void SetUp();
    virtual void TearDown();
    void TestBody() {}
  public:
    int64_t test_times_; //执行test的次数
    int64_t pc_time_;     //plan cache总时间
    int64_t parse_time_;  //parse 总时间
    int64_t select_hit_count_;
    int64_t update_hit_count_;
    int64_t insert_hit_count_;
    int64_t delete_hit_count_;
};

void ObTestPlanCachePerformance::SetUp()
{
}

void ObTestPlanCachePerformance::TearDown()
{
}

TEST_F(ObTestPlanCachePerformance, basic_test)
{
}

void init_pc(const char *schema_file)
{
  if (NULL == test_sql) {
    test_sql = new test::TestSQL(ObString(schema_file));
    test_sql->init();
  }
}

int ObTestPlanCachePerformance::init_engine(const char *schema_file) {
  int ret = OB_SUCCESS;
  test_times_ = 0;
  pc_time_ = 0;
  parse_time_ = 0;
  select_hit_count_ = 0;
  update_hit_count_ = 0;
  insert_hit_count_ = 0;
  delete_hit_count_ = 0;
  init_pc(schema_file);
  init();
  if (OB_FAIL(sql_engine.init(&stat_manager_,
                              (ObReqTransport*)1,
                              &partition_service_,
                              NULL, // no virtual data access
                              &part_cache_,
                              addr,
                              rs_mgr))) {
    SQL_PC_LOG(ERROR, "fail to init sql engine");
  }
  return ret;
}

int load_sql(const char *test_file, std::vector<string> &sql_array)
{
  int ret = OB_SUCCESS;
  std::ifstream if_tests(test_file);
  if (!if_tests.is_open()) {
    SQL_PC_LOG(ERROR, "maybe reach max file open");
    ret = OB_ERROR;
  }
  std::string line;
  std::string total_line;
  ;
  while (std::getline(if_tests, line)) {
    // allow human readable formatting
    if (line.size() <= 0) continue;
    std::size_t begin = line.find_first_not_of('\t');
    if (line.at(begin) == '#') continue;
    std::size_t end = line.find_last_not_of('\t');
    std::string exact_line = line.substr(begin, end - begin + 1);
    total_line += exact_line;
    total_line += " ";
    if (exact_line.at(exact_line.length() - 1) != ';') continue;
    else {
      sql_array.push_back(total_line);
      total_line = "";
    }
  }
  if_tests.close();
  return ret;
}

void *run(void * arg) {
  ThreadArgs *thread_args = (ThreadArgs *)(arg);

  ObTestPlanCachePerformance p;
  p.init();

  while (!g_stop) {
    for (int i = 0; i < (int)test_sql_array.size(); i ++) {
      p.test(test_sql_array[i].c_str());
    }
  }
  thread_args->exec_count = p.test_times_;
  thread_args->pc_total_time = p.pc_time_;
  thread_args->parse_total_time = p.parse_time_;
  thread_args->select_count = p.select_hit_count_;
  thread_args->update_count = p.update_hit_count_;
  thread_args->insert_count = p.insert_hit_count_;
  thread_args->delete_count = p.delete_hit_count_;
  return NULL;
}

void run_all(const Config &config) {
  load_sql(config.query_file, test_sql_array);
  pthread_t t_array[config.thread_num];
  ThreadArgs args[config.thread_num];
  for (int i = 0; i < config.thread_num; i ++) {
    args[i].config = &config;
    pthread_create(&t_array[i], NULL, run, &(args[i]));
  }

  int64_t start_time = ::oceanbase::common::ObTimeUtility::current_time();

  void *ret = NULL;
  for (int i = 0; i < config.thread_num; i ++) {
    pthread_join(t_array[i], &ret);
  }

  int64_t timeu = g_end_time - start_time;

  int64_t total = 0;
  int64_t pc_total_time = 0;
  int64_t parse_total_time = 0;
  int64_t select_hit_total = 0;
  int64_t update_hit_total = 0;
  int64_t insert_hit_total= 0;
  int64_t delete_hit_total = 0;
  for (int i = 0; i < config.thread_num; i ++) {
    total += args[i].exec_count;
    pc_total_time += args[i].pc_total_time;
    parse_total_time += args[i].parse_total_time;
    select_hit_total += args[i].select_count;
    update_hit_total += args[i].update_count;
    insert_hit_total += args[i].insert_count;
    delete_hit_total += args[i].delete_count;
  }
  _SQL_PC_LOG(ERROR, "===");
  _SQL_PC_LOG(ERROR, "===");
  _SQL_PC_LOG(ERROR, "===");
  _SQL_PC_LOG(ERROR, "total_cnt = %ld, agv_timeu = %ld, QPS = %f", total, timeu, ((double)total / (double)timeu) * 1000000.0);
  _SQL_PC_LOG(ERROR, "avg_parse_time = %ld, avg_pc_time = %ld", parse_total_time/total, pc_total_time/total);
  _SQL_PC_LOG(ERROR, "hit_total = %ld, select_hit = %ld, update_hit = %ld, insert_hit = %ld, delete_hit = %ld",
                     select_hit_total + update_hit_total + insert_hit_total + delete_hit_total,
                     select_hit_total, update_hit_total, insert_hit_total, delete_hit_total);
  _SQL_PC_LOG(ERROR, "===");
  _SQL_PC_LOG(ERROR, "===");
  _SQL_PC_LOG(ERROR, "===");
}

int ObTestPlanCachePerformance::test(const char * query_str) {
  int ret = OB_SUCCESS;
  ObSqlCtx context;
  ObArenaAllocator allocator(ObModIds::TEST);
  ObResultSet result(session_info_, allocator);
  int64_t t1 = 0, t2 =0, t3 = 0;
  test_times_ ++;

  t1 = ::oceanbase::common::ObTimeUtility::current_time();
  if (OB_SUCC(ret)) {
    context.schema_manager_ = test_sql->get_schema_manager();
    context.session_info_ = &session_info_;
    context.partition_location_cache_ = &part_cache_;
    session_info_.set_plan_cache(sql_engine.get_plan_cache(OB_SYS_TENANT_ID));
  }

  t2 = ::oceanbase::common::ObTimeUtility::current_time();
  if (OB_SUCC(ret)) {
    if OB_FAIL(sql_engine.handle_text_query(query_str, context, result)) {
      SQL_PC_LOG(WARN, "fail to handle text query", K(ret));
    }
  }
  t3 = ::oceanbase::common::ObTimeUtility::current_time();
  parse_time_ += t2-t1;
  pc_time_ += t3-t2;

  return ret;
}

void print_usage() {
  fprintf(stderr, "Usage: %s -t thread_num -s schema_file -q query_file [-l log_level]\n", "test_performance");
  exit(0);
}

void my_signal(int sig) {
  if (sig == SIGINT) {
    g_stop = true;
    g_end_time = ::oceanbase::common::ObTimeUtility::current_time();
  }
}

int main(int argc, char **argv)
{
  Config config;
  const char *log_level = "INFO";
  if (SIG_ERR == signal(SIGINT, my_signal)) {
    exit(-1);
  }
  int c = 0;
  while(-1 != (c = getopt(argc, argv, "t:s:q:l:"))) {
    switch (c) {
    case 't':
      config.thread_num = atoi(optarg);
      break;
    case 's':
      config.schema_file = optarg;
      break;
    case 'q':
      config.query_file = optarg;
      break;
    case 'l':
      log_level = optarg;
      break;
    default:
      print_usage();
    }
  }
  if (!config.is_valid()) {
    print_usage();
  }
  OB_LOGGER.set_log_level(log_level);
  ::oceanbase::sql::init_sql_factories();
  ObTestPlanCachePerformance p;
  p.init_engine(config.schema_file);
  run_all(config);
  return 0;
}