revert commit 4c0ae9138d1daccad6fa89deee802cce6645b701 and commit 2e865c128a3072550384423840df795b6f79dd2b

Co-authored-by: shenyunlong.syl <ylshen0919@gmail.com>

Author: 2 people

src/share/ob_debug_sync_point.h

@@ -757,8 +757,6 @@ 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);

src/share/vector_index/ob_hybrid_vector_refresh_task.cpp

@@ -508,7 +508,6 @@ 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;
@@ -529,8 +528,6 @@ 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)) {
@@ -659,7 +656,6 @@ 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));
@@ -668,8 +664,7 @@ 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, col_type, dim, http_timeout_us,
-                             http_max_retries, ctx_->task_status_.task_id_, ObEmbeddingTasSourceType::ASYNC_INDEX))) {
+                             endpoint->get_provider(), access_key, chunk_array, dim, http_timeout_us, http_max_retries))) {
             LOG_WARN("failed to init embedding task", K(ret), KPC(endpoint));
           } else {
             ObEmbeddingTaskHandler *embedding_handler = nullptr;

src/share/vector_index/ob_vector_embedding_handler.cpp

@@ -155,12 +155,6 @@ class ObEmbeddingTaskPhaseManager {
   }
 };
 
-enum class ObEmbeddingTasSourceType
-{
-  INDEX_PIPELINE = 0,
-  ASYNC_INDEX = 1,
-};
-
 class ObEmbeddingTaskHandler;
 
 // Constants for field lengths
@@ -175,14 +169,10 @@ 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,
-          int64_t source_task_id,
-          ObEmbeddingTasSourceType source_task_type,
-          storage::ObEmbeddingIOCallbackHandle *cb_handle = nullptr,
-          bool always_retry = false);
+          storage::ObEmbeddingIOCallbackHandle *cb_handle = nullptr);
   template <typename ThreadPoolType>
   int do_work(ThreadPoolType *thread_pool);
   int64_t get_task_id() const { return task_id_; }
@@ -191,56 +181,29 @@ 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_(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));
+                K_(process_callback_offset));
   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();
-  // 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 wait_for_completion();
   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;
@@ -252,35 +215,28 @@ 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(int ret_code);
+  void set_stop();
   int set_phase(ObEmbeddingTaskPhase new_phase);
-  int complete_task(int result_code, bool finished = true);
+  int complete_task(ObEmbeddingTaskPhase new_phase, int result_code, bool finished = true);
   int start_async_work();
   int check_async_progress();
 
@@ -300,24 +256,14 @@ 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 ObString &http_error_msg) const;
+  bool should_retry_http_request(int64_t http_error_code) 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) {}
@@ -419,31 +365,20 @@ 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);
 };

src/share/vector_index/ob_vector_embedding_handler.h

@@ -3622,70 +3622,6 @@ 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,

src/share/vector_index/ob_vector_index_util.cpp

@@ -832,9 +832,6 @@ 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,

src/share/vector_index/ob_vector_index_util.h

@@ -93,8 +93,7 @@ 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()),
-      table_id_(0)
+      all_vsag_use_mem_(MTL(ObPluginVectorIndexService *)->get_all_vsag_use_mem())
 {
 
 }
@@ -121,8 +120,6 @@ 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));
@@ -1664,18 +1661,13 @@ 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(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_))) {
+    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));
@@ -1687,12 +1679,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(*this);
+        embedmgr_ = new (buf) ObEmbeddingTaskMgr();
       }
     }
 
     if (OB_SUCC(ret)) {
-      if (OB_FAIL(embedmgr_->init(model_id_, col_type))) {
+      if (OB_FAIL(embedmgr_->init(model_id_))) {
         embedmgr_->~ObEmbeddingTaskMgr();
         op_allocator_.free(embedmgr_);
         embedmgr_ = nullptr;

src/storage/ddl/ob_ddl_pipeline.cpp

@@ -169,7 +169,6 @@ 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