Remove stages (#208)

### What problem were solved in this pull request?
ref #138 

Problem:

SEDA的线程池模式是每个处理环节分开，可能会放到不同的线程上，也可以调整不同stage对应的处理线程个数。但是会增加复杂性，比如session会切换线程、代码逻辑复杂。

### What is changed and how it works?
去掉了parse、resolve等stage，仅保留了session stage。代码中的名字依然叫stage，后续再调整。
在session
stage的处理逻辑中，直接调用其它流程处理，这样也使SQL处理流程变得更清晰。后续还可以考虑移除observer.ini，所有参数均通过命令行传递。

### Other information

CMakeLists.txt

@@ -53,6 +53,8 @@ IF (CONCURRENCY)
     SET(CMAKE_COMMON_FLAGS "${CMAKE_COMMON_FLAGS} -DCONCURRENCY")
     ADD_DEFINITIONS(-DCONCURRENCY)
 ENDIF (CONCURRENCY)
+# test coverage
+SET(CMAKE_COMMON_FLAGS "${CMAKE_COMMON_FLAGS} -fprofile-arcs -ftest-coverage")
 SET(CMAKE_CXX_FLAGS ${CMAKE_COMMON_FLAGS})
 SET(CMAKE_C_FLAGS ${CMAKE_COMMON_FLAGS})
 MESSAGE("CMAKE_CXX_FLAGS is " ${CMAKE_CXX_FLAGS})

etc/observer.ini

@@ -26,9 +26,7 @@ MaxEventHistoryNum=100
 # threadpools' name, it will contain the threadpool's section
 ThreadPools=SQLThreads,IOThreads,DefaultThreads
 # stage list
-STAGES=SessionStage,ExecuteStage,OptimizeStage,ParseStage,ResolveStage,\
-PlanCacheStage,QueryCacheStage,DefaultStorageStage,MemStorageStage,\
-TimerStage,MetricsStage
+STAGES=SessionStage
 
 [NET]
 CLIENT_ADDRESS=INADDR_ANY
@@ -57,39 +55,3 @@ count=3
 
 [SessionStage]
 ThreadId=SQLThreads
-NextStages=QueryCacheStage
-
-[QueryCacheStage]
-ThreadId=SQLThreads
-NextStages=ParseStage
-
-[ParseStage]
-ThreadId=SQLThreads
-NextStages=ResolveStage
-
-[ResolveStage]
-ThreadId=SQLThreads
-NextStages=PlanCacheStage
-
-[PlanCacheStage]
-ThreadId=SQLThreads
-NextStages=OptimizeStage
-
-[OptimizeStage]
-ThreadId=SQLThreads
-NextStages=ExecuteStage
-
-[ExecuteStage]
-ThreadId=SQLThreads
-NextStages=DefaultStorageStage,MemStorageStage
-
-[DefaultStorageStage]
-ThreadId=IOThreads
-BaseDir=./miniob
-SystemDb=sys
-
-[MemStorageStage]
-ThreadId=IOThreads
-
-[MetricsStage]
-NextStages=TimerStage

src/observer/common/init.cpp

@@ -24,9 +24,6 @@ See the Mulan PSL v2 for more details. */
 #include "common/os/signal.h"
 #include "common/seda/init.h"
 #include "common/seda/stage_factory.h"
-
-#include "common/metrics/log_reporter.h"
-#include "common/metrics/metrics_registry.h"
 #include "session/session.h"
 #include "session/session_stage.h"
 #include "sql/executor/execute_stage.h"
@@ -35,8 +32,6 @@ See the Mulan PSL v2 for more details. */
 #include "sql/parser/resolve_stage.h"
 #include "sql/plan_cache/plan_cache_stage.h"
 #include "sql/query_cache/query_cache_stage.h"
-#include "storage/default/default_storage_stage.h"
-#include "storage/mem/mem_storage_stage.h"
 #include "storage/buffer/disk_buffer_pool.h"
 #include "storage/default/default_handler.h"
 #include "storage/trx/trx.h"
@@ -153,14 +148,6 @@ void cleanup_log()
 int prepare_init_seda()
 {
   static StageFactory session_stage_factory("SessionStage", &SessionStage::make_stage);
-  static StageFactory resolve_stage_factory("ResolveStage", &ResolveStage::make_stage);
-  static StageFactory query_cache_stage_factory("QueryCacheStage", &QueryCacheStage::make_stage);
-  static StageFactory parse_stage_factory("ParseStage", &ParseStage::make_stage);
-  static StageFactory plan_cache_factory("PlanCacheStage", &PlanCacheStage::make_stage);
-  static StageFactory optimize_factory("OptimizeStage", &OptimizeStage::make_stage);
-  static StageFactory execute_factory("ExecuteStage", &ExecuteStage::make_stage);
-  static StageFactory default_storage_factory("DefaultStorageStage", &DefaultStorageStage::make_stage);
-  static StageFactory mem_storage_factory("MemStorageStage", &MemStorageStage::make_stage);
   return 0;
 }
 
@@ -170,6 +157,7 @@ int init_global_objects(ProcessParam *process_param, Ini &properties)
   BufferPoolManager::set_instance(GCTX.buffer_pool_manager_);
 
   GCTX.handler_ = new DefaultHandler();
+  
   DefaultHandler::set_default(GCTX.handler_);
 
   int ret = 0;
@@ -179,6 +167,12 @@ int init_global_objects(ProcessParam *process_param, Ini &properties)
     ret = -1;
   }
   GCTX.trx_kit_ = TrxKit::instance();
+
+  rc = GCTX.handler_->init("miniob");
+  if (OB_FAIL(rc)) {
+    LOG_ERROR("failed to init handler. rc=%s", strrc(rc));
+    return -1;
+  }
   return ret;
 }
 
@@ -257,11 +251,6 @@ int init(ProcessParam *process_param)
     return rc;
   }
 
-  LogReporter *log_reporter = get_log_reporter();
-  MetricsRegistry &metrics_registry = get_metrics_registry();
-
-  metrics_registry.add_reporter(log_reporter);
-
   // Block interrupt signals before creating child threads.
   // setSignalHandler(sig_handler);
   // sigset_t newSigset, oset;

src/observer/main.cpp

@@ -23,6 +23,7 @@ See the Mulan PSL v2 for more details. */
 #include "common/ini_setting.h"
 #include "common/os/process.h"
 #include "common/os/signal.h"
+#include "common/lang/string.h"
 #include "net/server.h"
 #include "net/server_param.h"
 
@@ -196,4 +197,4 @@ int main(int argc, char **argv)
  * MiniOB 是 OceanBase 与华中科技大学联合开发的、面向"零"基础同学的数据库入门学习项目。
  *
  * MiniOB 设计的目标是面向在校学生、数据库从业者、爱好者，或者对基础技术有兴趣的爱好者, 整体代码量少，易于上手并学习, 是一个系统性的数据库学习项目。miniob 设置了一系列由浅入深的题目，以帮助同学们"零"基础入门, 让同学们快速了解数据库并深入学习数据库内核，期望通过相关训练之后，能够熟练掌握数据库内核模块的功能与协同关系, 并能够在使用数据库时，设计出高效的 SQL 。miniob 为了更好的学习数据库实现原理, 对诸多模块都做了简化，比如不考虑并发操作, 安全特性, 复杂的事物管理等功能。
