fix lob macro block logic id duplicate after direct load.

deps/oblib/src/lib/utility/ob_tracepoint.h

@@ -637,6 +637,7 @@ class EventTable
       EN_DDL_REPORT_CHECKSUM_FAIL = 515,
       EN_DDL_REPORT_REPLICA_BUILD_STATUS_FAIL = 516,
       EN_DDL_DIRECT_LOAD_WAIT_TABLE_LOCK_FAIL = 517,
+      EN_DDL_LOBID_CACHE_SIZE_INJECTED = 518,
 
       // SQL Optimizer related 551-599
       EN_EXPLAIN_GENERATE_PLAN_WITH_OUTLINE = 551,

src/sql/engine/pdml/static/ob_px_sstable_insert_op.cpp

@@ -211,9 +211,16 @@ int ObPxMultiPartSSTableInsertOp::inner_get_next_row()
                                                                     slice_info))) {
         LOG_WARN("create sstable slice writer failed", K(ret), K(block_start_seq), K(slice_info));
       } else {
-        ObDDLInsertRowIterator row_iter(this, is_current_slice_empty /*is_slice_empty*/,
+        ObDDLInsertRowIterator row_iter;
-          notify_ls_id, notify_tablet_id, table_schema->get_rowkey_column_num(), snapshot_version_, slice_info.context_id_, parallel_idx);
+        if (OB_FAIL(row_iter.init(this,
-        if (OB_FAIL(tenant_direct_load_mgr->fill_sstable_slice(slice_info,
+                                is_current_slice_empty,
+                                notify_ls_id,
+                                notify_tablet_id,
+                                table_schema->get_rowkey_column_num(),
+                                snapshot_version_,
+                                slice_info.context_id_, parallel_idx))) {
+          LOG_WARN("init failed", K(ret));
+        } else if (OB_FAIL(tenant_direct_load_mgr->fill_sstable_slice(slice_info,
                                                               &row_iter,
                                                               affected_rows,
                                                               &insert_monitor))) {
@@ -221,7 +228,8 @@ int ObPxMultiPartSSTableInsertOp::inner_get_next_row()
         }
       }
       if (OB_SUCC(ret)) {
-        if (OB_FAIL(tenant_direct_load_mgr->close_sstable_slice(slice_info, &insert_monitor))) {
+        blocksstable::ObMacroDataSeq unused_seq;
+        if (OB_FAIL(tenant_direct_load_mgr->close_sstable_slice(slice_info, &insert_monitor, unused_seq))) {
           LOG_WARN("close sstable slice failed", K(ret), K(slice_info));
         }
         ctx_.get_physical_plan_ctx()->add_affected_rows(affected_rows);

src/storage/ddl/ob_complement_data_task.cpp

@@ -1358,19 +1358,27 @@ int ObComplementWriteTask::append_row(ObScan *scan)
     ObTenantDirectLoadMgr *tenant_direct_load_mgr = MTL(ObTenantDirectLoadMgr *);
     ObTabletDirectLoadMgrHandle direct_load_hdl;
     bool is_major_sstable_exist = false;
-    ObDDLInsertRowIterator row_iter(nullptr/*ObPxMultiPartSSTableInsertOp*/, false/*is_slice_empty*/,
+    ObDDLInsertRowIterator row_iter;
-          param_->dest_ls_id_, param_->dest_tablet_id_, 0/*unused_rowkey_num*/, param_->snapshot_version_, context_->context_id_, task_id_);
     blocksstable::ObNewRowBuilder new_row_builder;
     int64_t lob_inrow_threshold = OB_DEFAULT_LOB_INROW_THRESHOLD;
-    if (OB_ISNULL(tenant_direct_load_mgr)) {
+    if (OB_UNLIKELY(!is_inited_)) {
-      ret = OB_ERR_UNEXPECTED;
-      LOG_WARN("unexpected err", K(ret), K(MTL_ID()));
-    } else if (OB_UNLIKELY(!is_inited_)) {
       ret = OB_NOT_INIT;
       LOG_WARN("ObComplementWriteTask is not inited", K(ret));
-    } else if (OB_ISNULL(param_) || OB_ISNULL(scan) || OB_UNLIKELY(!param_->is_valid()) || OB_ISNULL(context_)) {
+    } else if (OB_ISNULL(scan)) {
       ret = OB_INVALID_ARGUMENT;
       LOG_WARN("invalid arguments", K(ret));
+    } else if (OB_ISNULL(tenant_direct_load_mgr)) {
+      ret = OB_ERR_UNEXPECTED;
+      LOG_WARN("unexpected err", K(ret), K(MTL_ID()));
+    } else if (OB_FAIL(row_iter.init(nullptr/*ObPxMultiPartSSTableInsertOp*/,
+                              false/*is_slice_empty*/,
+                              param_->dest_ls_id_,
+                              param_->dest_tablet_id_,
+                              0/*unused_rowkey_num*/,
+                              param_->snapshot_version_,
+                              context_->context_id_,
+                              task_id_))) {
+      LOG_WARN("init failed", K(ret));
     } else if (OB_FAIL(macro_start_seq.set_parallel_degree(task_id_))) {
       LOG_WARN("set parallel degree failed", K(ret), K(task_id_));
     } else if (OB_FAIL(context_->check_already_committed(param_->dest_ls_id_, param_->dest_tablet_id_, ddl_committed))) {

src/storage/ddl/ob_direct_insert_sstable_ctx_new.cpp

@@ -599,7 +599,10 @@ int ObTenantDirectLoadMgr::cancel(
   return ret;
 }
 
-int ObTenantDirectLoadMgr::close_sstable_slice(const ObDirectLoadSliceInfo &slice_info, ObInsertMonitor* insert_monitor)
+int ObTenantDirectLoadMgr::close_sstable_slice(
+    const ObDirectLoadSliceInfo &slice_info,
+    ObInsertMonitor* insert_monitor,
+    blocksstable::ObMacroDataSeq &next_seq)
 {
   int ret = OB_SUCCESS;
   ObTabletDirectLoadMgrHandle handle;
@@ -624,7 +627,12 @@ int ObTenantDirectLoadMgr::close_sstable_slice(const ObDirectLoadSliceInfo &slic
   } else if (OB_FAIL(tablet_exec_context_map_.get_refactored(exec_id, exec_context))) {
     LOG_WARN("get tablet execution context failed", K(ret));
   } else if (OB_FAIL(handle.get_obj()->close_sstable_slice(
-      slice_info.is_lob_slice_/*is_data_tablet_process_for_lob*/, slice_info, exec_context.start_scn_, exec_context.execution_id_, insert_monitor))) {
+      slice_info.is_lob_slice_/*is_data_tablet_process_for_lob*/,
+      slice_info,
+      exec_context.start_scn_,
+      exec_context.execution_id_,
+      insert_monitor,
+      next_seq))) {
     LOG_WARN("close sstable slice failed", K(ret), K(slice_info), "execution_start_scn", exec_context.start_scn_, "execution_id", exec_context.execution_id_);
   }
   return ret;
@@ -1731,9 +1739,11 @@ int ObTabletDirectLoadMgr::close_sstable_slice(
     const ObDirectLoadSliceInfo &slice_info,
     const share::SCN &start_scn,
     const int64_t execution_id,
-    ObInsertMonitor *insert_monitor)
+    ObInsertMonitor *insert_monitor,
+    blocksstable::ObMacroDataSeq &next_seq)
 {
   int ret = OB_SUCCESS;
+  next_seq.reset();
   if (OB_UNLIKELY(!is_inited_)) {
     ret = OB_NOT_INIT;
     LOG_WARN("not init", K(ret));
@@ -1748,7 +1758,7 @@ int ObTabletDirectLoadMgr::close_sstable_slice(
       ret = OB_ERR_UNEXPECTED;
       LOG_WARN("unexpected err", K(ret), K(slice_info));
     } else if (OB_FAIL(lob_mgr_handle_.get_obj()->close_sstable_slice(
-        false, slice_info, start_scn, execution_id))) {
+        false, slice_info, start_scn, execution_id, insert_monitor, next_seq))) {
       LOG_WARN("close lob sstable slice failed", K(ret), K(slice_info));
     }
   } else {
@@ -1761,6 +1771,8 @@ int ObTabletDirectLoadMgr::close_sstable_slice(
       LOG_WARN("unexpected err", K(ret), K(slice_info));
     } else if (OB_FAIL(slice_writer->close())) {
       LOG_WARN("close failed", K(ret), K(slice_info));
+    } else if (OB_FALSE_IT(next_seq = slice_writer->get_next_block_start_seq())) {
+      // block start seq after the close operation is the next availabled one.
     } else if (!slice_info.is_lob_slice_ && is_ddl_direct_load(direct_load_type_)) {
       int64_t task_finish_count = -1;
       {

src/storage/ddl/ob_direct_insert_sstable_ctx_new.h

@@ -133,7 +133,10 @@ public:
       const ObArray<common::ObObjMeta> &col_types,
       blocksstable::ObDatumRow &datum_row);
   // flush macro block, close and destroy slice writer.
-  int close_sstable_slice(const ObDirectLoadSliceInfo &slice_info, ObInsertMonitor *insert_monitor = NULL);
+  int close_sstable_slice(
+      const ObDirectLoadSliceInfo &slice_info,
+      ObInsertMonitor *insert_monitor,
+      blocksstable::ObMacroDataSeq &next_seq);
 
   // end direct load due to commit or abort.
   // @param [in] is_full_direct_load.
@@ -337,7 +340,8 @@ public:
       const ObDirectLoadSliceInfo &slice_info,
       const share::SCN &start_scn,
       const int64_t execution_id,
-      ObInsertMonitor *insert_monitor=NULL);
+      ObInsertMonitor *insert_monitor,
+      blocksstable::ObMacroDataSeq &next_seq);
 
   // for ref_cnt
   void inc_ref() { ATOMIC_INC(&ref_cnt_); }

src/storage/ddl/ob_direct_load_struct.cpp

@@ -49,14 +49,22 @@ int ObTabletDirectLoadInsertParam::assign(const ObTabletDirectLoadInsertParam &o
   return ret;
 }
 
-ObDDLInsertRowIterator::ObDDLInsertRowIterator(
+ObDDLInsertRowIterator::ObDDLInsertRowIterator()
-    sql::ObPxMultiPartSSTableInsertOp *op,
+  : is_inited_(false),
-    const bool is_slice_empty, const share::ObLSID &ls_id, const common::ObTabletID &tablet_id,
+    lob_allocator_(ObModIds::OB_LOB_ACCESS_BUFFER, OB_MALLOC_NORMAL_BLOCK_SIZE, MTL_ID()),
-    const int64_t rowkey_cnt, const int64_t snapshot_version, const int64_t context_id, const int64_t parallel_idx)
+    op_(nullptr),
-  : lob_allocator_(ObModIds::OB_LOB_ACCESS_BUFFER, OB_MALLOC_NORMAL_BLOCK_SIZE, MTL_ID()), op_(op), ls_id_(ls_id), current_tablet_id_(tablet_id), current_row_(), is_next_row_cached_(true),
+    ls_id_(),
-    is_slice_empty_(is_slice_empty), rowkey_count_(rowkey_cnt), snapshot_version_(snapshot_version), lob_slice_id_(0), context_id_(context_id), parallel_idx_(parallel_idx)
+    current_tablet_id_(),
+    current_row_(),
+    is_next_row_cached_(true),
+    is_slice_empty_(false),
+    rowkey_count_(-1),
+    snapshot_version_(-1),
+    lob_slice_id_(-1),
+    lob_id_cache_(),
+    context_id_(-1),
+    macro_seq_()
 {
-  lob_id_cache_.set(1/*start*/, 0/*end*/);
 }
 
 ObDDLInsertRowIterator::~ObDDLInsertRowIterator()
@@ -64,10 +72,48 @@ ObDDLInsertRowIterator::~ObDDLInsertRowIterator()
 
 }
 
+int ObDDLInsertRowIterator::init(
+    sql::ObPxMultiPartSSTableInsertOp *op,
+    const bool is_slice_empty,
+    const share::ObLSID &ls_id,
+    const common::ObTabletID &tablet_id,
+    const int64_t rowkey_cnt,
+    const int64_t snapshot_version,
+    const int64_t context_id,
+    const int64_t parallel_idx)
+{
+  int ret = OB_SUCCESS;
+  if (OB_UNLIKELY(is_inited_)) {
+    ret = OB_INIT_TWICE;
+    LOG_WARN("init twice", K(ret));
+  } else if (OB_UNLIKELY(!ls_id.is_valid() || !tablet_id.is_valid() || rowkey_cnt < 0
+      || snapshot_version < 0 || context_id < 0 || parallel_idx < 0)) {
+    ret = OB_INVALID_ARGUMENT;
+    LOG_WARN("invalid arg", K(ret), K(ls_id), K(tablet_id), K(rowkey_cnt), K(snapshot_version), K(context_id), K(parallel_idx));
+  } else if (OB_FAIL(macro_seq_.set_parallel_degree(parallel_idx))) {
+    LOG_WARN("set parallel failed", K(ret), K(parallel_idx));
+  } else {
+    op_ = op;
+    is_slice_empty_ = is_slice_empty;
+    ls_id_ = ls_id;
+    current_tablet_id_ = tablet_id;
+    rowkey_count_ = rowkey_cnt;
+    snapshot_version_ = snapshot_version;
+    context_id_ = context_id;
+    is_next_row_cached_ = true;
+    lob_id_cache_.set(1/*start*/, 0/*end*/);
+    is_inited_ = true;
+  }
+  return ret;
+}
+
 int ObDDLInsertRowIterator::close_lob_sstable_slice()
 {
   int ret = OB_SUCCESS;
-  if (lob_slice_id_ > 0) {
+  if (OB_UNLIKELY(!is_inited_)) {
+    ret = OB_NOT_INIT;
+    LOG_WARN("not init", K(ret));
+  } else if (lob_slice_id_ > 0) {
     ObDirectLoadSliceInfo slice_info;
     slice_info.is_full_direct_load_ = true;
     slice_info.is_lob_slice_ = true;
@@ -76,7 +122,8 @@ int ObDDLInsertRowIterator::close_lob_sstable_slice()
     slice_info.slice_id_ = lob_slice_id_;
     slice_info.context_id_ = context_id_;
     ObTenantDirectLoadMgr *tenant_direct_load_mgr = MTL(ObTenantDirectLoadMgr *);
-    if (OB_FAIL(tenant_direct_load_mgr->close_sstable_slice(slice_info))) {
+    ObMacroDataSeq unused_seq;
+    if (OB_FAIL(tenant_direct_load_mgr->close_sstable_slice(slice_info, nullptr/*insert_monitor*/, unused_seq))) {
       LOG_WARN("close sstable slice failed", K(ret), K(slice_info));
     } else {
       lob_slice_id_ = 0;
@@ -91,7 +138,10 @@ int ObDDLInsertRowIterator::get_next_row(
 {
   int ret = OB_SUCCESS;
   ObTenantDirectLoadMgr *tenant_direct_load_mgr = MTL(ObTenantDirectLoadMgr *);
-  if (OB_UNLIKELY(nullptr == op_ || snapshot_version_ <= 0 || nullptr == tenant_direct_load_mgr)) {
+  if (OB_UNLIKELY(!is_inited_)) {
+    ret = OB_NOT_INIT;
+    LOG_WARN("not init", K(ret));
+  } else if (OB_UNLIKELY(nullptr == op_ || snapshot_version_ <= 0 || nullptr == tenant_direct_load_mgr)) {
     ret = OB_ERR_UNEXPECTED;
     LOG_WARN("operator is null", K(ret), KP(op_), K(snapshot_version_), KP(tenant_direct_load_mgr), K(MTL_ID()));
   } else {
@@ -225,12 +275,19 @@ int ObDDLInsertRowIterator::switch_to_new_lob_slice()
   slice_info.data_tablet_id_ = current_tablet_id_;
   slice_info.slice_id_ = lob_slice_id_;
   slice_info.context_id_ = context_id_;
-  ObMacroDataSeq block_start_seq;
   ObTabletAutoincrementService &auto_inc = ObTabletAutoincrementService::get_instance();
   ObTenantDirectLoadMgr *tenant_direct_load_mgr = MTL(ObTenantDirectLoadMgr *);
   ObTabletDirectLoadMgrHandle direct_load_mgr_handle;
   direct_load_mgr_handle.reset();
-  if (OB_ISNULL(tenant_direct_load_mgr)) {
+  int64_t CACHE_SIZE_REQUESTED = AUTO_INC_CACHE_SIZE;
+#ifdef ERRSIM
+  int64_t negative_inject_num = OB_E(EventTable::EN_DDL_LOBID_CACHE_SIZE_INJECTED) OB_SUCCESS;
+  CACHE_SIZE_REQUESTED = negative_inject_num * -1;
+#endif
+  if (OB_UNLIKELY(!is_inited_)) {
+    ret = OB_NOT_INIT;
+    LOG_WARN("not init", K(ret));
+  } else if (OB_ISNULL(tenant_direct_load_mgr)) {
     ret = OB_ERR_UNEXPECTED;
     LOG_WARN("unexpected err", K(ret), K(MTL_ID()));
   } else if (OB_FAIL(tenant_direct_load_mgr->get_tablet_mgr(current_tablet_id_,
@@ -239,22 +296,20 @@ int ObDDLInsertRowIterator::switch_to_new_lob_slice()
   } else if (OB_FALSE_IT(lob_id_cache_.tablet_id_ =
     direct_load_mgr_handle.get_obj()->get_lob_meta_tablet_id())) {
     // fetch cache via lob meta tablet id.
-  } else if (OB_FALSE_IT(lob_id_cache_.cache_size_ = AUTO_INC_CACHE_SIZE)) {
+  } else if (OB_FALSE_IT(lob_id_cache_.cache_size_ = CACHE_SIZE_REQUESTED)) {
-  } else if (lob_slice_id_ > 0 &&
+  } else if (lob_slice_id_ > 0
-    OB_FAIL(tenant_direct_load_mgr->close_sstable_slice(slice_info))) {
+      && OB_FAIL(tenant_direct_load_mgr->close_sstable_slice(slice_info, nullptr/*insert_monitor*/, macro_seq_))) {
     LOG_WARN("close old lob slice failed", K(ret), K(slice_info));
   } else if (OB_FAIL(auto_inc.get_tablet_cache_interval(MTL_ID(), lob_id_cache_))) {
     LOG_WARN("get_autoinc_seq fail", K(ret), K(MTL_ID()), K(slice_info));
-  } else if (OB_UNLIKELY(AUTO_INC_CACHE_SIZE > lob_id_cache_.count())) {
+  } else if (OB_UNLIKELY(CACHE_SIZE_REQUESTED > lob_id_cache_.count())) {
     ret = OB_ERR_UNEXPECTED;
     LOG_WARN("unexpected autoincrement value count", K(ret), K(lob_id_cache_));
-  } else if (OB_FAIL(block_start_seq.set_parallel_degree(parallel_idx_))) {
-    LOG_WARN("set parall degree failed", K(ret), K(parallel_idx_));
   } else {
     // new slice info to open.
     slice_info.slice_id_ = 0;
-    if (OB_FAIL(tenant_direct_load_mgr->open_sstable_slice(block_start_seq, slice_info))) {
+    if (OB_FAIL(tenant_direct_load_mgr->open_sstable_slice(macro_seq_, slice_info))) {
-      LOG_WARN("open lob sstable slice failed", KR(ret), K(block_start_seq), K(slice_info));
+      LOG_WARN("open lob sstable slice failed", KR(ret), K(macro_seq_), K(slice_info));
     } else {
       lob_slice_id_ = slice_info.slice_id_;
     }

src/storage/ddl/ob_direct_load_struct.h

@@ -264,7 +264,9 @@ public:
 class ObDDLInsertRowIterator : public ObIStoreRowIterator
 {
 public:
-  ObDDLInsertRowIterator(
+  ObDDLInsertRowIterator();
+  virtual ~ObDDLInsertRowIterator();
+  int init(
       sql::ObPxMultiPartSSTableInsertOp *op,
       const bool is_slice_empty,
       const share::ObLSID &ls_id,
@@ -273,15 +275,14 @@ public:
       const int64_t snapshot_version,
       const int64_t context_id,
       const int64_t parallel_idx);
-  virtual ~ObDDLInsertRowIterator();
   virtual int get_next_row(const blocksstable::ObDatumRow *&row) override
   {
     const bool skip_lob = false;
     return get_next_row(skip_lob, row);
   }
   int get_next_row(const bool skip_lob, const blocksstable::ObDatumRow *&row);
-  TO_STRING_KV(K_(ls_id), K_(current_tablet_id), K_(current_row), K_(is_slice_empty), K_(is_next_row_cached), K_(rowkey_count), K_(snapshot_version),
+  TO_STRING_KV(K_(is_inited), K_(ls_id), K_(current_tablet_id), K_(current_row), K_(is_slice_empty), K_(is_next_row_cached), K_(rowkey_count), K_(snapshot_version),
-      K_(lob_slice_id), K_(lob_id_cache), K_(context_id));
+      K_(lob_slice_id), K_(lob_id_cache), K_(context_id), K_(macro_seq));
 public:
   int switch_to_new_lob_slice();
   int close_lob_sstable_slice();
@@ -289,6 +290,7 @@ public:
   inline share::ObTabletCacheInterval &get_lob_id_cache() { return lob_id_cache_; }
 private:
   static const int64_t AUTO_INC_CACHE_SIZE = 5000000; // 500w.
+  bool is_inited_;
   ObArenaAllocator lob_allocator_;
   sql::ObPxMultiPartSSTableInsertOp *op_;
   share::ObLSID ls_id_;
@@ -301,7 +303,7 @@ private:
   int64_t lob_slice_id_;
   share::ObTabletCacheInterval lob_id_cache_;
   int64_t context_id_;
-  int64_t parallel_idx_;
+  blocksstable::ObMacroDataSeq macro_seq_;
 };
 
 class ObLobMetaRowIterator : public ObIStoreRowIterator
@@ -393,6 +395,7 @@ public:
   virtual int append_row(const blocksstable::ObDatumRow &datum_row) = 0;
   virtual int close() = 0;
   virtual int64_t get_row_count() const { return 0; } // dummy one
+  virtual int64_t get_next_block_start_seq() const { return -1; } // invalid block start seq.
   DECLARE_PURE_VIRTUAL_TO_STRING;
 };
 
@@ -438,6 +441,7 @@ public:
       const share::SCN &start_scn);
   virtual int append_row(const blocksstable::ObDatumRow &datum_row) override;
   virtual int close() override;
+  virtual int64_t get_next_block_start_seq() const override { return macro_block_writer_.get_last_macro_seq(); }
   TO_STRING_KV(K(is_inited_), K(macro_block_writer_));
 private:
   bool is_inited_;
@@ -510,6 +514,7 @@ public:
   bool need_column_store() const { return need_column_store_; }
   ObTabletSliceStore *get_slice_store() const { return slice_store_; }
   void cancel() { ATOMIC_SET(&is_canceled_, true); }
+  int64_t get_next_block_start_seq() const { return nullptr == slice_store_ ? 0 /*slice empty*/ : slice_store_->get_next_block_start_seq(); }
   TO_STRING_KV(K(is_inited_), K(need_column_store_), K(is_canceled_), K(start_seq_), KPC(slice_store_), K(row_offset_));
 private:
   int fill_lob_into_memtable( // for version < 4.3.0.0

src/storage/direct_load/ob_direct_load_insert_table_ctx.cpp

@@ -407,7 +407,8 @@ int ObDirectLoadInsertTabletContext::close_sstable_slice(const int64_t slice_id)
     slice_info.data_tablet_id_ = param_.tablet_id_;
     slice_info.slice_id_ = slice_id;
     slice_info.context_id_ = param_.context_id_;
-    if (OB_FAIL(sstable_insert_mgr->close_sstable_slice(slice_info))) {
+    blocksstable::ObMacroDataSeq unused_seq;
+    if (OB_FAIL(sstable_insert_mgr->close_sstable_slice(slice_info, nullptr/*insert_monitor*/, unused_seq))) {
       LOG_WARN("fail to close tablet direct load", KR(ret), K(slice_id),
                K(param_.tablet_id_));
     }
@@ -430,7 +431,8 @@ int ObDirectLoadInsertTabletContext::close_lob_sstable_slice(const int64_t slice
     slice_info.data_tablet_id_ = param_.tablet_id_;
     slice_info.slice_id_ = slice_id;
     slice_info.context_id_ = param_.context_id_;
-    if (OB_FAIL(sstable_insert_mgr->close_sstable_slice(slice_info))) {
+    blocksstable::ObMacroDataSeq unused_seq;
+    if (OB_FAIL(sstable_insert_mgr->close_sstable_slice(slice_info, nullptr/*insert_monitor*/, unused_seq))) {
       LOG_WARN("fail to close tablet direct load", KR(ret), K(slice_id),
                 K(param_.tablet_id_));
     }