Fix direct load exec session info

src/observer/table_load/ob_table_load_begin_processor.cpp

@@ -222,8 +222,11 @@ int ObTableLoadBeginP::create_table_ctx(const ObTableLoadParam &param,
       LOG_WARN("fail to alloc table ctx", KR(ret), K(param));
     } else if (OB_FAIL(table_ctx->init(param, ddl_param, &session_info))) {
       LOG_WARN("fail to init table ctx", KR(ret));
+    } else if (OB_FAIL(table_ctx->init_client_exec_ctx())) {
+      LOG_WARN("fail to init client exec ctx", KR(ret));
     } else if (OB_FAIL(ObTableLoadCoordinator::init_ctx(table_ctx, idx_array_,
-                                                        session_info.get_priv_user_id()))) {
+                                                        session_info.get_priv_user_id(),
+                                                        table_ctx->client_exec_ctx_))) {
       LOG_WARN("fail to coordinator init ctx", KR(ret));
     } else if (OB_FAIL(ObTableLoadService::add_ctx(table_ctx))) {
       LOG_WARN("fail to add ctx", KR(ret));

src/observer/table_load/ob_table_load_commit_processor.cpp

@@ -28,7 +28,6 @@ int ObTableLoadCommitP::process()
     LOG_WARN("fail to check_user_access", KR(ret));
   } else {
     ObTableLoadTableCtx *table_ctx = nullptr;
-    ObExecContext *exec_ctx = nullptr;
     ObTableLoadUniqueKey key(arg_.table_id_, arg_.task_id_);
     if (OB_FAIL(ObTableLoadService::get_ctx(key, table_ctx))) {
       LOG_WARN("fail to get table ctx", KR(ret), K(key));
@@ -36,7 +35,7 @@ int ObTableLoadCommitP::process()
       ObTableLoadCoordinator coordinator(table_ctx);
       if (OB_FAIL(coordinator.init())) {
         LOG_WARN("fail to init coordinator", KR(ret));
-      } else if (OB_FAIL(coordinator.commit(exec_ctx, result_.result_info_))) {
+      } else if (OB_FAIL(coordinator.commit(result_.result_info_))) {
         LOG_WARN("fail to coordinator commit", KR(ret));
       } else if (OB_FAIL(ObTableLoadService::remove_ctx(table_ctx))) {
         LOG_WARN("fail to remove table ctx", KR(ret), K(key));

src/observer/table_load/ob_table_load_coordinator.cpp

@@ -64,13 +64,13 @@ bool ObTableLoadCoordinator::is_ctx_inited(ObTableLoadTableCtx *ctx)
 }
 
 int ObTableLoadCoordinator::init_ctx(ObTableLoadTableCtx *ctx, const ObIArray<int64_t> &idx_array,
-                                     uint64_t user_id)
+                                     uint64_t user_id, ObTableLoadExecCtx *exec_ctx)
 {
   int ret = OB_SUCCESS;
   if (OB_ISNULL(ctx)) {
     ret = OB_INVALID_ARGUMENT;
     LOG_WARN("invalid agrs", KR(ret));
-  } else if (OB_FAIL(ctx->init_coordinator_ctx(idx_array, user_id))) {
+  } else if (OB_FAIL(ctx->init_coordinator_ctx(idx_array, user_id, exec_ctx))) {
     LOG_WARN("fail to init coordinator ctx", KR(ret));
   }
   return ret;
@@ -167,7 +167,8 @@ int ObTableLoadCoordinator::abort_redef_table(ObTableLoadTableCtx *ctx)
   ObTableLoadRedefTableAbortArg arg;
   arg.tenant_id_ = ctx->param_.tenant_id_;
   arg.task_id_ = ctx->ddl_param_.task_id_;
-  if (OB_FAIL(ObTableLoadRedefTable::abort(arg, *ctx->session_info_))) {
+  if (OB_FAIL(
+        ObTableLoadRedefTable::abort(arg, *ctx->coordinator_ctx_->exec_ctx_->get_session_info()))) {
     LOG_WARN("fail to abort redef table", KR(ret), K(arg));
   }
   return ret;
@@ -640,7 +641,8 @@ int ObTableLoadCoordinator::commit_redef_table()
   arg.dest_table_id_ = ctx_->ddl_param_.dest_table_id_;
   arg.task_id_ = ctx_->ddl_param_.task_id_;
   arg.schema_version_ = ctx_->ddl_param_.schema_version_;
-  if (OB_FAIL(ObTableLoadRedefTable::finish(arg, *ctx_->session_info_))) {
+  if (OB_FAIL(
+        ObTableLoadRedefTable::finish(arg, *coordinator_ctx_->exec_ctx_->get_session_info()))) {
     LOG_WARN("fail to finish redef table", KR(ret), K(arg));
   }
   return ret;
@@ -648,8 +650,7 @@ int ObTableLoadCoordinator::commit_redef_table()
 
 // commit() = px_commit_data() + px_commit_ddl()
 // used in non px_mode
-int ObTableLoadCoordinator::commit(ObExecContext *exec_ctx,
+int ObTableLoadCoordinator::commit(ObTableLoadResultInfo &result_info)
-                                   ObTableLoadResultInfo &result_info)
 {
   int ret = OB_SUCCESS;
   if (IS_NOT_INIT) {
@@ -664,7 +665,8 @@ int ObTableLoadCoordinator::commit(ObExecContext *exec_ctx,
     } else if (OB_FAIL(commit_peers(sql_statistics))) {
       LOG_WARN("fail to commit peers", KR(ret));
     } else if (param_.online_opt_stat_gather_ &&
-               OB_FAIL(drive_sql_stat(*exec_ctx, sql_statistics))) {
+               OB_FAIL(
+                 drive_sql_stat(coordinator_ctx_->exec_ctx_->get_exec_ctx(), sql_statistics))) {
       LOG_WARN("fail to drive sql stat", KR(ret));
     } else if (OB_FAIL(commit_redef_table())) {
       LOG_WARN("fail to commit redef table", KR(ret));
@@ -679,7 +681,7 @@ int ObTableLoadCoordinator::commit(ObExecContext *exec_ctx,
 
 // used in insert /*+ append */ into select clause
 // commit data loaded
-int ObTableLoadCoordinator::px_commit_data(ObExecContext *exec_ctx)
+int ObTableLoadCoordinator::px_commit_data()
 {
   int ret = OB_SUCCESS;
   if (IS_NOT_INIT) {
@@ -694,7 +696,8 @@ int ObTableLoadCoordinator::px_commit_data(ObExecContext *exec_ctx)
     } else if (OB_FAIL(commit_peers(sql_statistics))) {
       LOG_WARN("fail to commit peers", KR(ret));
     } else if (param_.online_opt_stat_gather_ &&
-               OB_FAIL(drive_sql_stat(*exec_ctx, sql_statistics))) {
+               OB_FAIL(
+                 drive_sql_stat(coordinator_ctx_->exec_ctx_->get_exec_ctx(), sql_statistics))) {
       LOG_WARN("fail to drive sql stat", KR(ret));
     }
   }
@@ -722,7 +725,9 @@ int ObTableLoadCoordinator::px_commit_ddl()
   return ret;
 }
 
-int ObTableLoadCoordinator::drive_sql_stat(ObExecContext &ctx, ObTableLoadSqlStatistics &sql_statistics) {
+int ObTableLoadCoordinator::drive_sql_stat(ObExecContext *ctx,
+                                           ObTableLoadSqlStatistics &sql_statistics)
+{
   int ret = OB_SUCCESS;
   const uint64_t tenant_id = MTL_ID();
   const uint64_t table_id = ctx_->ddl_param_.dest_table_id_;
@@ -730,28 +735,28 @@ int ObTableLoadCoordinator::drive_sql_stat(ObExecContext &ctx, ObTableLoadSqlSta
   ObSchemaGetterGuard *tmp_schema_guard = nullptr;
   ObSchemaGetterGuard *tmp_schema_guard2 = nullptr;
   const ObTableSchema *table_schema = nullptr;
-  if (sql_statistics.is_empty()) {
+  if (OB_UNLIKELY(nullptr == ctx || sql_statistics.is_empty())) {
-    ret = OB_ERR_UNEXPECTED;
+    ret = OB_INVALID_ARGUMENT;
-    LOG_WARN("sql statistics is empty", K(ret));
+    LOG_WARN("invalid args", KR(ret), KPC(ctx), K(sql_statistics));
   } else if (OB_FAIL(ObTableLoadSchema::get_table_schema(tenant_id, table_id, schema_guard,
                                                          table_schema))) {
     LOG_WARN("fail to get table schema", KR(ret), K(tenant_id), K(table_id));
   } else {
-    tmp_schema_guard = ctx.get_virtual_table_ctx().schema_guard_;
+    tmp_schema_guard = ctx->get_virtual_table_ctx().schema_guard_;
-    tmp_schema_guard2 = ctx.get_das_ctx().get_schema_guard();
+    tmp_schema_guard2 = ctx->get_das_ctx().get_schema_guard();
-    ctx.get_sql_ctx()->schema_guard_ = &schema_guard;
+    ctx->get_sql_ctx()->schema_guard_ = &schema_guard;
-    ctx.get_das_ctx().get_schema_guard() = &schema_guard;
+    ctx->get_das_ctx().get_schema_guard() = &schema_guard;
   }
   ObSEArray<ObOptColumnStat *, 64> part_column_stats;
   if (OB_FAIL(ret)) {
   } else if (OB_FAIL(sql_statistics.get_col_stat_array(part_column_stats))) {
     LOG_WARN("failed to get column stat array");
   } else if (OB_FAIL(ObIncrementalStatEstimator::derive_global_stat_by_direct_load(
-                 ctx, sql_statistics.table_stat_array_, part_column_stats))) {
+               *ctx, sql_statistics.table_stat_array_, part_column_stats))) {
     LOG_WARN("fail to drive global stat by direct load", KR(ret));
   }
-  ctx.get_sql_ctx()->schema_guard_ = tmp_schema_guard;
+  ctx->get_sql_ctx()->schema_guard_ = tmp_schema_guard;
-  ctx.get_das_ctx().get_schema_guard() = tmp_schema_guard2;
+  ctx->get_das_ctx().get_schema_guard() = tmp_schema_guard2;
   return ret;
 }
 

src/observer/table_load/ob_table_load_coordinator.h

@@ -14,6 +14,7 @@ namespace oceanbase
 {
 namespace observer
 {
+class ObTableLoadExecCtx;
 class ObTableLoadTableCtx;
 class ObTableLoadCoordinatorCtx;
 class ObTableLoadCoordinatorTrans;
@@ -26,7 +27,7 @@ public:
   ObTableLoadCoordinator(ObTableLoadTableCtx *ctx);
   static bool is_ctx_inited(ObTableLoadTableCtx *ctx);
   static int init_ctx(ObTableLoadTableCtx *ctx, const common::ObIArray<int64_t> &idx_array,
-                      uint64_t user_id);
+                      uint64_t user_id, ObTableLoadExecCtx *exec_ctx);
   static void abort_ctx(ObTableLoadTableCtx *ctx);
   int init();
   bool is_valid() const { return is_inited_; }
@@ -39,8 +40,8 @@ private:
 public:
   int begin();
   int finish();
-  int commit(sql::ObExecContext *exec_ctx, table::ObTableLoadResultInfo &result_info);
+  int commit(table::ObTableLoadResultInfo &result_info);
-  int px_commit_data(sql::ObExecContext *exec_ctx);
+  int px_commit_data();
   int px_commit_ddl();
   int get_status(table::ObTableLoadStatusType &status, int &error_code);
 private:
@@ -50,7 +51,7 @@ private:
   int start_merge_peers();
   int commit_peers(table::ObTableLoadSqlStatistics &sql_statistics);
   int commit_redef_table();
-  int drive_sql_stat(sql::ObExecContext &ctx, table::ObTableLoadSqlStatistics &sql_statistics);
+  int drive_sql_stat(sql::ObExecContext *ctx, table::ObTableLoadSqlStatistics &sql_statistics);
 private:
   int add_check_merge_result_task();
   int check_peers_merge_result(bool &is_finish);

src/observer/table_load/ob_table_load_coordinator_ctx.cpp

@@ -27,6 +27,7 @@ ObTableLoadCoordinatorCtx::ObTableLoadCoordinatorCtx(ObTableLoadTableCtx *ctx)
   : ctx_(ctx),
     allocator_("TLD_CoordCtx"),
     task_scheduler_(nullptr),
+    exec_ctx_(nullptr),
     last_trans_gid_(1024),
     next_session_id_(0),
     status_(ObTableLoadStatusType::NONE),
@@ -40,17 +41,20 @@ ObTableLoadCoordinatorCtx::~ObTableLoadCoordinatorCtx()
   destroy();
 }
 
-int ObTableLoadCoordinatorCtx::init(const ObIArray<int64_t> &idx_array, uint64_t user_id)
+int ObTableLoadCoordinatorCtx::init(const ObIArray<int64_t> &idx_array, uint64_t user_id,
+                                    ObTableLoadExecCtx *exec_ctx)
 {
   int ret = OB_SUCCESS;
   if (IS_INIT) {
     ret = OB_INIT_TWICE;
     LOG_WARN("ObTableLoadCoordinatorCtx init twice", KR(ret), KP(this));
-  } else if (OB_UNLIKELY(idx_array.count() != ctx_->param_.column_count_ ||
+  } else if (OB_UNLIKELY(
-                         OB_INVALID_ID == user_id)) {
+               idx_array.count() != ctx_->param_.column_count_ || OB_INVALID_ID == user_id ||
+               nullptr == exec_ctx || !exec_ctx->is_valid() ||
+               (ctx_->param_.online_opt_stat_gather_ && nullptr == exec_ctx->get_exec_ctx()))) {
     ret = OB_INVALID_ARGUMENT;
-    LOG_WARN("invalid args", KR(ret), K(idx_array.count()), K_(ctx_->param_.column_count),
+    LOG_WARN("invalid args", KR(ret), K(ctx_->param_), K(idx_array.count()), K(user_id),
-             K(user_id));
+             KPC(exec_ctx));
   } else {
     allocator_.set_tenant_id(MTL_ID());
     if (OB_FAIL(target_schema_.init(ctx_->param_.tenant_id_, ctx_->ddl_param_.dest_table_id_))) {
@@ -116,6 +120,7 @@ int ObTableLoadCoordinatorCtx::init(const ObIArray<int64_t> &idx_array, uint64_t
       LOG_WARN("fail to start task scheduler", KR(ret));
     }
     if (OB_SUCC(ret)) {
+      exec_ctx_ = exec_ctx;
       is_inited_ = true;
     } else {
       destroy();

src/observer/table_load/ob_table_load_coordinator_ctx.h

@@ -7,6 +7,7 @@
 #include "lib/container/ob_se_array.h"
 #include "lib/hash/ob_hashmap.h"
 #include "lib/hash/ob_link_hashmap.h"
+#include "observer/table_load/ob_table_load_exec_ctx.h"
 #include "observer/table_load/ob_table_load_object_allocator.h"
 #include "observer/table_load/ob_table_load_partition_calc.h"
 #include "observer/table_load/ob_table_load_partition_location.h"
@@ -33,7 +34,8 @@ class ObTableLoadCoordinatorCtx
 public:
   ObTableLoadCoordinatorCtx(ObTableLoadTableCtx *ctx);
   ~ObTableLoadCoordinatorCtx();
-  int init(const common::ObIArray<int64_t> &idx_array, uint64_t user_id);
+  int init(const common::ObIArray<int64_t> &idx_array, uint64_t user_id,
+           ObTableLoadExecCtx *exec_ctx);
   void stop();
   void destroy();
   bool is_valid() const { return is_inited_; }
@@ -122,6 +124,7 @@ public:
   ObTableLoadPartitionCalc partition_calc_;
   ObITableLoadTaskScheduler *task_scheduler_;
   common::ObArray<int64_t> idx_array_;
+  ObTableLoadExecCtx *exec_ctx_;
   table::ObTableLoadResultInfo result_info_;
   common::ObString credential_;
   share::schema::ObSequenceSchema sequence_schema_;

src/observer/table_load/ob_table_load_exec_ctx.cpp

@@ -11,7 +11,32 @@ namespace oceanbase
 {
 namespace observer
 {
-int ObTableLoadExecCtx::check_status()
+using namespace common;
+using namespace sql;
+
+/**
+ * ObTableLoadSqlExecCtx
+ */
+
+ObIAllocator *ObTableLoadSqlExecCtx::get_allocator()
+{
+  ObIAllocator *allocator = nullptr;
+  if (nullptr != exec_ctx_) {
+    allocator = &exec_ctx_->get_allocator();
+  }
+  return allocator;
+}
+
+ObSQLSessionInfo *ObTableLoadSqlExecCtx::get_session_info()
+{
+  ObSQLSessionInfo *session_info = nullptr;
+  if (nullptr != exec_ctx_) {
+    session_info = exec_ctx_->get_my_session();
+  }
+  return session_info;
+}
+
+int ObTableLoadSqlExecCtx::check_status()
 {
   int ret = OB_SUCCESS;
   if (OB_UNLIKELY(SS_STOPPING == GCTX.status_ || SS_STOPPED == GCTX.status_)) {
@@ -23,5 +48,19 @@ int ObTableLoadExecCtx::check_status()
   return ret;
 }
 
-}  // namespace observer
+/**
-}  // namespace oceanbase
+ * ObTableLoadClientExecCtx
+ */
+
+int ObTableLoadClientExecCtx::check_status()
+{
+  int ret = OB_SUCCESS;
+  if (OB_UNLIKELY(SS_STOPPING == GCTX.status_ || SS_STOPPED == GCTX.status_)) {
+    ret = OB_SERVER_IS_STOPPING;
+    LOG_WARN("observer is stopped", KR(ret), K(GCTX.status_));
+  }
+  return ret;
+}
+
+} // namespace observer
+} // namespace oceanbase

src/observer/table_load/ob_table_load_exec_ctx.h

@@ -5,26 +5,61 @@
 #pragma once
 
 #include "lib/allocator/ob_allocator.h"
+#include "lib/utility/ob_print_utils.h"
 
 namespace oceanbase
 {
 namespace sql
 {
 class ObExecContext;
-}
+class ObSQLSessionInfo;
+} // namespace sql
 namespace observer
 {
+
 class ObTableLoadExecCtx
 {
 public:
-  ObTableLoadExecCtx() : exec_ctx_(nullptr), allocator_(nullptr) {}
+  ObTableLoadExecCtx() = default;
-  virtual ~ObTableLoadExecCtx() {};
+  virtual ~ObTableLoadExecCtx() = default;
-  virtual int check_status();
+  virtual common::ObIAllocator *get_allocator() = 0;
-
+  virtual sql::ObSQLSessionInfo *get_session_info() = 0;
-public:
+  virtual sql::ObExecContext *get_exec_ctx() = 0; // for sql statistics
-  sql::ObExecContext *exec_ctx_;
+  virtual int check_status() = 0;
-  common::ObIAllocator *allocator_;
+  virtual bool is_valid() const = 0;
+  DECLARE_PURE_VIRTUAL_TO_STRING;
 };
 
-}  // namespace observer
+class ObTableLoadSqlExecCtx : public ObTableLoadExecCtx
-}  // namespace oceanbase
+{
+public:
+  ObTableLoadSqlExecCtx() : exec_ctx_(nullptr) {}
+  virtual ~ObTableLoadSqlExecCtx() = default;
+  common::ObIAllocator *get_allocator() override;
+  sql::ObSQLSessionInfo *get_session_info() override;
+  sql::ObExecContext *get_exec_ctx() override { return exec_ctx_; }
+  int check_status() override;
+  bool is_valid() const override { return nullptr != exec_ctx_; }
+  TO_STRING_KV(KP_(exec_ctx));
+public:
+  sql::ObExecContext *exec_ctx_;
+};
+
+class ObTableLoadClientExecCtx : public ObTableLoadExecCtx
+{
+public:
+  ObTableLoadClientExecCtx() : allocator_(nullptr), session_info_(nullptr) {}
+  virtual ~ObTableLoadClientExecCtx() = default;
+  common::ObIAllocator *get_allocator() override { return allocator_; }
+  sql::ObSQLSessionInfo *get_session_info() override { return session_info_; }
+  sql::ObExecContext *get_exec_ctx() override { return nullptr; } // not support sql statistics
+  int check_status() override;
+  bool is_valid() const override { return nullptr != allocator_ && nullptr != session_info_; }
+  TO_STRING_KV(KP_(allocator), KP_(session_info));
+public:
+  common::ObIAllocator *allocator_;
+  sql::ObSQLSessionInfo *session_info_;
+};
+
+} // namespace observer
+} // namespace oceanbase

src/observer/table_load/ob_table_load_instance.cpp

@@ -56,12 +56,12 @@ int ObTableLoadInstance::init(ObTableLoadParam &param, const ObIArray<int64_t> &
   if (IS_INIT) {
     ret = OB_INIT_TWICE;
     LOG_WARN("ObTableLoadInstance init twice", KR(ret), KP(this));
-  } else if (OB_UNLIKELY(nullptr == execute_ctx->exec_ctx_)) {
+  } else if (OB_UNLIKELY(!param.is_valid() || !execute_ctx->is_valid())) {
     ret = OB_INVALID_ARGUMENT;
-    LOG_WARN("invalid args", KR(ret), K(param), K(execute_ctx));
+    LOG_WARN("invalid args", KR(ret), K(param), KPC(execute_ctx));
   } else {
     execute_ctx_ = execute_ctx;
-    allocator_ = execute_ctx->allocator_;
+    allocator_ = execute_ctx->get_allocator();
     if (OB_FAIL(param.normalize())) {
       LOG_WARN("fail to normalize param", KR(ret));
     }
@@ -70,7 +70,7 @@ int ObTableLoadInstance::init(ObTableLoadParam &param, const ObIArray<int64_t> &
       LOG_WARN("fail to check support direct load", KR(ret), K(param.table_id_));
     }
     // create table ctx
-    else if (OB_FAIL(create_table_ctx(param, idx_array, execute_ctx_->exec_ctx_->get_my_session()))) {
+    else if (OB_FAIL(create_table_ctx(param, idx_array))) {
       LOG_WARN("fail to create table ctx", KR(ret));
     }
     // begin
@@ -91,11 +91,11 @@ int ObTableLoadInstance::init(ObTableLoadParam &param, const ObIArray<int64_t> &
 }
 
 int ObTableLoadInstance::create_table_ctx(ObTableLoadParam &param,
-                                          const ObIArray<int64_t> &idx_array,
+                                          const ObIArray<int64_t> &idx_array)
-                                          sql::ObSQLSessionInfo *session_info)
 {
   int ret = OB_SUCCESS;
   ObTableLoadTableCtx *table_ctx = nullptr;
+  ObSQLSessionInfo *session_info = execute_ctx_->get_session_info();
   ObTableLoadDDLParam ddl_param;
   // start redef table
   ObTableLoadRedefTableStartArg start_arg;
@@ -122,8 +122,8 @@ int ObTableLoadInstance::create_table_ctx(ObTableLoadParam &param,
       LOG_WARN("fail to alloc table ctx", KR(ret), K(param));
     } else if (OB_FAIL(table_ctx->init(param, ddl_param, session_info))) {
       LOG_WARN("fail to init table ctx", KR(ret));
-    } else if (OB_FAIL(ObTableLoadCoordinator::init_ctx(table_ctx, idx_array,
+    } else if (OB_FAIL(ObTableLoadCoordinator::init_ctx(
-                                                        session_info->get_priv_user_id()))) {
+                 table_ctx, idx_array, session_info->get_priv_user_id(), execute_ctx_))) {
       LOG_WARN("fail to coordinator init ctx", KR(ret));
     } else if (OB_FAIL(ObTableLoadService::add_ctx(table_ctx))) {
       LOG_WARN("fail to add ctx", KR(ret));
@@ -303,7 +303,7 @@ int ObTableLoadInstance::commit(ObTableLoadResultInfo &result_info)
       LOG_WARN("fail to check merged", KR(ret));
     }
     // commit
-    else if (OB_FAIL(coordinator.commit(execute_ctx_->exec_ctx_, result_info))) {
+    else if (OB_FAIL(coordinator.commit(result_info))) {
       LOG_WARN("fail to commit", KR(ret));
     } else {
       is_committed_ = true;
@@ -333,7 +333,7 @@ int ObTableLoadInstance::px_commit_data()
       LOG_WARN("fail to check merged", KR(ret));
     }
     // commit
-    else if (OB_FAIL(coordinator.px_commit_data(execute_ctx_->exec_ctx_))) {
+    else if (OB_FAIL(coordinator.px_commit_data())) {
       LOG_WARN("fail to do px_commit_data", KR(ret));
     }
   }

src/observer/table_load/ob_table_load_instance.h

@@ -39,8 +39,7 @@ public:
     ATOMIC_AAF(&job_stat_->parsed_bytes_, parsed_bytes);
   }
 private:
-  int create_table_ctx(ObTableLoadParam &param, const common::ObIArray<int64_t> &idx_array,
+  int create_table_ctx(ObTableLoadParam &param, const common::ObIArray<int64_t> &idx_array);
-                       sql::ObSQLSessionInfo *session_info);
   int begin();
   int start_trans();
   int check_trans_committed();

src/observer/table_load/ob_table_load_table_ctx.cpp

@@ -24,7 +24,8 @@ using namespace table;
 using namespace obrpc;
 
 ObTableLoadTableCtx::ObTableLoadTableCtx()
-  : coordinator_ctx_(nullptr),
+  : client_exec_ctx_(nullptr),
+    coordinator_ctx_(nullptr),
     store_ctx_(nullptr),
     job_stat_(nullptr),
     session_info_(nullptr),
@@ -160,7 +161,29 @@ void ObTableLoadTableCtx::unregister_job_stat()
   }
 }
 
-int ObTableLoadTableCtx::init_coordinator_ctx(const ObIArray<int64_t> &idx_array, uint64_t user_id)
+int ObTableLoadTableCtx::init_client_exec_ctx()
+{
+  int ret = OB_SUCCESS;
+  if (IS_NOT_INIT) {
+    ret = OB_NOT_INIT;
+    LOG_WARN("ObTableLoadTableCtx not init", KR(ret));
+  } else if (OB_NOT_NULL(client_exec_ctx_)) {
+    ret = OB_ERR_UNEXPECTED;
+    LOG_WARN("client table ctx already exist", KR(ret));
+  } else {
+    if (OB_ISNULL(client_exec_ctx_ = OB_NEWx(ObTableLoadClientExecCtx, &allocator_))) {
+      ret = OB_ALLOCATE_MEMORY_FAILED;
+      LOG_WARN("fail to new ObTableLoadClientExecCtx", KR(ret));
+    } else {
+      client_exec_ctx_->allocator_ = &allocator_;
+      client_exec_ctx_->session_info_ = session_info_;
+    }
+  }
+  return ret;
+}
+
+int ObTableLoadTableCtx::init_coordinator_ctx(const ObIArray<int64_t> &idx_array, uint64_t user_id,
+                                              ObTableLoadExecCtx *exec_ctx)
 {
   int ret = OB_SUCCESS;
   if (IS_NOT_INIT) {
@@ -174,7 +197,7 @@ int ObTableLoadTableCtx::init_coordinator_ctx(const ObIArray<int64_t> &idx_array
     if (OB_ISNULL(coordinator_ctx = OB_NEWx(ObTableLoadCoordinatorCtx, (&allocator_), this))) {
       ret = OB_ALLOCATE_MEMORY_FAILED;
       LOG_WARN("fail to new ObTableLoadCoordinatorCtx", KR(ret));
-    } else if (OB_FAIL(coordinator_ctx->init(idx_array, user_id))) {
+    } else if (OB_FAIL(coordinator_ctx->init(idx_array, user_id, exec_ctx))) {
       LOG_WARN("fail to init coordinator ctx", KR(ret));
     } else if (OB_FAIL(coordinator_ctx->set_status_inited())) {
       LOG_WARN("fail to set coordinator status inited", KR(ret));
@@ -240,6 +263,11 @@ void ObTableLoadTableCtx::stop()
 void ObTableLoadTableCtx::destroy()
 {
   abort_unless(0 == get_ref_count());
+  if (nullptr != client_exec_ctx_) {
+    client_exec_ctx_->~ObTableLoadClientExecCtx();
+    allocator_.free(client_exec_ctx_);
+    client_exec_ctx_ = nullptr;
+  }
   if (nullptr != coordinator_ctx_) {
     coordinator_ctx_->~ObTableLoadCoordinatorCtx();
     allocator_.free(coordinator_ctx_);

src/observer/table_load/ob_table_load_table_ctx.h

@@ -6,11 +6,12 @@
 
 #include "lib/list/ob_dlink_node.h"
 #include "lib/utility/ob_print_utils.h"
+#include "observer/table_load/ob_table_load_exec_ctx.h"
+#include "observer/table_load/ob_table_load_object_allocator.h"
 #include "observer/table_load/ob_table_load_schema.h"
 #include "observer/table_load/ob_table_load_struct.h"
 #include "share/table/ob_table_load_define.h"
 #include "sql/engine/cmd/ob_load_data_utils.h"
-#include "observer/table_load/ob_table_load_object_allocator.h"
 #include "sql/session/ob_sql_session_info.h"
 #include "sql/session/ob_sql_session_mgr.h"
 
@@ -41,7 +42,9 @@ public:
   TO_STRING_KV(K_(param), KP_(coordinator_ctx), KP_(store_ctx), "ref_count", get_ref_count(),
                K_(is_dirty), K_(is_inited));
 public:
-  int init_coordinator_ctx(const common::ObIArray<int64_t> &idx_array, uint64_t user_id);
+  int init_client_exec_ctx();
+  int init_coordinator_ctx(const common::ObIArray<int64_t> &idx_array, uint64_t user_id,
+                           ObTableLoadExecCtx *exec_ctx);
   int init_store_ctx(
     const table::ObTableLoadArray<table::ObTableLoadLSIdAndPartitionId> &partition_id_array,
     const table::ObTableLoadArray<table::ObTableLoadLSIdAndPartitionId> &target_partition_id_array);
@@ -57,6 +60,7 @@ public:
   ObTableLoadParam param_;
   ObTableLoadDDLParam ddl_param_;
   ObTableLoadSchema schema_;
+  ObTableLoadClientExecCtx *client_exec_ctx_; // for java client
   ObTableLoadCoordinatorCtx *coordinator_ctx_; // 只在控制节点构造
   ObTableLoadStoreCtx *store_ctx_; // 只在数据节点构造
   sql::ObLoadDataGID gid_;

src/sql/engine/cmd/ob_load_data_direct_impl.cpp

@@ -76,7 +76,8 @@ bool ObLoadDataDirectImpl::LoadExecuteParam::is_valid() const
  */
 
 ObLoadDataDirectImpl::LoadExecuteContext::LoadExecuteContext()
-  : direct_loader_(nullptr),
+  : allocator_(nullptr),
+    direct_loader_(nullptr),
     job_stat_(nullptr),
     logger_(nullptr)
 {
@@ -84,8 +85,8 @@ ObLoadDataDirectImpl::LoadExecuteContext::LoadExecuteContext()
 
 bool ObLoadDataDirectImpl::LoadExecuteContext::is_valid() const
 {
-  return nullptr != exec_ctx_ && nullptr != allocator_ && nullptr != direct_loader_ &&
+  return exec_ctx_.is_valid() && nullptr != direct_loader_ && nullptr != job_stat_ &&
-         nullptr != job_stat_ && nullptr != logger_;
+         nullptr != logger_;
 }
 
 /**
@@ -1126,7 +1127,7 @@ int ObLoadDataDirectImpl::FileLoadExecutor::execute()
       LOG_WARN("fail to prepare execute", KR(ret));
     }
 
-    while (OB_SUCC(ret) && OB_SUCC(execute_ctx_->check_status())) {
+    while (OB_SUCC(ret) && OB_SUCC(execute_ctx_->exec_ctx_.check_status())) {
       TaskHandle *handle = nullptr;
       if (OB_FAIL(get_next_task_handle(handle))) {
         if (OB_UNLIKELY(OB_ITER_END != ret)) {
@@ -1618,7 +1619,7 @@ int ObLoadDataDirectImpl::MultiFilesLoadTaskProcessor::process()
       }
     }
   }
-  while (OB_SUCC(ret) && OB_SUCC(execute_ctx_->check_status())) {
+  while (OB_SUCC(ret) && OB_SUCC(execute_ctx_->exec_ctx_.check_status())) {
     if (OB_FAIL(handle_->data_buffer_.squash())) {
       LOG_WARN("fail to squash data buffer", KR(ret));
     } else if (OB_FAIL(data_reader_.get_next_raw_buffer(handle_->data_buffer_))) {
@@ -1880,7 +1881,7 @@ int ObLoadDataDirectImpl::execute(ObExecContext &ctx, ObLoadDataStmt &load_stmt)
     if (OB_FAIL(direct_loader_.commit(result_info))) {
       LOG_WARN("fail to commit direct loader", KR(ret));
     } else {
-      ObPhysicalPlanCtx *phy_plan_ctx = execute_ctx_.exec_ctx_->get_physical_plan_ctx();
+      ObPhysicalPlanCtx *phy_plan_ctx = ctx.get_physical_plan_ctx();
       phy_plan_ctx->set_affected_rows(result_info.rows_affected_);
       phy_plan_ctx->set_row_matched_count(total_line_count);
       phy_plan_ctx->set_row_deleted_count(result_info.deleted_);
@@ -2078,7 +2079,7 @@ int ObLoadDataDirectImpl::init_store_column_idxs(ObIArray<int64_t> &store_column
 int ObLoadDataDirectImpl::init_execute_context()
 {
   int ret = OB_SUCCESS;
-  execute_ctx_.exec_ctx_ = ctx_;
+  execute_ctx_.exec_ctx_.exec_ctx_ = ctx_;
   execute_ctx_.allocator_ = &ctx_->get_allocator();
   ObTableLoadParam load_param;
   load_param.tenant_id_ = execute_param_.tenant_id_;
@@ -2094,8 +2095,8 @@ int ObLoadDataDirectImpl::init_execute_context()
   load_param.sql_mode_ = execute_param_.sql_mode_;
   load_param.px_mode_ = false;
   load_param.online_opt_stat_gather_ = execute_param_.online_opt_stat_gather_;
-  if (OB_FAIL(direct_loader_.init(load_param,
+  if (OB_FAIL(direct_loader_.init(load_param, execute_param_.store_column_idxs_,
-      execute_param_.store_column_idxs_, &execute_ctx_))) {
+                                  &execute_ctx_.exec_ctx_))) {
     LOG_WARN("fail to init direct loader", KR(ret));
   } else if (OB_FAIL(init_logger())) {
     LOG_WARN("fail to init logger", KR(ret));

src/sql/engine/cmd/ob_load_data_direct_impl.h

@@ -92,13 +92,15 @@ private:
       store_column_idxs_; // Mapping of stored columns to source data columns
   };
 
-  struct LoadExecuteContext : public observer::ObTableLoadExecCtx
+  struct LoadExecuteContext
   {
   public:
     LoadExecuteContext();
     bool is_valid() const;
-    TO_STRING_KV(KP_(exec_ctx), KP_(allocator), KP_(direct_loader), KP_(job_stat), KP_(logger));
+    TO_STRING_KV(K_(exec_ctx), KP_(allocator), KP_(direct_loader), KP_(job_stat), KP_(logger));
   public:
+    observer::ObTableLoadSqlExecCtx exec_ctx_;
+    common::ObIAllocator *allocator_;
     observer::ObTableLoadInstance *direct_loader_;
     sql::ObLoadDataStat *job_stat_;
     Logger *logger_;

src/sql/engine/cmd/ob_table_direct_insert_ctx.cpp

@@ -37,16 +37,15 @@ int ObTableDirectInsertCtx::init(ObExecContext *exec_ctx,
     ret = OB_INVALID_ARGUMENT;
     LOG_WARN("exec_ctx cannot be null", KR(ret));
   } else {
-    load_exec_ctx_ = (ObTableLoadExecCtx *)exec_ctx->get_allocator().alloc(sizeof(ObTableLoadExecCtx));
+    if (OB_ISNULL(load_exec_ctx_ = OB_NEWx(ObTableLoadSqlExecCtx, &exec_ctx->get_allocator()))) {
-    table_load_instance_ = (ObTableLoadInstance *)exec_ctx->get_allocator().alloc(sizeof(ObTableLoadInstance));
-    if (OB_ISNULL(load_exec_ctx_) || OB_ISNULL(table_load_instance_)) {
       ret = OB_ALLOCATE_MEMORY_FAILED;
-      LOG_WARN("failed to allocate memory", KR(ret));
+      LOG_WARN("fail to new ObTableLoadSqlExecCtx", KR(ret));
+    } else if (OB_ISNULL(table_load_instance_ =
+                           OB_NEWx(ObTableLoadInstance, &exec_ctx->get_allocator()))) {
+      ret = OB_ALLOCATE_MEMORY_FAILED;
+      LOG_WARN("fail to new ObTableLoadInstance", KR(ret));
     } else {
-      new (load_exec_ctx_) ObTableLoadExecCtx;
-      new (table_load_instance_) ObTableLoadInstance;
       load_exec_ctx_->exec_ctx_ = exec_ctx;
-      load_exec_ctx_->allocator_ = &(exec_ctx->get_allocator());
       uint64_t sql_mode = 0;
       ObSEArray<int64_t, 16> store_column_idxs;
       omt::ObTenant *tenant = nullptr;
@@ -118,7 +117,7 @@ void ObTableDirectInsertCtx::destroy()
     table_load_instance_ = nullptr;
   }
   if (OB_NOT_NULL(load_exec_ctx_)) {
-    load_exec_ctx_->~ObTableLoadExecCtx();
+    load_exec_ctx_->~ObTableLoadSqlExecCtx();
     load_exec_ctx_ = nullptr;
   }
 }

src/sql/engine/cmd/ob_table_direct_insert_ctx.h

@@ -10,7 +10,7 @@ namespace oceanbase
 {
 namespace observer
 {
-class ObTableLoadExecCtx;
+class ObTableLoadSqlExecCtx;
 class ObTableLoadInstance;
 }
 
@@ -36,7 +36,7 @@ private:
   int init_store_column_idxs(const uint64_t tenant_id, const uint64_t table_id,
                              common::ObIArray<int64_t> &store_column_idxs);
 private:
-  observer::ObTableLoadExecCtx *load_exec_ctx_;
+  observer::ObTableLoadSqlExecCtx *load_exec_ctx_;
   observer::ObTableLoadInstance *table_load_instance_;
   bool is_inited_;
 };