- */
\ No newline at end of file
+ */

src/observer/net/server.cpp

@@ -36,7 +36,6 @@ See the Mulan PSL v2 for more details. */
 #include "session/session.h"
 #include "common/ini_setting.h"
 #include "net/communicator.h"
-#include <common/metrics/metrics_registry.h>
 
 using namespace common;
 

src/observer/net/server.h

@@ -15,7 +15,6 @@ See the Mulan PSL v2 for more details. */
 #pragma once
 
 #include "common/defs.h"
-#include "common/metrics/metrics.h"
 #include "common/seda/stage.h"
 #include "net/server_param.h"
 

src/observer/session/session_stage.cpp

@@ -19,10 +19,8 @@ See the Mulan PSL v2 for more details. */
 
 #include "common/conf/ini.h"
 #include "common/log/log.h"
-#include "common/seda/timer_stage.h"
-
 #include "common/lang/mutex.h"
-#include "common/metrics/metrics_registry.h"
+#include "common/lang/string.h"
 #include "common/seda/callback.h"
 #include "event/session_event.h"
 #include "event/sql_event.h"
@@ -32,10 +30,8 @@ See the Mulan PSL v2 for more details. */
 
 using namespace common;
 
-const std::string SessionStage::SQL_METRIC_TAG = "SessionStage.sql";
-
 // Constructor
-SessionStage::SessionStage(const char *tag) : Stage(tag), query_cache_stage_(nullptr), sql_metric_(nullptr)
+SessionStage::SessionStage(const char *tag) : Stage(tag)
 {}
 
 // Destructor
@@ -71,24 +67,13 @@ bool SessionStage::set_properties()
 // Initialize stage params and validate outputs
 bool SessionStage::initialize()
 {
-  std::list<Stage *>::iterator stgp = next_stage_list_.begin();
-  query_cache_stage_ = *(stgp++);
-
-  MetricsRegistry &metricsRegistry = get_metrics_registry();
-  sql_metric_ = new SimpleTimer();
-  metricsRegistry.register_metric(SQL_METRIC_TAG, sql_metric_);
   return true;
 }
 
 // Cleanup after disconnection
 void SessionStage::cleanup()
 {
-  MetricsRegistry &metricsRegistry = get_metrics_registry();
-  if (sql_metric_ != nullptr) {
-    metricsRegistry.unregister(SQL_METRIC_TAG);
-    delete sql_metric_;
-    sql_metric_ = nullptr;
-  }
+
 }
 
 void SessionStage::handle_event(StageEvent *event)
@@ -108,8 +93,6 @@ void SessionStage::handle_request(StageEvent *event)
     return;
   }
 
-  TimerStat sql_stat(*sql_metric_);
-
   std::string sql = sev->query();
   if (common::is_blank(sql.c_str())) {
     return;
@@ -118,7 +101,7 @@ void SessionStage::handle_request(StageEvent *event)
   Session::set_current_session(sev->session());
   sev->session()->set_current_request(sev);
   SQLStageEvent *sql_event = new SQLStageEvent(sev, sql);
-  query_cache_stage_->handle_event(sql_event);
+  (void)handle_sql(sql_event);
 
   Communicator *communicator = sev->get_communicator();
   bool need_disconnect = false;
@@ -130,3 +113,48 @@ void SessionStage::handle_request(StageEvent *event)
   sev->session()->set_current_request(nullptr);
   Session::set_current_session(nullptr);
 }
+
+/**
+ * 处理一个SQL语句经历这几个阶段。
+ * 虽然看起来流程比较多，但是对于大多数SQL来说，更多的可以关注parse和executor阶段。
+ * 通常只有select、delete等带有查询条件的语句才需要进入optimize。
+ * 对于DDL语句，比如create table、create index等，没有对应的查询计划，可以直接搜索
+ * create_table_executor、create_index_executor来看具体的执行代码。
+ * select、delete等DML语句，会产生一些执行计划，如果感觉繁琐，可以跳过optimize直接看
+ * execute_stage中的执行，通过explain语句看需要哪些operator，然后找对应的operator来
+ * 调试或者看代码执行过程即可。
+ */
+RC SessionStage::handle_sql(SQLStageEvent *sql_event)
+{
+  RC rc = query_cache_stage_.handle_request(sql_event);
+  if (OB_FAIL(rc)) {
+    LOG_TRACE("failed to do query cache. rc=%s", strrc(rc));
+    return rc;
+  }
+
+  rc = parse_stage_.handle_request(sql_event);
+  if (OB_FAIL(rc)) {
+    LOG_TRACE("failed to do parse. rc=%s", strrc(rc));
+    return rc;
+  }
+
+  rc = resolve_stage_.handle_request(sql_event);
+  if (OB_FAIL(rc)) {
+    LOG_TRACE("failed to do resolve. rc=%s", strrc(rc));
+    return rc;
+  }
+  
+  rc = optimize_stage_.handle_request(sql_event);
+  if (rc != RC::UNIMPLENMENT && rc != RC::SUCCESS) {
+    LOG_TRACE("failed to do optimize. rc=%s", strrc(rc));
+    return rc;
+  }
+  
+  rc = execute_stage_.handle_request(sql_event);
+  if (OB_FAIL(rc)) {
+    LOG_TRACE("failed to do execute. rc=%s", strrc(rc));
+    return rc;
+  }
+
+  return rc;
+}

src/observer/session/session_stage.h

@@ -15,7 +15,11 @@ See the Mulan PSL v2 for more details. */
 #pragma once
 
 #include "common/seda/stage.h"
-#include "common/metrics/metrics.h"
+#include "sql/query_cache/query_cache_stage.h"
+#include "sql/parser/parse_stage.h"
+#include "sql/parser/resolve_stage.h"
+#include "sql/optimizer/optimize_stage.h"
+#include "sql/executor/execute_stage.h"
 
 /**
  * @brief SEDA处理的stage
@@ -50,9 +54,12 @@ protected:
 
 protected:
   void handle_request(common::StageEvent *event);
+  RC   handle_sql(SQLStageEvent *sql_event);
 
 private:
-  Stage *query_cache_stage_ = nullptr;
-  common::SimpleTimer *sql_metric_ = nullptr;
-  static const std::string SQL_METRIC_TAG;
+  QueryCacheStage query_cache_stage_;
+  ParseStage      parse_stage_;
+  ResolveStage    resolve_stage_;
+  OptimizeStage   optimize_stage_;
+  ExecuteStage    execute_stage_;
 };

src/observer/sql/executor/command_executor.cpp

@@ -23,6 +23,7 @@ See the Mulan PSL v2 for more details. */
 #include "sql/executor/trx_begin_executor.h"
 #include "sql/executor/trx_end_executor.h"
 #include "sql/executor/set_variable_executor.h"
+#include "sql/executor/load_data_executor.h"
 #include "common/log/log.h"
 
 RC CommandExecutor::execute(SQLStageEvent *sql_event)
@@ -71,6 +72,11 @@ RC CommandExecutor::execute(SQLStageEvent *sql_event)
       return executor.execute(sql_event);
     }
 
+    case StmtType::LOAD_DATA: {
+      LoadDataExecutor executor;
+      return executor.execute(sql_event);
+    }
+
     case StmtType::EXIT: {
       return RC::SUCCESS;
     }
