[CP] fix semantic index post creating failed and the failure of embedding for overly long data.

obdev · shenyunlong · JLY2015 · ob-robot · commit c9495bac4e09 · 2026-01-23T10:11:08.000Z
Co-authored-by: shenyunlong.syl &lt;ylshen0919@gmail.com&gt;
Co-authored-by: JLY2015 &lt;1623359870@qq.com&gt;
diff --git a/src/share/ob_debug_sync_point.h b/src/share/ob_debug_sync_point.h
@@ -757,6 +757,8 @@ class ObString;
     ACT(BEFORE_INC_MAJOR_DDL_MERGE_UPDATE_TABLE_STORE,)\
     ACT(AFTER_INC_MAJOR_MERGE_GET_SSTABLE,)\
     ACT(AFTER_GET_MTL_TENANT_LOCK,)\
+    ACT(BEFORE_DO_EMBEDDING_TASK,)\
+    ACT(BEFORE_COMPLETE_EMBEDDING_TASK,)\
     ACT(MAX_DEBUG_SYNC_POINT,)
 
 DECLARE_ENUM(ObDebugSyncPoint, debug_sync_point, OB_DEBUG_SYNC_POINT_DEF);
diff --git a/src/share/vector_index/ob_hybrid_vector_refresh_task.cpp b/src/share/vector_index/ob_hybrid_vector_refresh_task.cpp
@@ -508,6 +508,7 @@ int ObHybridVectorRefreshTask::prepare_for_embedding(ObPluginVectorIndexAdaptor
   storage::ObTableScanParam *&table_scan_param = task_ctx->table_scan_param_;
   schema::ObTableParam *&table_param = task_ctx->table_param_;
   storage::ObValueRowIterator &delta_delete_iter = task_ctx->delta_delete_iter_;
+  ObCollationType col_type = CS_TYPE_INVALID;
   int64_t dim = 0;
   int64_t loop_cnt = 0;
   int64_t http_timeout_us = 0;
@@ -528,6 +529,8 @@ int ObHybridVectorRefreshTask::prepare_for_embedding(ObPluginVectorIndexAdaptor
     LOG_WARN("unexpected error", K(ret), KPC(task_ctx));
   } else if (OB_FAIL(adaptor.get_dim(dim))) {
     LOG_WARN("get dim failed", K(ret));
+  } else if (OB_FAIL(ObVectorIndexUtil::get_index_column_collation_type(tenant_id_, adaptor.get_embedded_table_id(), col_type))) {
+    LOG_WARN("failed to get chunc column col_type", K(ret), K(adaptor));
   } else {
     if (OB_NOT_NULL(tsc_iter) || OB_NOT_NULL(table_scan_param) || OB_NOT_NULL(table_param)) {
       if (OB_ISNULL(tsc_iter) || OB_ISNULL(table_scan_param) || OB_ISNULL(table_param)) {
@@ -656,6 +659,7 @@ int ObHybridVectorRefreshTask::prepare_for_embedding(ObPluginVectorIndexAdaptor
           const ObAiModelEndpointInfo *endpoint = task_ctx->endpoint_; // endpoint should not be null after init.
           task_ctx->embedding_task_ = new(task_buf)ObEmbeddingTask(task_ctx->allocator_);
           ObPluginVectorIndexService *service = MTL(ObPluginVectorIndexService *);
+
           if (OB_ISNULL(service)) {
             ret = OB_ERR_UNEXPECTED;
             LOG_WARN("unexpected null ptr", K(ret), KPC(service));
@@ -664,7 +668,8 @@ int ObHybridVectorRefreshTask::prepare_for_embedding(ObPluginVectorIndexAdaptor
           } else if (OB_FAIL(ob_write_string(task_ctx->allocator_, endpoint->get_url(), url, true))) {
             LOG_WARN("fail to write string", K(ret));
           } else if (OB_FAIL(task_ctx->embedding_task_->init(url, endpoint->get_request_model_name(),
-                             endpoint->get_provider(), access_key, chunk_array, dim, http_timeout_us, http_max_retries))) {
+                             endpoint->get_provider(), access_key, chunk_array, col_type, dim, http_timeout_us,
+                             http_max_retries, ctx_->task_status_.task_id_, ObEmbeddingTasSourceType::ASYNC_INDEX))) {
             LOG_WARN("failed to init embedding task", K(ret), KPC(endpoint));
           } else {
             ObEmbeddingTaskHandler *embedding_handler = nullptr;
diff --git a/src/share/vector_index/ob_vector_embedding_handler.cpp b/src/share/vector_index/ob_vector_embedding_handler.cpp
diff --git a/src/share/vector_index/ob_vector_embedding_handler.h b/src/share/vector_index/ob_vector_embedding_handler.h
@@ -155,6 +155,12 @@ class ObEmbeddingTaskPhaseManager {
   }
 };
 
+enum class ObEmbeddingTasSourceType
+{
+  INDEX_PIPELINE = 0,
+  ASYNC_INDEX = 1,
+};
+
 class ObEmbeddingTaskHandler;
 
 // Constants for field lengths
@@ -169,10 +175,14 @@ class ObEmbeddingTask
           const ObString &provider,
           const ObString &user_key,
           const ObIArray<ObString> &input_chunks,
+          const ObCollationType col_type,
           int64_t dimension,
           int64_t http_timeout_us,
           int64_t http_max_retries,
-          storage::ObEmbeddingIOCallbackHandle *cb_handle = nullptr);
+          int64_t source_task_id,
+          ObEmbeddingTasSourceType source_task_type,
+          storage::ObEmbeddingIOCallbackHandle *cb_handle = nullptr,
+          bool always_retry = false);
   template <typename ThreadPoolType>
   int do_work(ThreadPoolType *thread_pool);
   int64_t get_task_id() const { return task_id_; }
@@ -181,29 +191,56 @@ class ObEmbeddingTask
 
   TO_STRING_KV(K_(is_inited),
                 K_(task_id),
+                K_(phase),
                 K_(model_url),
                 K_(model_name),
-                K_(user_key),
                 K(input_chunks_.count()),
                 K(output_vectors_.count()),
                 K_(dimension),
                 K_(batch_size),
+                K_(current_batch_idx),
+                K_(http_total_retry_count),
+                K_(http_retry_count),
+                K_(http_max_retry_count),
+                K_(http_retry_start_time_us),
+                K_(http_last_retry_time_us),
                 K_(processed_chunks),
                 K_(total_chunks),
-                K_(process_callback_offset));
+                K_(process_callback_offset),
+                K_(col_type),
+                K_(current_input_token_limit),
+                K_(need_cancel),
+                K_(source_task_id),
+                K_(source_task_type),
+                K_(always_retry),
+                K_(next_handle_task_time_us),
+                K_(internal_error_code));
   bool is_completed();
   void retain_if_managed();
   void release_if_managed();
   int get_async_result(ObArray<float*> &output_vectors);
   // 公共方法用于外部设置任务失败
   int mark_task_failed(int error_code);
   int maybe_callback();
-  int wait_for_completion();
+  // Wait for task completion with periodic check interval
+  // @param check_interval_us: the interval to wait before returning, must be > 0
+  // @return:
+  //   - OB_INVALID_ARGUMENT: if check_interval_us <= 0
+  //   - OB_NOT_INIT: if task is not initialized
+  //   - OB_SUCCESS: if task completed (callback_done_ is true)
+  //   - OB_TIMEOUT: if task has exceeded wait_for_completion_timeout_us_ (real timeout)
+  //   - OB_EAGAIN: if wait timed out but task not yet exceeded total timeout,
+  //                caller should check cancel status and retry
+  // Usage: call in a loop, check for cancel between calls, handle OB_TIMEOUT as task failure
+  int wait_for_completion(int64_t check_interval_us);
   int wake_up();
   void disable_callback();
   void set_callback_done();
   bool need_callback() { return cb_handle_ != nullptr ? true : false; };
-
+  void set_need_cancel() { ATOMIC_STORE(&need_cancel_, true); }
+  bool need_cancel() const { return ATOMIC_LOAD(&need_cancel_); }
+  common::ObCurTraceId::TraceId get_trace_id() const { return trace_id_; }
+  void set_always_retry(bool always_retry) { always_retry_ = always_retry; }
 public:
   static const ObString MODEL_URL_NAME;
   static const ObString MODEL_NAME_NAME;
@@ -215,28 +252,35 @@ class ObEmbeddingTask
   static const ObString USER_KEY_NAME;
   static const ObString INPUT_NAME;
   static const ObString DIMENSIONS_NAME;
+  static const ObString REQUEST_TOO_LARGE_ERROR_MSG;
 
   static const int64_t HTTP_REQUEST_TIMEOUT; // 20 seconds
 
   // Reschedule related constants
   static const int64_t MAX_RESCHEDULE_RETRY_CNT;
   static const int64_t RESCHEDULE_RETRY_INTERVAL_US;
+  static const int64_t MAX_NEXT_HANDLE_INTERVAL_US;
 
   // HTTP retry related constants
   static const int64_t MAX_HTTP_RETRY_CNT;
   static const int64_t HTTP_RETRY_BASE_INTERVAL_US;
   static const int64_t HTTP_RETRY_MAX_INTERVAL_US;
   static const int64_t HTTP_RETRY_MULTIPLIER;
+  static const int64_t ALWAYS_RETRY_MIN_INTERVAL_US;
+  static const int64_t MIN_EMBEDDING_MODEL_RPM;
+  static const int64_t EMBEDDING_MODEL_WAIT_RATIO;
 
   // Callback related constants
   static const int64_t CALLBACK_BATCH_SIZE;
+  static const int64_t MAX_INPUT_TOKEN; // Default max token count for each input: 512
 
 private:
+  void init_members();  // Common initialization for all constructors
   void reset();
   bool is_finished() const;  // Internal use only - no lock needed
-  void set_stop();
+  void set_stop(int ret_code);
   int set_phase(ObEmbeddingTaskPhase new_phase);
-  int complete_task(ObEmbeddingTaskPhase new_phase, int result_code, bool finished = true);
+  int complete_task(int result_code, bool finished = true);
   int start_async_work();
   int check_async_progress();
 
@@ -256,14 +300,24 @@ class ObEmbeddingTask
   int parse_embedding_response(const char *response_data, size_t response_size);
 
   // Helper methods for retry logic
-  bool should_retry_http_request(int64_t http_error_code) const;
+  bool should_retry_http_request(int64_t http_error_code, const ObString &http_error_msg) const;
   bool is_batch_size_related_error(int64_t http_error_code) const;
   int64_t calculate_retry_interval() const;
   int adjust_batch_size_for_retry();
+  int adjust_input_token_limit_for_retry();
   void reset_retry_state();
   int map_http_error_to_internal_error(int64_t http_error_code) const;
   void try_increase_batch_size();
   int init_curl_handler(const ObString &model_url, const ObString &user_key, const int64_t http_timeout_us);
+  bool is_request_too_large(int64_t http_error_code, const ObString &content) const;
+  int truncate_text_by_token_count(ObString &text, const ObCollationType cs_type, const int64_t max_token_count) const;
+  bool can_retry_request() const { return http_retry_count_ < http_max_retry_count_; }
+  bool is_ready_to_handle() const; // Check if current time exceeds next_handle_task_time_us_
+  int finish_task();
+  int try_rescheule_task();
+  int calc_max_wait_completion_time_us(int64_t http_timeout_us, int64_t http_max_retry_count,
+                                       int64_t batch_size, int64_t input_chunks_count,
+                                       int64_t &max_wait_completion_time_us) const;
 
   struct HttpResponseData {
     HttpResponseData(ObIAllocator &allocator) : data(nullptr), size(0), allocator(allocator) {}
@@ -365,20 +419,31 @@ class ObEmbeddingTask
   int64_t wait_for_completion_timeout_us_; // For controlling the maximum timeout of waiting for completion
 
   bool need_retry_flag_;
+  bool always_retry_; // If true, task can retry even after exceeding max retry count, but must wait at least 3 minutes
+  int64_t last_exceeded_retry_time_us_; // Time when retry count was exceeded
 
   // Batch size adjustment for retry
   uint32_t original_batch_size_;
   bool batch_size_adjusted_;
   uint32_t current_batch_size_;
   uint32_t successful_requests_count_;
 
+  // Input token limit adjustment for retry (when request too large)
+  int64_t current_input_token_limit_;
+
   ObThreadCond task_cond_;
   bool callback_done_;
 
   // TODO(fanfangyao.ffy): use taskhandle to manage task reference count
   // ref_cnt_ is only used to track the reference count of the post create embedding task
   int64_t ref_cnt_;
-
+  ObCollationType col_type_;
+  bool need_cancel_;
+  common::ObCurTraceId::TraceId trace_id_;
+  int64_t source_task_id_;
+  ObEmbeddingTasSourceType source_task_type_;
+  int64_t next_handle_task_time_us_;
+  ObEmbeddingTaskHandler *thread_pool_;
   private:
     DISALLOW_COPY_AND_ASSIGN(ObEmbeddingTask);
 };
diff --git a/src/share/vector_index/ob_vector_index_util.cpp b/src/share/vector_index/ob_vector_index_util.cpp
@@ -3622,6 +3622,70 @@ int ObVectorIndexUtil::check_index_param(
 
   return ret;
 }
+// index_table_id must be table which has vector column
+int ObVectorIndexUtil::get_index_column_collation_type(
+    const int64_t tenant_id,
+    const uint64_t index_table_id,
+    ObCollationType &col_type)
+{
+  int ret = OB_SUCCESS;
+  const ObTableSchema *data_table_schema = nullptr;
+  const ObTableSchema *table_schema = nullptr;
+  int64_t main_table_id = OB_INVALID_ID;
+  ObArray<uint64_t> tmp_column_ids;
+  col_type = CS_TYPE_INVALID;
+  ObSchemaGetterGuard schema_guard;
+
+  if (!is_valid_tenant_id(tenant_id) || OB_INVALID_ID == index_table_id) {
+    ret = OB_INVALID_ARGUMENT;
+    LOG_WARN("invalid argument", K(ret), K(index_table_id));
+  } else if (OB_FAIL(ObMultiVersionSchemaService::get_instance().get_tenant_schema_guard(tenant_id, schema_guard))) {
+    LOG_WARN("fail to get tenant schema guard", K(ret), K(MTL_ID()));
+  } else if (OB_FAIL(schema_guard.get_table_schema(tenant_id, index_table_id, table_schema))) {
+    LOG_WARN("fail to get table scheam", K(ret), K(tenant_id), K(index_table_id));
+  } else if (OB_ISNULL(table_schema)) {
+    ret = OB_TABLE_NOT_EXIST;
+    LOG_INFO("table not exit", K(ret), K(tenant_id), K(index_table_id));
+  } else if (OB_FALSE_IT(main_table_id = table_schema->get_data_table_id())) {
+  } else if (OB_INVALID_ID == main_table_id) {
+    ret = OB_ERR_UNEXPECTED;
+    LOG_WARN("unexpected invalid id", K(ret), K(main_table_id));
+  } else if (OB_FAIL(schema_guard.get_table_schema(tenant_id, main_table_id, data_table_schema))) {
+    LOG_WARN("fail to get table scheam", K(ret), K(tenant_id), K(index_table_id));
+  } else if (OB_ISNULL(data_table_schema)) {
+    ret = OB_TABLE_NOT_EXIST;
+    LOG_INFO("table not exit", K(ret), K(tenant_id), K(main_table_id));
+  } else if (OB_FAIL(table_schema->get_column_ids(tmp_column_ids))) {
+    LOG_WARN("fail to get index table all column ids", K(ret), K(data_table_schema));
+  } else {
+    for (int64_t i = 0; OB_SUCC(ret) && i < tmp_column_ids.count() && col_type == CS_TYPE_INVALID; ++i) {
+      const ObColumnSchemaV2 *col_schema = data_table_schema->get_column_schema(tmp_column_ids[i]);
+      if (OB_ISNULL(col_schema)) {
+        ret = OB_ERR_UNEXPECTED;
+        LOG_WARN("unexpected null column schema ptr", K(ret));
+      } else if (!col_schema->is_vec_hnsw_vector_column()) {
+        // only need vector column
+      } else {
+        ObArray<uint64_t> cascaded_column_ids;
+        if (OB_FAIL(col_schema->get_cascaded_column_ids(cascaded_column_ids))) {
+          LOG_WARN("failed to get cascaded column ids", K(ret));
+        } else {
+          for (int64_t j = 0; OB_SUCC(ret) && j < cascaded_column_ids.count() && col_type == CS_TYPE_INVALID; ++j) {
+            const ObColumnSchemaV2 *cascaded_column = NULL;
+            if (OB_ISNULL(cascaded_column = data_table_schema->get_column_schema(cascaded_column_ids.at(j)))) {
+              ret = OB_ERR_UNEXPECTED;
+              LOG_WARN("unexpected cascaded column", K(ret));
+            } else {
+              col_type = cascaded_column->get_collation_type();
+              LOG_DEBUG("get vector index collation type", K(col_type));
+            }
+          }
+        }
+      }
+    }
+  }
+  return ret;
+}
 
 int ObVectorIndexUtil::get_vector_index_type(
     sql::ObRawExpr *&raw_expr,
diff --git a/src/share/vector_index/ob_vector_index_util.h b/src/share/vector_index/ob_vector_index_util.h
@@ -832,6 +832,9 @@ class ObVectorIndexUtil final
                                                const ObString &new_idx_params,
                                                const ObTableSchema &index_table_schema,
                                                bool &need_embedding_when_rebuild);
+
+  static int get_index_column_collation_type(const int64_t tenant_id, const uint64_t index_table_id, ObCollationType &col_type);
+
 private:
   static void save_column_schema(
       const ObColumnSchemaV2 *&old_column,
diff --git a/src/storage/ddl/ob_ddl_pipeline.cpp b/src/storage/ddl/ob_ddl_pipeline.cpp
@@ -93,7 +93,8 @@ ObVectorIndexTabletContext::ObVectorIndexTabletContext()
       lob_inrow_threshold_(0), rowkey_cnt_(0), column_cnt_(0), snapshot_version_(0), index_type_(share::VIAT_MAX), helper_(nullptr),
       allocator_("VecIndexCtx", OB_MALLOC_NORMAL_BLOCK_SIZE, MTL_ID()),
       memory_context_(MTL(ObPluginVectorIndexService *)->get_memory_context()),
-      all_vsag_use_mem_(MTL(ObPluginVectorIndexService *)->get_all_vsag_use_mem())
+      all_vsag_use_mem_(MTL(ObPluginVectorIndexService *)->get_all_vsag_use_mem()),
+      table_id_(0)
 {
 
 }
@@ -120,6 +121,8 @@ int ObVectorIndexTabletContext::init(
     rowkey_cnt_ = ddl_table_schema.table_item_.rowkey_column_num_;
     column_cnt_ = ddl_table_schema.column_items_.count();
     snapshot_version_ = snapshot_version;
+    table_id_ = ddl_table_schema.table_id_;
+
     if (schema::is_vec_index_snapshot_data_type(index_type)) {
       if (OB_FAIL(init_hnsw_index(ddl_table_schema))) {
         LOG_WARN("init hnsw index failed", K(ret));
@@ -1661,13 +1664,18 @@ int ObHNSWEmbeddingOperator::init(const ObTabletID &tablet_id)
     ret = OB_ERR_UNEXPECTED;
     LOG_WARN("error unexpected, vector index ctx is null", K(ret));
   } else {
+    const uint64_t table_id = vector_index_ctx->table_id_;
     vec_dim_ = vector_index_ctx->vec_dim_;
     rowkey_cnt_ = vector_index_ctx->rowkey_cnt_;
     text_col_idx_ = vector_index_ctx->vector_chunk_col_idx_;
     extra_column_idxs_.reset();
     ObVectorIndexParam index_param;
+    ObSchemaGetterGuard schema_guard;
+    ObCollationType col_type = CS_TYPE_INVALID;
 
-    if (OB_FAIL(vector_index_ctx->build_extra_column_idxs(static_cast<int32_t>(text_col_idx_), extra_column_idxs_))) {
+    if (OB_FAIL(ObVectorIndexUtil::get_index_column_collation_type(MTL_ID(), table_id, col_type))) {
+      LOG_WARN("fail to get vector column collation type", K(ret), K(text_col_idx_), K(table_id));
+    } else if (OB_FAIL(vector_index_ctx->build_extra_column_idxs(static_cast<int32_t>(text_col_idx_), extra_column_idxs_))) {
       LOG_WARN("build_extra_column_idxs failed", K(ret), K(text_col_idx_));
     } else if (OB_FAIL(ObVectorIndexUtil::parser_params_from_string(vector_index_ctx->vec_idx_param_, ObVectorIndexType::VIT_HNSW_INDEX, index_param, false))) {
       LOG_WARN("failed to parser params from string", K(ret));
@@ -1679,12 +1687,12 @@ int ObHNSWEmbeddingOperator::init(const ObTabletID &tablet_id)
         ret = OB_ALLOCATE_MEMORY_FAILED;
         LOG_WARN("failed to alloc ObEmbeddingTaskMgr", K(ret));
       } else {
-        embedmgr_ = new (buf) ObEmbeddingTaskMgr();
+        embedmgr_ = new (buf) ObEmbeddingTaskMgr(*this);
       }
     }
 
     if (OB_SUCC(ret)) {
-      if (OB_FAIL(embedmgr_->init(model_id_))) {
+      if (OB_FAIL(embedmgr_->init(model_id_, col_type))) {
         embedmgr_->~ObEmbeddingTaskMgr();
         op_allocator_.free(embedmgr_);
         embedmgr_ = nullptr;
diff --git a/src/storage/ddl/ob_ddl_pipeline.h b/src/storage/ddl/ob_ddl_pipeline.h
@@ -169,6 +169,7 @@ TO_STRING_KV(K_(tenant_id), K_(ls_id), K_(tablet_id), K_(snapshot_version), K_(i
   common::ObArenaAllocator allocator_;
   lib::MemoryContext &memory_context_;
   uint64_t *all_vsag_use_mem_;
+  uint64_t table_id_;
 };
 
 class ObVectorIndexRowIterator
diff --git a/src/storage/ddl/ob_hnsw_embedmgr.cpp b/src/storage/ddl/ob_hnsw_embedmgr.cpp
diff --git a/src/storage/ddl/ob_hnsw_embedmgr.h b/src/storage/ddl/ob_hnsw_embedmgr.h
