[CP] [fix] insert on dup error 5500

Co-authored-by: footka <672528926@qq.com>
Co-authored-by: 884244693 <884244693@qq.com>

Author: 3 people

src/share/vector_index/ob_hybrid_vector_refresh_task.cpp

@@ -425,7 +425,10 @@ int ObHybridVectorRefreshTask::get_embedded_table_column_ids(ObPluginVectorIndex
     }
   }
   task_ctx->part_key_num_ = part_key_column_ids.count();
-  if (OB_SUCC(ret) && OB_FAIL(task_ctx->embedded_table_column_ids_.push_back(vector_column_id))) {
+  if (OB_FAIL(ret)) {
+  } else if (OB_FAIL(task_ctx->embedded_table_column_ids_.push_back(vector_column_id))) {
+    LOG_WARN("failed to push vector column id.", K(ret));
+  } else if (OB_FAIL(task_ctx->embedded_table_update_ids_.push_back(vector_column_id))) {
     LOG_WARN("failed to push vector column id.", K(ret));
   }
 
@@ -902,7 +905,7 @@ int ObHybridVectorRefreshTask::after_embedding(ObPluginVectorIndexAdaptor &adapt
   ObArray<float*> output_vector;
   float *vector_buf = nullptr;
   ObHybridVectorRefreshTaskCtx *task_ctx = static_cast<ObHybridVectorRefreshTaskCtx *>(get_task_ctx());
-  storage::ObValueRowIterator embedded_iter;
+  ObVecIndexATaskUpdIterator embedded_iter;
   storage::ObValueRowIterator index_id_iter;
   storage::ObValueRowIterator delta_delete_iter;
   embedded_iter.init();
@@ -946,13 +949,15 @@ int ObHybridVectorRefreshTask::after_embedding(ObPluginVectorIndexAdaptor &adapt
     ret = OB_ALLOCATE_MEMORY_FAILED;
     LOG_WARN("failed to alloc mem.", K(ret), K(dim));
   } else {
-    HEAP_VARS_3((blocksstable::ObDatumRow, datum_row, tenant_id_), (storage::ObTableScanParam, vid_rowkey_scan_param), (schema::ObTableParam, vid_rowkey_table_param, allocator_)) {
+    HEAP_VARS_4((blocksstable::ObDatumRow, datum_row, tenant_id_), (blocksstable::ObDatumRow, new_row, tenant_id_), (storage::ObTableScanParam, vid_rowkey_scan_param), (schema::ObTableParam, vid_rowkey_table_param, allocator_)) {
       ObArenaAllocator scan_allocator("VecEmbedding", OB_MALLOC_NORMAL_BLOCK_SIZE, MTL_ID());
       common::ObNewRowIterator *vid_rowkey_iter = nullptr;
       ObTableScanIterator *table_scan_iter = nullptr;
       int64_t loop_cnt = 0;
       if (OB_FAIL(datum_row.init(task_ctx->embedded_table_column_ids_.count()))) {
         LOG_WARN("fail to init datum row", K(ret), K(task_ctx->embedded_table_column_ids_), K(datum_row));
+      } else if (OB_FAIL(new_row.init(task_ctx->embedded_table_column_ids_.count()))) {
+        LOG_WARN("fail to init datum row", K(ret), K(task_ctx->embedded_table_column_ids_), K(new_row));
       } else if (adaptor.get_is_need_vid() && OB_FAIL(ObPluginVectorIndexUtils::read_local_tablet(ls_id_,
               &adaptor,
               ctx_->task_status_.target_scn_,
@@ -970,7 +975,7 @@ int ObHybridVectorRefreshTask::after_embedding(ObPluginVectorIndexAdaptor &adapt
         for (int64_t i = 0; i < dim; i++) {
           vector_buf[i] = output_vector.at(row_id)[i];
         }
-        datum_row.storage_datums_[task_ctx->embedded_table_column_ids_.count() - 1].set_string(reinterpret_cast<char *>(vector_buf), dim * sizeof(float));
+        datum_row.storage_datums_[task_ctx->embedded_table_column_ids_.count() - 1].set_null();
         for (int i = task_ctx->embedded_table_column_ids_.count() - 2; i > task_ctx->embedded_table_column_ids_.count() - 2 - task_ctx->part_key_num_; i--) {
           datum_row.storage_datums_[i].set_null();
         }
@@ -1007,11 +1012,13 @@ int ObHybridVectorRefreshTask::after_embedding(ObPluginVectorIndexAdaptor &adapt
           }
         }
         if (OB_SUCC(ret)) {
-          if (OB_FAIL(embedded_iter.add_row(datum_row))) {
+          if (OB_FAIL(new_row.deep_copy(datum_row, allocator_))) {
+            LOG_WARN("failed to copy row", K(ret), K(datum_row));
+          } else if (FALSE_IT(new_row.storage_datums_[task_ctx->embedded_table_column_ids_.count() - 1].set_string(reinterpret_cast<char *>(vector_buf), dim * sizeof(float)))) {
+          } else if (OB_FAIL(embedded_iter.add_row(datum_row, new_row))) {
             LOG_WARN("failed to add row to index id iter", K(ret));
           }
           datum_row.reuse();
-
         }
 
         CHECK_TASK_CANCELLED_IN_PROCESS(ret, loop_cnt, ctx_);
@@ -1039,7 +1046,7 @@ int ObHybridVectorRefreshTask::after_embedding(ObPluginVectorIndexAdaptor &adapt
       LOG_WARN("unexpected error", K(ret), KPC(task_ctx), K(oas));
     } else if (OB_FAIL(init_dml_param(adaptor.get_embedded_table_id(), dml_param, table_dml_param, task_ctx->embedded_table_column_ids_, tx_desc, snapshot, store_ctx_guard))) {
       LOG_WARN("failed to init dml param", K(ret), K(dml_param), K(table_dml_param));
-    } else if (OB_FAIL(oas->insert_rows(ls_id_, adaptor.get_embedded_tablet_id(), *tx_desc, dml_param, task_ctx->embedded_table_column_ids_, &embedded_iter, affected_rows))) {
+    } else if (OB_FAIL(oas->update_rows(ls_id_, adaptor.get_embedded_tablet_id(), *tx_desc, dml_param, task_ctx->embedded_table_column_ids_, task_ctx->embedded_table_update_ids_, &embedded_iter, affected_rows))) {
       LOG_WARN("failed to insert rows to embedded table", K(ret), K(adaptor.get_embedded_tablet_id()));
     }
     store_ctx_guard.reset();

src/share/vector_index/ob_hybrid_vector_refresh_task.h

@@ -72,6 +72,7 @@ struct ObHybridVectorRefreshTaskCtx : public ObVecIndexAsyncTaskCtx
         embedding_task_(nullptr),
         index_id_column_ids_(),
         embedded_table_column_ids_(),
+        embedded_table_update_ids_(),
         ai_service_(),
         endpoint_(),
         adp_guard_(),
@@ -97,6 +98,7 @@ struct ObHybridVectorRefreshTaskCtx : public ObVecIndexAsyncTaskCtx
   ObEmbeddingTask *embedding_task_;
   ObSEArray<uint64_t, 4> index_id_column_ids_;
   ObSEArray<uint64_t, 4> embedded_table_column_ids_;
+  ObSEArray<uint64_t, 4> embedded_table_update_ids_;
   omt::ObAiServiceGuard ai_service_;
   const ObAiModelEndpointInfo *endpoint_;
   ObPluginVectorIndexAdapterGuard adp_guard_;

src/share/vector_index/ob_vector_index_async_task_util.cpp

@@ -1184,6 +1184,47 @@ int ObVecIndexIAsyncTask::init(
   return ret;
 }
 
+int ObVecIndexATaskUpdIterator::init() {
+  int ret = OB_SUCCESS;
+  if (OB_FAIL(old_row_.init())) {
+    LOG_WARN("fail to init old rows iter", K(ret));
+  } else if (OB_FAIL(new_row_.init())) {
+    LOG_WARN("fail to init new rows iter", K(ret));
+  }
+  return ret;
+}
+
+int ObVecIndexATaskUpdIterator::add_row(blocksstable::ObDatumRow &old_datum_row, blocksstable::ObDatumRow &new_datum_row) {
+  int ret = OB_SUCCESS;
+  if (OB_FAIL(old_row_.add_row(old_datum_row))) {
+    LOG_WARN("failed to add row to iter", K(ret));
+  } else if (OB_FAIL(new_row_.add_row(new_datum_row))) {
+    LOG_WARN("fail to init new rows iter", K(ret));
+  }
+  return ret;
+}
+
+int ObVecIndexATaskUpdIterator::get_next_row(blocksstable::ObDatumRow *&row)
+{
+  int ret = OB_SUCCESS;
+  if (!got_old_row_) {
+    got_old_row_ = true;
+    if (OB_FAIL(old_row_.get_next_row(row))) {
+      if (OB_ITER_END != ret) {
+        LOG_WARN("fail to get next old row", K(ret));
+      }
+    }
+  } else {
+    got_old_row_ = false;
+    if (OB_FAIL(new_row_.get_next_row(row))) {
+      if (OB_ITER_END != ret) {
+        LOG_WARN("fail to get next new row", K(ret));
+      }
+    }
+  }
+  return ret;
+}
+
 /**************************** ObVecIndexAsyncTask ******************************/
 int ObVecIndexAsyncTask::do_work()
 {

src/share/vector_index/ob_vector_index_async_task_util.h

@@ -25,6 +25,7 @@
 #include "lib/thread/thread_mgr_interface.h"
 #include "storage/access/ob_dml_param.h"
 #include "storage/tx/ob_trans_define_v4.h"
+#include "storage/ob_value_row_iterator.h"
 
 namespace oceanbase
 {
@@ -316,6 +317,36 @@ class ObVecIndexIAsyncTask
   DISALLOW_COPY_AND_ASSIGN(ObVecIndexIAsyncTask);
 };
 
+class ObVecIndexATaskUpdIterator : public blocksstable::ObDatumRowIterator
+{
+public:
+  ObVecIndexATaskUpdIterator()
+    : got_old_row_(false),
+      is_iter_end_(false)
+  {}
+
+  virtual ~ObVecIndexATaskUpdIterator() {
+    old_row_.reset();
+    new_row_.reset();
+  }
+
+  int init();
+  int add_row(blocksstable::ObDatumRow &old_datum_row, blocksstable::ObDatumRow &new_datum_row);
+
+  virtual int get_next_row(blocksstable::ObDatumRow *&row) override;
+  virtual void reset() override {}
+
+private:
+  // disallow copy
+  DISALLOW_COPY_AND_ASSIGN(ObVecIndexATaskUpdIterator);
+
+private:
+  storage::ObValueRowIterator old_row_;
+  storage::ObValueRowIterator new_row_;
+  bool got_old_row_;
+  bool is_iter_end_;
+};
+
 class ObVecIndexAsyncTask : public ObVecIndexIAsyncTask
 {
 public:

src/sql/das/ob_das_dml_vec_iter.cpp

@@ -686,16 +686,14 @@ int ObEmbeddedVecDMLIterator::generate_domain_rows(const ObChunkDatumStore::Stor
     bool is_sync_interval = false;
     if (OB_FAIL(check_sync_interval(is_sync_interval))) {
       LOG_WARN("fail to check sync interval", K(ret));
-    } else if (is_sync_interval && OB_FAIL(generate_embedded_vec_row(store_row))) {
+    } else if (OB_FAIL(generate_embedded_vec_row(store_row, is_sync_interval))) {
       LOG_WARN("failed to generate embedded vec row", K(ret));
-    } else if (!is_sync_interval) {
-      ret = OB_ITER_END;
     }
   }
   return ret;
 }
 
-int ObEmbeddedVecDMLIterator::generate_embedded_vec_row(const ObChunkDatumStore::StoredRow *store_row)
+int ObEmbeddedVecDMLIterator::generate_embedded_vec_row(const ObChunkDatumStore::StoredRow *store_row, bool is_sync)
 {
   int ret = OB_SUCCESS;
 
@@ -729,14 +727,14 @@ int ObEmbeddedVecDMLIterator::generate_embedded_vec_row(const ObChunkDatumStore:
           int64_t vid = OB_INVALID_ID;
           if (OB_FAIL(get_vid(store_row, vid_idx, vid))) {
             LOG_WARN("failed to get vid", K(ret));
+          } else if (!is_sync) {
+            obj_arr[vid_idx].set_int(vid);
+            obj_arr[embedded_vec_idx].set_null();
           } else if (OB_FAIL(get_chunk_data(store_row, embedded_vec_idx, chunk))) {
             LOG_WARN("failed to project chunk columns for embedding", K(ret));
           } else if (!is_old_row_ && chunk.empty()) {
             obj_arr[vid_idx].set_int(vid);
             obj_arr[embedded_vec_idx].set_null();
-            if (OB_FAIL(rows_.push_back(row))) {
-              LOG_WARN("fail to push back row", K(ret));
-            }
           } else {
             obj_arr[vid_idx].set_int(vid);
             ObString embedded_vector;

src/sql/das/ob_das_dml_vec_iter.h

@@ -159,7 +159,7 @@ class ObEmbeddedVecDMLIterator final : public ObDomainDMLIterator
 protected:
   int generate_domain_rows(const ObChunkDatumStore::StoredRow *store_row) override;
 private:
-  int generate_embedded_vec_row(const ObChunkDatumStore::StoredRow *store_row);
+  int generate_embedded_vec_row(const ObChunkDatumStore::StoredRow *store_row, bool is_sync);
   int get_embedded_vec_column_idxs(int64_t &vid_idx, int64_t &embedded_vec_idx);
   int get_vid(const ObChunkDatumStore::StoredRow *store_row, const int64_t vid_idx, int64_t &vid);
   int get_chunk_data(const ObChunkDatumStore::StoredRow *store_row, const int64_t embedded_vec_idx, ObString &chunk);