@@ -82,4 +88,4 @@ RC CommandExecutor::execute(SQLStageEvent *sql_event)
   }
 
   return RC::INTERNAL;
-}
\ No newline at end of file
+}

src/observer/sql/executor/execute_stage.cpp

@@ -30,66 +30,9 @@ See the Mulan PSL v2 for more details. */
 using namespace std;
 using namespace common;
 
-//! Constructor
-ExecuteStage::ExecuteStage(const char *tag) : Stage(tag)
-{}
-
-//! Destructor
-ExecuteStage::~ExecuteStage()
-{}
-
-//! Parse properties, instantiate a stage object
-Stage *ExecuteStage::make_stage(const std::string &tag)
+RC ExecuteStage::handle_request(SQLStageEvent *sql_event)
 {
-  ExecuteStage *stage = new (std::nothrow) ExecuteStage(tag.c_str());
-  if (stage == nullptr) {
-    LOG_ERROR("new ExecuteStage failed");
-    return nullptr;
-  }
-  stage->set_properties();
-  return stage;
-}
-
-//! Set properties for this object set in stage specific properties
-bool ExecuteStage::set_properties()
-{
-  //  std::string stageNameStr(stageName);
-  //  std::map<std::string, std::string> section = theGlobalProperties()->get(
-  //    stageNameStr);
-  //
-  //  std::map<std::string, std::string>::iterator it;
-  //
-  //  std::string key;
-
-  return true;
-}
-
-//! Initialize stage params and validate outputs
-bool ExecuteStage::initialize()
-{
-  std::list<Stage *>::iterator stgp = next_stage_list_.begin();
-  default_storage_stage_ = *(stgp++);
-  mem_storage_stage_ = *(stgp++);
-
-  return true;
-}
-
-//! Cleanup after disconnection
-void ExecuteStage::cleanup()
-{
-}
-
-void ExecuteStage::handle_event(StageEvent *event)
-{
-  handle_request(event);
-
-  return;
-}
-
-RC ExecuteStage::handle_request(common::StageEvent *event)
-{
-  SQLStageEvent *sql_event = static_cast<SQLStageEvent *>(event);
-
+  RC rc = RC::SUCCESS;
   const unique_ptr<PhysicalOperator> &physical_operator = sql_event->physical_operator();
   if (physical_operator != nullptr) {
     return handle_request_with_physical_operator(sql_event);
@@ -100,12 +43,12 @@ RC ExecuteStage::handle_request(common::StageEvent *event)
   Stmt *stmt = sql_event->stmt();
   if (stmt != nullptr) {
     CommandExecutor command_executor;
-    RC rc = command_executor.execute(sql_event);
+    rc = command_executor.execute(sql_event);
     session_event->sql_result()->set_return_code(rc);
   } else {
-    default_storage_stage_->handle_event(event);
+    return RC::INTERNAL;
   }
-  return RC::SUCCESS;
+  return rc;
 }
 
 RC ExecuteStage::handle_request_with_physical_operator(SQLStageEvent *sql_event)
@@ -146,4 +89,3 @@ RC ExecuteStage::handle_request_with_physical_operator(SQLStageEvent *sql_event)
   sql_result->set_operator(std::move(physical_operator));
   return rc;
 }
-

src/observer/sql/executor/execute_stage.h

@@ -14,7 +14,6 @@ See the Mulan PSL v2 for more details. */
 
 #pragma once
 
-#include "common/seda/stage.h"
 #include "sql/parser/parse.h"
 #include "common/rc.h"
 
@@ -28,25 +27,10 @@ class SelectStmt;
  * @details 根据前面阶段生成的结果，有些语句会生成执行计划，有些不会。
  * 整体上分为两类，带执行计划的，或者 CommandExecutor 可以直接执行的。
  */
-class ExecuteStage : public common::Stage 
+class ExecuteStage
 {
 public:
-  virtual ~ExecuteStage();
-  static Stage *make_stage(const std::string &tag);
 
-protected:
-  // common function
-  ExecuteStage(const char *tag);
-  bool set_properties() override;
-
-  bool initialize() override;
-  void cleanup() override;
-  void handle_event(common::StageEvent *event) override;
-
-  RC handle_request(common::StageEvent *event);
+  RC handle_request(SQLStageEvent *event);
   RC handle_request_with_physical_operator(SQLStageEvent *sql_event);
-
-private:
-  Stage *default_storage_stage_ = nullptr;
-  Stage *mem_storage_stage_ = nullptr;
 };

src/observer/storage/default/default_storage_stage.cpp → src/observer/sql/executor/load_data_executor.cpp

@@ -9,174 +9,27 @@ MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 See the Mulan PSL v2 for more details. */
 
 //
-// Created by Longda on 2021/4/13.
+// Created by Wangyunlai on 2023/7/12.
 //
 
-#include <string.h>
-#include <string>
-
-#include "storage/default/default_storage_stage.h"
-
-#include "common/conf/ini.h"
-#include "common/io/io.h"
-#include "common/lang/string.h"
-#include "common/log/log.h"
-#include "common/seda/timer_stage.h"
-#include "common/metrics/metrics_registry.h"
-#include "common/rc.h"
-#include "storage/default/default_handler.h"
-#include "storage/common/condition_filter.h"
-#include "storage/table/table.h"
-#include "storage/table/table_meta.h"
-#include "storage/trx/trx.h"
-#include "event/session_event.h"
+#include "sql/executor/load_data_executor.h"
 #include "event/sql_event.h"
-#include "event/storage_event.h"
-#include "session/session.h"
+#include "event/session_event.h"
+#include "sql/executor/sql_result.h"
+#include "common/lang/string.h"
+#include "sql/stmt/load_data_stmt.h"
 
 using namespace common;
 
-const std::string DefaultStorageStage::QUERY_METRIC_TAG = "DefaultStorageStage.query";
-const char *CONF_BASE_DIR = "BaseDir";
-const char *CONF_SYSTEM_DB = "SystemDb";
-
-const char *DEFAULT_SYSTEM_DB = "sys";
-
-//! Constructor
-DefaultStorageStage::DefaultStorageStage(const char *tag) : Stage(tag), handler_(nullptr)
-{}
-
-//! Destructor
-DefaultStorageStage::~DefaultStorageStage()
-{
-  delete handler_;
-  handler_ = nullptr;
-}
-
-//! Parse properties, instantiate a stage object
-Stage *DefaultStorageStage::make_stage(const std::string &tag)
-{
-  DefaultStorageStage *stage = new (std::nothrow) DefaultStorageStage(tag.c_str());
-  if (stage == nullptr) {
-    LOG_ERROR("new DefaultStorageStage failed");
-    return nullptr;
-  }
-  stage->set_properties();
-  return stage;
-}
-
-//! Set properties for this object set in stage specific properties
-bool DefaultStorageStage::set_properties()
-{
-  std::string stageNameStr(stage_name_);
-  std::map<std::string, std::string> section = get_properties()->get(stageNameStr);
-
-  // 初始化时打开默认的database，没有的话会自动创建
-  std::map<std::string, std::string>::iterator iter = section.find(CONF_BASE_DIR);
-  if (iter == section.end()) {
-    LOG_ERROR("Config cannot be empty: %s", CONF_BASE_DIR);
-    return false;
-  }
-
-  const char *base_dir = iter->second.c_str();
-
-  const char *sys_db = DEFAULT_SYSTEM_DB;
-  iter = section.find(CONF_SYSTEM_DB);
-  if (iter != section.end()) {
-    sys_db = iter->second.c_str();
-    LOG_INFO("Use %s as system db", sys_db);
-  }
-
-  handler_ = &DefaultHandler::get_default();
-  if (RC::SUCCESS != handler_->init(base_dir)) {
-    LOG_ERROR("Failed to init default handler");
-    return false;
-  }
-
-  RC ret = handler_->create_db(sys_db);
-  if (ret != RC::SUCCESS && ret != RC::SCHEMA_DB_EXIST) {
-    LOG_ERROR("Failed to create system db");
-    return false;
-  }
-
-  ret = handler_->open_db(sys_db);
-  if (ret != RC::SUCCESS) {
-    LOG_ERROR("Failed to open system db. rc=%s", strrc(ret));
-    return false;
-  }
-
-  Session &default_session = Session::default_session();
-  default_session.set_current_db(sys_db);
-
-  LOG_INFO("Open system db success: %s", sys_db);
-  return true;
-}
-
-//! Initialize stage params and validate outputs
-bool DefaultStorageStage::initialize()
-{
-  MetricsRegistry &metricsRegistry = get_metrics_registry();
-  query_metric_ = new SimpleTimer();
-  metricsRegistry.register_metric(QUERY_METRIC_TAG, query_metric_);
-
-  return true;
-}
-
-//! Cleanup after disconnection
-void DefaultStorageStage::cleanup()
+RC LoadDataExecutor::execute(SQLStageEvent *sql_event)
 {
-  if (handler_) {
-    handler_->destroy();
-    handler_ = nullptr;
-  }
-}
-
-void DefaultStorageStage::handle_event(StageEvent *event)
-{
-  TimerStat timerStat(*query_metric_);
-
-  SQLStageEvent *sql_event = static_cast<SQLStageEvent *>(event);
-  Command *cmd = sql_event->command().get();
-  SessionEvent *session_event = sql_event->session_event();
-  Session *session = session_event->session();
-  SqlResult *sql_result = session_event->sql_result();
-  Db *db = session->get_current_db();
-  const char *dbname = db->name();
-  Trx *current_trx = session->current_trx();
-
   RC rc = RC::SUCCESS;
-
-  switch (cmd->flag) {
-    case SCF_LOAD_DATA: {
-      /*
-        从文件导入数据，如果做性能测试，需要保持这些代码可以正常工作
-        load data infile `your/file/path` into table `table-name`;
-       */
-      const char *table_name = cmd->load_data.relation_name.c_str();
-      const char *file_name = cmd->load_data.file_name.c_str();
-      load_data(dbname, table_name, file_name, sql_result);
-    } break;
-    default:
-      char response[256];
-      snprintf(response, sizeof(response), "Unsupported sql: %d\n", cmd->flag);
-      sql_result->set_return_code(RC::UNIMPLENMENT);
-      sql_result->set_state_string(response);
-      break;
-  }
-
-  if (rc == RC::SUCCESS && !session->is_trx_multi_operation_mode()) {
-    rc = current_trx->commit();
-    if (rc != RC::SUCCESS) {
-      LOG_ERROR("Failed to commit trx. rc=%d:%s", rc, strrc(rc));
-    }
-  }
-}
-
-void DefaultStorageStage::callback_event(StageEvent *event, CallbackContext *context)
-{
-  StorageEvent *storage_event = static_cast<StorageEvent *>(event);
-  storage_event->sql_event()->done_immediate();
-  return;
+  SqlResult *sql_result = sql_event->session_event()->sql_result();
+  LoadDataStmt *stmt = static_cast<LoadDataStmt *>(sql_event->stmt());
+  Table *table = stmt->table();
+  const char *file_name = stmt->filename();
+  load_data(table, file_name, sql_result);
+  return rc;
 }
 
 /**
@@ -263,19 +116,9 @@ RC insert_record_from_file(Table *table,
   return rc;
 }
 
-void DefaultStorageStage::load_data(const char *db_name, 
-                                    const char *table_name, 
-                                    const char *file_name, 
-                                    SqlResult *sql_result)
+void LoadDataExecutor::load_data(Table *table, const char *file_name, SqlResult *sql_result)
 {
   std::stringstream result_string;
-  Table *table = handler_->find_table(db_name, table_name);
-  if (nullptr == table) {
-    result_string << "No such table " << db_name << "." << table_name << std::endl;
-    sql_result->set_return_code(RC::SCHEMA_TABLE_NOT_EXIST);
-    sql_result->set_state_string(result_string.str());
-    return;
-  }
 
   std::fstream fs;
   fs.open(file_name, std::ios_base::in | std::ios_base::binary);

src/observer/storage/default/default_storage_stage.h → src/observer/sql/executor/load_data_executor.h

-/* Copyright (c) 2021 Xie Meiyi(xiemeiyi@hust.edu.cn) and OceanBase and/or its affiliates. All rights reserved.
+/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
 miniob is licensed under Mulan PSL v2.
 You can use this software according to the terms and conditions of the Mulan PSL v2.
 You may obtain a copy of Mulan PSL v2 at:
@@ -9,44 +9,29 @@ MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 See the Mulan PSL v2 for more details. */
 
 //
-// Created by Meiyi & Longda on 2021/4/13.
+// Created by Wangyunlai on 2023/7/12.
 //
 
 #pragma once
 
-#include "common/seda/stage.h"
-#include "common/metrics/metrics.h"
+#include "common/rc.h"
 
-class DefaultHandler;
+class SQLStageEvent;
+class Table;
 class SqlResult;
 
 /**
- * @brief 当前仅实现了load data，是准备删除的一个SQL stage
- * @ingroup SQLStage
+ * @brief 导入数据的执行器
+ * @ingroup Executor
  */
-class DefaultStorageStage : public common::Stage 
+class LoadDataExecutor
 {
 public:
-  ~DefaultStorageStage();
-  static Stage *make_stage(const std::string &tag);
-
-protected:
-  // common function
-  DefaultStorageStage(const char *tag);
-  bool set_properties() override;
-
-  bool initialize() override;
-  void cleanup() override;
-  void handle_event(common::StageEvent *event) override;
-  void callback_event(common::StageEvent *event, common::CallbackContext *context) override;
-
-private:
-  void load_data(const char *db_name, const char *table_name, const char *file_name, SqlResult *sql_result);
-
-protected:
-  common::SimpleTimer *query_metric_ = nullptr;
-  static const std::string QUERY_METRIC_TAG;
+  LoadDataExecutor() = default;
+  virtual ~LoadDataExecutor() = default;
 
+  RC execute(SQLStageEvent *sql_event);
+  
 private:
-  DefaultHandler *handler_;
+  void load_data(Table *table, const char *file_name, SqlResult *sql_result);
 };

src/observer/sql/optimizer/optimize_stage.cpp

@@ -21,7 +21,6 @@ See the Mulan PSL v2 for more details. */
 #include "common/io/io.h"
 #include "common/lang/string.h"
 #include "common/log/log.h"
-#include "common/seda/timer_stage.h"
 #include "sql/expr/expression.h"
 #include "sql/operator/logical_operator.h"
 #include "sql/operator/project_logical_operator.h"
@@ -45,66 +44,6 @@ See the Mulan PSL v2 for more details. */
 using namespace std;
 using namespace common;
 
-//! Constructor
-OptimizeStage::OptimizeStage(const char *tag) : Stage(tag)
-{}
-
-//! Destructor
-OptimizeStage::~OptimizeStage()
-{}
-
-//! Parse properties, instantiate a stage object
-Stage *OptimizeStage::make_stage(const std::string &tag)
-{
-  OptimizeStage *stage = new (std::nothrow) OptimizeStage(tag.c_str());
-  if (stage == nullptr) {
-    LOG_ERROR("new OptimizeStage failed");
-    return nullptr;
-  }
-  stage->set_properties();
-  return stage;
-}
-
-//! Set properties for this object set in stage specific properties
-bool OptimizeStage::set_properties()
-{
-  //  std::string stageNameStr(stage_name_);
-  //  std::map<std::string, std::string> section = g_properties()->get(
-  //    stageNameStr);
-  //
-  //  std::map<std::string, std::string>::iterator it;
-  //
-  //  std::string key;
-
-  return true;
-}
-
-//! Initialize stage params and validate outputs
-bool OptimizeStage::initialize()
-{
-  std::list<Stage *>::iterator stgp = next_stage_list_.begin();
-  execute_stage_ = *(stgp++);
-
-  return true;
-}
-
-//! Cleanup after disconnection
-void OptimizeStage::cleanup()
-{
-}
-
-void OptimizeStage::handle_event(StageEvent *event)
-{
-  SQLStageEvent *sql_event = static_cast<SQLStageEvent *>(event);
-  SqlResult *sql_result = sql_event->session_event()->sql_result();
-
-  RC rc = handle_request(sql_event);
-  if (rc != RC::UNIMPLENMENT && rc != RC::SUCCESS) {
-    sql_result->set_return_code(rc);
-  } else {
-    execute_stage_->handle_event(event);
-  }
-}
 RC OptimizeStage::handle_request(SQLStageEvent *sql_event)
 {
   unique_ptr<LogicalOperator> logical_operator;

src/observer/sql/optimizer/optimize_stage.h

@@ -17,7 +17,6 @@ See the Mulan PSL v2 for more details. */
 #include <memory>
 
 #include "common/rc.h"
-#include "common/seda/stage.h"
 #include "sql/operator/logical_operator.h"
 #include "sql/operator/physical_operator.h"
 #include "sql/optimizer/physical_plan_generator.h"
@@ -40,24 +39,12 @@ class ExplainStmt;
  * 不过并不是所有的语句都需要生成计划，有些可以直接执行，比如create table、create index等。
  * 这些语句可以参考 @class CommandExecutor。
  */
-class OptimizeStage : public common::Stage
+class OptimizeStage
 {
 public:
-  ~OptimizeStage();
-  static Stage *make_stage(const std::string &tag);
-
-protected:
-  // common function
-  OptimizeStage(const char *tag);
-  bool set_properties();
-
-  bool initialize();
-  void cleanup();
-  void handle_event(common::StageEvent *event);
-
-private:
   RC handle_request(SQLStageEvent *event);
 
+private:
   /**
    * @brief 根据SQL生成逻辑计划
    * @details 由于SQL语句种类比较多，并且SQL语句可能会有嵌套的情况，比如带有SQL子查询的语句，那就需要递归的创建逻辑计划。
@@ -98,7 +85,6 @@ private:
       std::unique_ptr<LogicalOperator> &logical_operator, std::unique_ptr<PhysicalOperator> &physical_operator);
 
 private:
-  Stage *execute_stage_ = nullptr;    /// 生成计划
   PhysicalPlanGenerator physical_plan_generator_; /// 根据逻辑计划生成物理计划
   Rewriter rewriter_;  /// 逻辑计划改写
 };

src/observer/sql/parser/parse_stage.cpp

@@ -21,80 +21,16 @@ See the Mulan PSL v2 for more details. */
 #include "common/io/io.h"
 #include "common/lang/string.h"
 #include "common/log/log.h"
-#include "common/seda/timer_stage.h"
 #include "event/session_event.h"
 #include "event/sql_event.h"
 #include "sql/parser/parse.h"
 
 using namespace common;
 
-//! Constructor
-ParseStage::ParseStage(const char *tag) : Stage(tag)
-{}
-
-//! Destructor
-ParseStage::~ParseStage()
-{}
-
-//! Parse properties, instantiate a stage object
-Stage *ParseStage::make_stage(const std::string &tag)
-{
-  ParseStage *stage = new (std::nothrow) ParseStage(tag.c_str());
-  if (stage == nullptr) {
-    LOG_ERROR("new ParseStage failed");
-    return nullptr;
-  }
-  stage->set_properties();
-  return stage;
-}
-
-//! Set properties for this object set in stage specific properties
-bool ParseStage::set_properties()
-{
-  //  std::string stageNameStr(stageName);
-  //  std::map<std::string, std::string> section = theGlobalProperties()->get(
-  //    stageNameStr);
-  //
-  //  std::map<std::string, std::string>::iterator it;
-  //
-  //  std::string key;
-
-  return true;
-}
-
-//! Initialize stage params and validate outputs
-bool ParseStage::initialize()
-{
-  std::list<Stage *>::iterator stgp = next_stage_list_.begin();
-  // optimize_stage_ = *(stgp++);
-  resolve_stage_ = *(stgp++);
-
-  return true;
-}
-
-//! Cleanup after disconnection
-void ParseStage::cleanup()
-{
-  LOG_TRACE("Enter");
-
-  LOG_TRACE("Exit");
-}
-
-void ParseStage::handle_event(StageEvent *event)
+RC ParseStage::handle_request(SQLStageEvent *sql_event)
 {
-  RC rc = handle_request(event);
-  if (RC::SUCCESS != rc) {
-    LOG_ERROR("Failed to handle request");
-  }
-
-  SQLStageEvent *sql_event = static_cast<SQLStageEvent *>(event);
-  sql_event->done_immediate();
-  return;
-}
-
-RC ParseStage::handle_request(StageEvent *event)
-{
-  SQLStageEvent *sql_event = static_cast<SQLStageEvent *>(event);
+  RC rc = RC::SUCCESS;
+  
   SqlResult *sql_result = sql_event->session_event()->sql_result();
   const std::string &sql = sql_event->sql();
 
@@ -114,13 +50,13 @@ RC ParseStage::handle_request(StageEvent *event)
   std::unique_ptr<Command> cmd = std::move(parsed_sql_result.commands().front());
   if (cmd->flag == SCF_ERROR) {
     // set error information to event
-    sql_result->set_return_code(RC::SQL_SYNTAX);
+    rc = RC::SQL_SYNTAX;
+    sql_result->set_return_code(rc);
     sql_result->set_state_string("Failed to parse sql");
-    return RC::INTERNAL;
+    return rc;
   }
 
   sql_event->set_command(std::move(cmd));
 
-  resolve_stage_->handle_event(event);
   return RC::SUCCESS;
 }

src/observer/sql/parser/parse_stage.h

@@ -14,31 +14,16 @@ See the Mulan PSL v2 for more details. */
 
 #pragma once
 
-#include "common/seda/stage.h"
 #include "common/rc.h"
 
+class SQLStageEvent;
+
 /**
  * @brief 解析SQL语句，解析后的结果可以参考parse_defs.h
  * @ingroup SQLStage
  */
-class ParseStage : public common::Stage 
+class ParseStage
 {
 public:
-  virtual ~ParseStage();
-  static Stage *make_stage(const std::string &tag);
-
-protected:
-  // common function
-  ParseStage(const char *tag);
-  bool set_properties() override;
-
-  bool initialize() override;
-  void cleanup() override;
-  void handle_event(common::StageEvent *event) override;
-
-protected:
-  RC handle_request(common::StageEvent *event);
-
-private:
-  Stage *resolve_stage_ = nullptr;  /// 解析后执行的下一个阶段，就是Resolve
+  RC handle_request(SQLStageEvent *sql_event);
 };

src/observer/sql/parser/resolve_stage.cpp

@@ -21,7 +21,6 @@ See the Mulan PSL v2 for more details. */
 #include "common/io/io.h"
 #include "common/lang/string.h"
 #include "common/log/log.h"
-#include "common/seda/timer_stage.h"
 #include "event/sql_event.h"
 #include "event/session_event.h"
 #include "session/session.h"
@@ -29,83 +28,31 @@ See the Mulan PSL v2 for more details. */
 
 using namespace common;
 
-//! Constructor
-ResolveStage::ResolveStage(const char *tag) : Stage(tag)
-{}
-
-//! Destructor
-ResolveStage::~ResolveStage()
-{}
-
-//! Parse properties, instantiate a stage object
-Stage *ResolveStage::make_stage(const std::string &tag)
-{
-  ResolveStage *stage = new (std::nothrow) ResolveStage(tag.c_str());
-  if (stage == nullptr) {
-    LOG_ERROR("new ResolveStage failed");
-    return nullptr;
-  }
-  stage->set_properties();
-  return stage;
-}
-
-//! Set properties for this object set in stage specific properties
-bool ResolveStage::set_properties()
+RC ResolveStage::handle_request(SQLStageEvent *sql_event)
 {
-  //  std::string stageNameStr(stage_name_);
-  //  std::map<std::string, std::string> section = g_properties()->get(
-  //    stageNameStr);
-  //
-  //  std::map<std::string, std::string>::iterator it;
-  //
-  //  std::string key;
-
-  return true;
-}
-
-//! Initialize stage params and validate outputs
-bool ResolveStage::initialize()
-{
-  std::list<Stage *>::iterator stgp = next_stage_list_.begin();
-  plan_cache_stage_ = *(stgp++);
-
-  return true;
-}
-
-//! Cleanup after disconnection
-void ResolveStage::cleanup()
-{
-}
-
-void ResolveStage::handle_event(StageEvent *event)
-{
-  SQLStageEvent *sql_event = static_cast<SQLStageEvent *>(event);
-  if (nullptr == sql_event) {
-    LOG_WARN("failed to get sql stage event");
-    return;
-  }
-
+  RC rc = RC::SUCCESS;
   SessionEvent *session_event = sql_event->session_event();
   SqlResult *sql_result = session_event->sql_result();
 
   Db *db = session_event->session()->get_current_db();
   if (nullptr == db) {
-    LOG_ERROR("cannot current db");
-    return;
+    LOG_ERROR("cannot find current db");
+    rc = RC::SCHEMA_DB_NOT_EXIST;
+    sql_result->set_return_code(rc);
+    sql_result->set_state_string("no db selected");
+    return rc;
   }
 
   Command *cmd = sql_event->command().get();
   Stmt *stmt = nullptr;
-  RC rc = Stmt::create_stmt(db, *cmd, stmt);
+  rc = Stmt::create_stmt(db, *cmd, stmt);
   if (rc != RC::SUCCESS && rc != RC::UNIMPLENMENT) {
     LOG_WARN("failed to create stmt. rc=%d:%s", rc, strrc(rc));
     sql_result->set_return_code(rc);
-    return;
+    return rc;
   }
 
   sql_event->set_stmt(stmt);
 
-  plan_cache_stage_->handle_event(sql_event);
-
-  return;
+  return rc;
 }

src/observer/sql/parser/resolve_stage.h

@@ -14,28 +14,16 @@ See the Mulan PSL v2 for more details. */
 
 #pragma once
 
-#include "common/seda/stage.h"
+#include "common/rc.h"
+
+class SQLStageEvent;
 
 /**
  * @brief 执行Resolve，将解析后的SQL语句，转换成各种Stmt(Statement), 同时会做错误检查
  * @ingroup SQLStage
  */
-class ResolveStage : public common::Stage 
+class ResolveStage
 {
 public:
-  ~ResolveStage();
-  static Stage *make_stage(const std::string &tag);
-
-protected:
-  // common function
-  ResolveStage(const char *tag);
-  bool set_properties();
-
-  bool initialize();
-  void cleanup();
-  void handle_event(common::StageEvent *event);
-
-protected:
-private:
-  Stage *plan_cache_stage_ = nullptr;
+  RC handle_request(SQLStageEvent *sql_event);
 };

src/observer/sql/parser/yacc_sql.cpp

@@ -592,7 +592,7 @@ static const yytype_int16 yyrline[] =
      310,   311,   314,   330,   333,   344,   347,   350,   358,   370,
      385,   408,   415,   427,   432,   443,   446,   460,   463,   476,
      479,   485,   488,   493,   500,   512,   524,   536,   551,   552,
-     553,   554,   555,   556,   560,   570,   578,   588,   589
+     553,   554,   555,   556,   560,   573,   581,   591,   592
 };
 #endif
 
@@ -2205,25 +2205,28 @@ yyreduce:
   case 74: /* load_data: LOAD DATA INFILE SSS INTO TABLE ID  */
 #line 561 "yacc_sql.y"
     {
+      char *tmp_file_name = common::substr((yyvsp[-3].string), 1, strlen((yyvsp[-3].string)) - 2);
+      
       (yyval.command) = new Command(SCF_LOAD_DATA);
       (yyval.command)->load_data.relation_name = (yyvsp[0].string);
-      (yyval.command)->load_data.file_name = (yyvsp[-3].string);
+      (yyval.command)->load_data.file_name = tmp_file_name;
       free((yyvsp[0].string));
+      free(tmp_file_name);
     }
-#line 2214 "yacc_sql.cpp"
+#line 2217 "yacc_sql.cpp"
     break;
 
   case 75: /* explain: EXPLAIN command_wrapper  */
-#line 571 "yacc_sql.y"
+#line 574 "yacc_sql.y"
     {
       (yyval.command) = new Command(SCF_EXPLAIN);
       (yyval.command)->explain.cmd = std::unique_ptr<Command>((yyvsp[0].command));
     }
-#line 2223 "yacc_sql.cpp"
+#line 2226 "yacc_sql.cpp"
     break;
 
   case 76: /* set_variable: SET ID EQ value  */
-#line 579 "yacc_sql.y"
+#line 582 "yacc_sql.y"
     {
       (yyval.command) = new Command(SCF_SET_VARIABLE);
       (yyval.command)->set_variable.name = (yyvsp[-2].string);
@@ -2231,11 +2234,11 @@ yyreduce:
       free((yyvsp[-2].string));
       delete (yyvsp[0].value);
     }
-#line 2235 "yacc_sql.cpp"
+#line 2238 "yacc_sql.cpp"
     break;
 
 
-#line 2239 "yacc_sql.cpp"
+#line 2242 "yacc_sql.cpp"
 
       default: break;
     }
@@ -2465,7 +2468,7 @@ yyreturn:
   return yyresult;
 }
 
-#line 591 "yacc_sql.y"
+#line 594 "yacc_sql.y"
 
 //_____________________________________________________________________
 extern void scan_string(const char *str, yyscan_t scanner);

src/observer/sql/parser/yacc_sql.y

@@ -559,10 +559,13 @@ comp_op:
 load_data:
     LOAD DATA INFILE SSS INTO TABLE ID 
     {
+      char *tmp_file_name = common::substr($4, 1, strlen($4) - 2);
+      
       $$ = new Command(SCF_LOAD_DATA);
       $$->load_data.relation_name = $7;
-      $$->load_data.file_name = $4;
+      $$->load_data.file_name = tmp_file_name;
       free($7);
+      free(tmp_file_name);
     }
     ;
 

src/observer/sql/plan_cache/plan_cache_stage.cpp

@@ -21,74 +21,5 @@ See the Mulan PSL v2 for more details. */
 #include "common/io/io.h"
 #include "common/lang/string.h"
 #include "common/log/log.h"
-#include "common/seda/timer_stage.h"
 
 using namespace common;
-
-//! Constructor
-PlanCacheStage::PlanCacheStage(const char *tag) : Stage(tag)
-{}
-
-//! Destructor
-PlanCacheStage::~PlanCacheStage()
-{}
-
-//! Parse properties, instantiate a stage object
-Stage *PlanCacheStage::make_stage(const std::string &tag)
-{
-  PlanCacheStage *stage = new (std::nothrow) PlanCacheStage(tag.c_str());
-  if (stage == nullptr) {
-    LOG_ERROR("new PlanCacheStage failed");
-    return nullptr;
-  }
-  stage->set_properties();
-  return stage;
-}
-
-//! Set properties for this object set in stage specific properties
-bool PlanCacheStage::set_properties()
-{
-  //  std::string stageNameStr(stage_name_);
-  //  std::map<std::string, std::string> section = g_properties()->get(
-  //    stageNameStr);
-  //
-  //  std::map<std::string, std::string>::iterator it;
-  //
-  //  std::string key;
-
-  return true;
-}
-
-//! Initialize stage params and validate outputs
-bool PlanCacheStage::initialize()
-{
-  std::list<Stage *>::iterator stgp = next_stage_list_.begin();
-  //  execute_stage = *(stgp++);
-  optimizer_stage_ = *(stgp++);
-
-  return true;
-}
-
-//! Cleanup after disconnection
-void PlanCacheStage::cleanup()
-{
-}
-
-void PlanCacheStage::handle_event(StageEvent *event)
-{
-  // Add callback to update plan cache
-  /*
-  CompletionCallback *cb = new (std::nothrow) CompletionCallback(this, nullptr);
-  if (cb == nullptr) {
-    LOG_ERROR("Failed to new callback for SQLStageEvent");
-    event->done_immediate();
-    return;
-  }
-
-  event->push_callback(cb);
-   */
-  // do nothing here, pass the event to the next stage
-  optimizer_stage_->handle_event(event);
-
-  return;
-}

src/observer/sql/plan_cache/plan_cache_stage.h

@@ -14,7 +14,7 @@ See the Mulan PSL v2 for more details. */
 
 #pragma once
 
-#include "common/seda/stage.h"
+#include "common/rc.h"
 
 /**
  * @brief 尝试从Plan的缓存中获取Plan，如果没有命中，则执行Optimizer
@@ -22,22 +22,6 @@ See the Mulan PSL v2 for more details. */
  * @details 实际上现在什么都没做。不过PlanCache对数据库的优化提升明显，是一个非常有趣的功能，
  * 感兴趣的同学可以参考OceanBase的实现
  */
-class PlanCacheStage : public common::Stage 
+class PlanCacheStage
 {
-public:
-  virtual ~PlanCacheStage();
-  static Stage *make_stage(const std::string &tag);
-
-protected:
-  // common function
-  PlanCacheStage(const char *tag);
-  bool set_properties() override;
-
-  bool initialize() override;
-  void cleanup() override;
-  void handle_event(common::StageEvent *event) override;
-
-protected:
-private:
-  Stage *optimizer_stage_ = nullptr;
 };

src/observer/sql/query_cache/query_cache_stage.cpp

@@ -21,61 +21,10 @@ See the Mulan PSL v2 for more details. */
 #include "common/io/io.h"
 #include "common/lang/string.h"
 #include "common/log/log.h"
-#include "common/seda/timer_stage.h"
 
 using namespace common;
 
-//! Constructor
-QueryCacheStage::QueryCacheStage(const char *tag) : Stage(tag)
-{}
-
-//! Destructor
-QueryCacheStage::~QueryCacheStage()
-{}
-
-//! Parse properties, instantiate a stage object
-Stage *QueryCacheStage::make_stage(const std::string &tag)
-{
-  QueryCacheStage *stage = new (std::nothrow) QueryCacheStage(tag.c_str());
-  if (stage == nullptr) {
-    LOG_ERROR("new QueryCacheStage failed");
-    return nullptr;
-  }
-  stage->set_properties();
-  return stage;
-}
-
-//! Set properties for this object set in stage specific properties
-bool QueryCacheStage::set_properties()
-{
-  //  std::string stageNameStr(stage_name_);
-  //  std::map<std::string, std::string> section = g_properties()->get(
-  //    stageNameStr);
-  //
-  //  std::map<std::string, std::string>::iterator it;
-  //
-  //  std::string key;
-
-  return true;
-}
-
-//! Initialize stage params and validate outputs
-bool QueryCacheStage::initialize()
+RC QueryCacheStage::handle_request(SQLStageEvent *sql_event)
 {
-  std::list<Stage *>::iterator stgp = next_stage_list_.begin();
-  parser_stage_ = *(stgp++);
-
-  return true;
-}
-
-//! Cleanup after disconnection
-void QueryCacheStage::cleanup()
-{
-}
-
-void QueryCacheStage::handle_event(StageEvent *event)
-{
-  parser_stage_->handle_event(event);
-
-  return;
+  return RC::SUCCESS;
 }

src/observer/sql/query_cache/query_cache_stage.h

@@ -14,29 +14,21 @@ See the Mulan PSL v2 for more details. */
 
 #pragma once
 
-#include "common/seda/stage.h"
+#include "common/rc.h"
+
+class SQLStageEvent;
 
 /**
  * @brief 查询缓存处理
  * @ingroup SQLStage
  * @details 当前什么都没做
  */
-class QueryCacheStage : public common::Stage
+class QueryCacheStage
 {
 public:
-  virtual ~QueryCacheStage();
-  static Stage *make_stage(const std::string &tag);
-
-protected:
-  // common function
-  QueryCacheStage(const char *tag);
-  bool set_properties() override;
+  QueryCacheStage() = default;
+  virtual ~QueryCacheStage() = default;
 
-  bool initialize() override;
-  void cleanup() override;
-  void handle_event(common::StageEvent *event) override;
-
-protected:
-private:
-  Stage *parser_stage_ = nullptr;
+public:
+  RC handle_request(SQLStageEvent *sql_event);
 };

src/observer/storage/mem/mem_storage_stage.cpp → src/observer/sql/stmt/load_data_stmt.cpp

@@ -9,78 +9,40 @@ MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 See the Mulan PSL v2 for more details. */
 
 //
-// Created by Longda on 2021/4/13.
+// Created by Wangyunlai on 2023/7/12.
 //
 
-#include <string.h>
-#include <string>
-
-#include "mem_storage_stage.h"
-
-#include "common/conf/ini.h"
-#include "common/io/io.h"
-#include "common/lang/string.h"
+#include <unistd.h>
+#include "sql/stmt/load_data_stmt.h"
 #include "common/log/log.h"
-#include "common/seda/timer_stage.h"
-#include "common/metrics/metrics_registry.h"
+#include "common/lang/string.h"
+#include "storage/db/db.h"
+#include "storage/table/table.h"
 
 using namespace common;
 
-//! Constructor
-MemStorageStage::MemStorageStage(const char *tag) : Stage(tag)
-{}
-
-//! Destructor
-MemStorageStage::~MemStorageStage()
-{}
-
-//! Parse properties, instantiate a stage object
-Stage *MemStorageStage::make_stage(const std::string &tag)
+RC LoadDataStmt::create(Db *db, const LoadData &load_data, Stmt *&stmt)
 {
-  MemStorageStage *stage = new (std::nothrow) MemStorageStage(tag.c_str());
-  if (stage == nullptr) {
-    LOG_ERROR("new MemStorageStage failed");
-    return nullptr;
+  RC rc = RC::SUCCESS;
+  const char *table_name = load_data.relation_name.c_str();
+  if (is_blank(table_name) || is_blank(load_data.file_name.c_str())) {
+    LOG_WARN("invalid argument. db=%p, table_name=%p, file name=%s",
+        db, table_name, load_data.file_name.c_str());
+    return RC::INVALID_ARGUMENT;
   }
-  stage->set_properties();
-  return stage;
-}
-
-//! Set properties for this object set in stage specific properties
-bool MemStorageStage::set_properties()
-{
-  //  std::string stageNameStr(stage_name_);
-  //  std::map<std::string, std::string> section = g_properties()->get(
-  //    stageNameStr);
-  //
-  //  std::map<std::string, std::string>::iterator it;
-  //
-  //  std::string key;
 
-  return true;
-}
-
-//! Initialize stage params and validate outputs
-bool MemStorageStage::initialize()
-{
-  return true;
-}
-
-//! Cleanup after disconnection
-void MemStorageStage::cleanup()
-{
-}
-
-void MemStorageStage::handle_event(StageEvent *event)
-{
-  TimerStat timerStat(*queryMetric);
-
-  event->done_immediate();
+  // check whether the table exists
+  Table *table = db->find_table(table_name);
+  if (nullptr == table) {
+    LOG_WARN("no such table. db=%s, table_name=%s", db->name(), table_name);
+    return RC::SCHEMA_TABLE_NOT_EXIST;
+  }
 
-  return;
-}
+  if (0 != access(load_data.file_name.c_str(), R_OK)) {
+    LOG_WARN("no such file to load. file name=%s, error=%s", load_data.file_name.c_str(), strerror(errno));
+    return RC::FILE_NOT_EXIST;
+  }
 
-void MemStorageStage::callback_event(StageEvent *event, CallbackContext *context)
-{
-  return;
+  stmt = new LoadDataStmt(table, load_data.file_name.c_str());
+  return rc;
 }

src/observer/storage/mem/mem_storage_stage.h → src/observer/sql/stmt/load_data_stmt.h

@@ -9,32 +9,33 @@ MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 See the Mulan PSL v2 for more details. */
 
 //
-// Created by Longda on 2021/4/13.
+// Created by Wangyunlai on 2023/7/12.
 //
 
 #pragma once
 
-#include "common/seda/stage.h"
-#include "common/metrics/metrics.h"
+#include <string>
+#include <vector>
 
-class MemStorageStage : public common::Stage {
+#include "sql/stmt/stmt.h"
+
+class Table;
+
+class LoadDataStmt : public Stmt
+{
 public:
-  ~MemStorageStage();
-  static Stage *make_stage(const std::string &tag);
+  LoadDataStmt(Table *table, const char *filename) : table_(table), filename_(filename)
+  {}
+  virtual ~LoadDataStmt() = default;
 
-protected:
-  // common function
-  MemStorageStage(const char *tag);
-  bool set_properties();
+  StmtType type() const override { return StmtType::LOAD_DATA; }
 
-  bool initialize();
-  void cleanup();
-  void handle_event(common::StageEvent *event);
-  void callback_event(common::StageEvent *event, common::CallbackContext *context);
+  Table *table() const { return table_; }
+  const char *filename() const { return filename_.c_str(); }
 
-protected:
-  common::SimpleTimer *queryMetric = nullptr;
-  static const std::string QUERY_METRIC_TAG;
+  static RC create(Db *db, const LoadData &load_data, Stmt *&stmt);
 
 private:
+  Table *table_ = nullptr;
+  std::string filename_;
 };

src/observer/sql/stmt/stmt.cpp

@@ -27,6 +27,7 @@ See the Mulan PSL v2 for more details. */
 #include "sql/stmt/trx_end_stmt.h"
 #include "sql/stmt/exit_stmt.h"
 #include "sql/stmt/set_variable_stmt.h"
+#include "sql/stmt/load_data_stmt.h"
 
 RC Stmt::create_stmt(Db *db, const Command &cmd, Stmt *&stmt)
 {
@@ -84,6 +85,10 @@ RC Stmt::create_stmt(Db *db, const Command &cmd, Stmt *&stmt)
       return SetVariableStmt::create(cmd.set_variable, stmt);
     }
 
+    case SCF_LOAD_DATA: {
+      return LoadDataStmt::create(db, cmd.load_data, stmt);
+    }
+
     default: {
       LOG_INFO("Command::type %d doesn't need to create statement.", cmd.flag);
     } break;

src/observer/storage/default/default_handler.cpp

@@ -24,6 +24,7 @@ See the Mulan PSL v2 for more details. */
 #include "storage/table/table.h"
 #include "storage/common/condition_filter.h"
 #include "storage/clog/clog.h"
+#include "session/session.h"
 
 static DefaultHandler *default_handler = nullptr;
 
@@ -62,6 +63,23 @@ RC DefaultHandler::init(const char *base_dir)
   base_dir_ = base_dir;
   db_dir_ = tmp + "/";
 
+  const char *sys_db = "sys";
+
+  RC ret = create_db(sys_db);
+  if (ret != RC::SUCCESS && ret != RC::SCHEMA_DB_EXIST) {
+    LOG_ERROR("Failed to create system db");
+    return ret;
+  }
+
+  ret = open_db(sys_db);
+  if (ret != RC::SUCCESS) {
+    LOG_ERROR("Failed to open system db. rc=%s", strrc(ret));
+    return ret;
+  }
+
+  Session &default_session = Session::default_session();
+  default_session.set_current_db(sys_db);
+
   LOG_INFO("Default handler init with %s success", base_dir);
   return RC::SUCCESS;
 }