perf: optimize full-text index build performance

src/plugin/interface/ob_plugin_ftparser_intf.h

@@ -112,13 +112,17 @@ class ObFTParserParam final : public ObFTParserParamExport
   {
     ObFTParserParamExport::reset();
     allocator_ = nullptr;
+    metadata_allocator_ = nullptr;
+    scratch_allocator_ = nullptr;
     ngram_token_size_ = NGRAM_TOKEN_SIZE;
   }
 
   INHERIT_TO_STRING_KV("base", ObFTParserParamExport, KP_(allocator), K_(ngram_token_size));
 
 public:
   common::ObIAllocator *allocator_ = nullptr;
+  common::ObIAllocator *metadata_allocator_ = nullptr;
+  common::ObIAllocator *scratch_allocator_ = nullptr;
 
   // ik parser params
   ObFTIKParam ik_param_;
@@ -137,6 +141,11 @@ class ObITokenIterator
       int64_t &word_len,
       int64_t &char_cnt,
       int64_t &word_freq) = 0;
+  virtual int reuse_parser(const char *fulltext, const int64_t fulltext_len)
+  {
+    UNUSEDx(fulltext, fulltext_len);
+    return OB_NOT_SUPPORTED;
+  }
   DECLARE_PURE_VIRTUAL_TO_STRING;
 };
 

src/rootserver/ob_ddl_operator.cpp

@@ -4227,6 +4227,13 @@ int ObDDLOperator::drop_table(
                       ObObjectType::VIEW))) {
     LOG_WARN("failed to delete_schema_object_dependency", K(ret), K(1UL),
     K(table_schema.get_table_id()));
+  } else if (table_schema.is_index_table()
+            && OB_FAIL(ObDependencyInfo::delete_schema_object_dependency(
+                      trans,
+                      table_schema.get_table_id(),
+                      table_schema.get_schema_version(),
+                      ObObjectType::INDEX))) {
+    LOG_WARN("failed to delete index dependency", K(ret), K(table_schema.get_table_id()));
   }
 
   if (OB_FAIL(ret)) {

src/rootserver/ob_ddl_service.cpp

@@ -65,6 +65,7 @@
 #include "rootserver/parallel_ddl/ob_ddl_helper.h"
 #include "storage/ddl/ob_ddl_alter_auto_part_attr.h"
 #include "sql/resolver/ddl/ob_vec_index_builder_util.h"
+#include "sql/resolver/ddl/ob_fts_index_builder_util.h"
 #include "rootserver/direct_load/ob_direct_load_partition_exchange.h"
 #include "rootserver/truncate_info/ob_truncate_info_service.h"
 #include "share/truncate_info/ob_truncate_info_util.h"
@@ -2281,6 +2282,9 @@ int ObDDLService::create_index_or_mlog_table_in_trans(
     } else if (OB_ISNULL(sql_trans)
         && OB_FAIL(trans.start(sql_proxy_, refreshed_schema_version))) {
       LOG_WARN("failed to start trans", KR(ret), K(refreshed_schema_version));
+    } else if (OB_FAIL(ObFtsIndexBuilderUtil::try_load_and_lock_dictionary_tables(
+                           table_schema, trans))) {
+      LOG_WARN("fail to load fulltext dictionaries", KR(ret), K(table_schema));
     } else if (OB_FAIL(ddl_operator.create_table(table_schema,
                                                  trans,
                                                  ddl_stmt_str))) {
@@ -13931,7 +13935,7 @@ int ObDDLService::alter_table_in_trans(obcall::ObAlterTableArg &alter_table_arg,
                     need_update_index_table))) {
           LOG_WARN("failed to set new table options", K(ret), K(new_table_schema),
           K(*orig_table_schema), K(ret));
-        } else {
+        } else if (!orig_table_schema->is_fulltext_dict()) {
           new_table_schema.set_table_flags(alter_table_schema.get_table_flags());
         }
       }

src/rootserver/parallel_ddl/ob_drop_table_helper.cpp

@@ -1382,6 +1382,21 @@ int ObDropTableHelper::drop_table_(const ObTableSchema &table_schema, const ObSt
   int64_t new_schema_version = OB_INVALID_VERSION;
   if (OB_FAIL(check_inner_stat_())) {
     LOG_WARN("fail to check inner stat", KR(ret));
+  } else if (table_schema.is_fulltext_dict()) {
+    ObArray<ObDependencyInfo> dependencies;
+    if (OB_FAIL(ObDependencyInfo::collect_dep_infos(
+                    table_schema.get_table_id(), *sql_proxy_, dependencies))) {
+      LOG_WARN("fail to collect dictionary dependencies", K(ret), K(table_schema));
+    } else {
+      for (int64_t i = 0; OB_SUCC(ret) && i < dependencies.count(); ++i) {
+        if (ObObjectType::INDEX == dependencies.at(i).get_dep_obj_type()) {
+          ret = OB_OP_NOT_ALLOW;
+          LOG_USER_ERROR(OB_OP_NOT_ALLOW, "dropping a referenced FULLTEXT_DICT table is");
+        }
+      }
+    }
+  }
+  if (OB_FAIL(ret)) {
   } else if (OB_ISNULL(schema_service_impl = schema_service_->get_schema_service())) {
     ret = OB_ERR_UNEXPECTED;
     LOG_WARN("schema service impl is null", KR(ret));

src/share/schema/ob_schema_struct.h

@@ -184,6 +184,7 @@ static const uint64_t OB_MIN_ID  = 0;//used for lower_bound
 #define EXTERNAL_TABLE_AUTO_REFRESH_INTERVAL_FLAG (INT64_C(1) << 3)
 #define EXTERNAL_TABLE_AUTO_REFRESH_FLAG_OFFSET 2
 #define EXTERNAL_TABLE_AUTO_REFRESH_FLAG_BITS 2
+#define FULLTEXT_DICT_FLAG (INT64_C(1) << 5)
 
 // schema array size
 static const int64_t SCHEMA_SMALL_MALLOC_BLOCK_SIZE = 64;

src/share/schema/ob_table_schema.cpp

@@ -2756,6 +2756,42 @@ int ObTableSchema::set_view_definition(const common::ObString &view_definition)
   return view_schema_.set_view_definition(view_definition);
 }
 
+int ObTableSchema::check_fulltext_dict_structure() const
+{
+  int ret = OB_SUCCESS;
+  const ObColumnSchemaV2 *word_column = nullptr;
+  if (!is_fulltext_dict()) {
+    ret = OB_INVALID_ARGUMENT;
+  } else if (!is_index_organized_table()) {
+    ret = OB_INVALID_ARGUMENT;
+    LOG_USER_ERROR(OB_INVALID_ARGUMENT, "FULLTEXT_DICT table must use ORGANIZATION INDEX");
+  } else if (CHARSET_UTF8MB4 != get_charset_type()) {
+    ret = OB_INVALID_ARGUMENT;
+    LOG_USER_ERROR(OB_INVALID_ARGUMENT, "FULLTEXT_DICT table charset must be utf8mb4");
+  } else if (1 != get_column_count()
+             || OB_ISNULL(word_column = get_column_schema_by_idx(0))) {
+    ret = OB_INVALID_ARGUMENT;
+    LOG_USER_ERROR(OB_INVALID_ARGUMENT, "FULLTEXT_DICT table must contain only the word column");
+  } else if (0 != word_column->get_column_name_str().case_compare("word")) {
+    ret = OB_INVALID_ARGUMENT;
+    LOG_USER_ERROR(OB_INVALID_ARGUMENT, "FULLTEXT_DICT table column must be named word");
+  } else if (ObVarcharType != word_column->get_data_type()) {
+    ret = OB_INVALID_ARGUMENT;
+    LOG_USER_ERROR(OB_INVALID_ARGUMENT, "FULLTEXT_DICT word column must be VARCHAR");
+  } else if (word_column->get_data_length() < 1 || word_column->get_data_length() > 500) {
+    ret = OB_INVALID_ARGUMENT;
+    LOG_USER_ERROR(OB_INVALID_ARGUMENT, "FULLTEXT_DICT word column length must be between 1 and 500");
+  } else if (CHARSET_UTF8MB4 != word_column->get_charset_type()) {
+    ret = OB_INVALID_ARGUMENT;
+    LOG_USER_ERROR(OB_INVALID_ARGUMENT, "FULLTEXT_DICT word column charset must be utf8mb4");
+  } else if (!word_column->is_rowkey_column() || 1 != word_column->get_rowkey_position()
+             || 1 != get_rowkey_column_num()) {
+    ret = OB_INVALID_ARGUMENT;
+    LOG_USER_ERROR(OB_INVALID_ARGUMENT, "FULLTEXT_DICT word column must be the primary key");
+  }
+  return ret;
+}
+
 int ObTableSchema::set_parser_name_and_properties(
     const common::ObString &parser_name,
     const common::ObString &parser_properties)

src/share/schema/ob_table_schema.h

@@ -1413,6 +1413,9 @@ class ObTableSchema : public ObSimpleTableSchemaV2
   int set_external_properties(const common::ObString &format) { return deep_copy_str(format, external_properties_); }
   void set_external_table_auto_refresh(const int64_t flag) { table_flags_ |= (flag << EXTERNAL_TABLE_AUTO_REFRESH_FLAG_OFFSET); }
   inline void set_user_specified_partition_for_external_table() { table_flags_ |= EXTERNAL_TABLE_USER_SPECIFIED_PARTITION_FLAG; }
+  inline void set_fulltext_dict() { table_flags_ |= FULLTEXT_DICT_FLAG; }
+  inline bool is_fulltext_dict() const { return (table_flags_ & FULLTEXT_DICT_FLAG) != 0; }
+  int check_fulltext_dict_structure() const;
   template<typename ColumnType>
   int add_column(const ColumnType &column);
   int delete_column(const common::ObString &column_name);

src/sql/das/ob_das_domain_utils.cpp

@@ -41,8 +41,25 @@ ObObjDatumMapType ObFTIndexRowCache::FTS_DOC_WORD_TYPES[] = {OBJ_DATUM_STRING, O
 ObExprOperatorType ObFTIndexRowCache::FTS_INDEX_EXPR_TYPE[] = {T_FUN_SYS_WORD_SEGMENT, T_FUN_SYS_DOC_ID, T_FUN_SYS_WORD_COUNT, T_FUN_SYS_DOC_LENGTH};
 ObExprOperatorType ObFTIndexRowCache::FTS_DOC_WORD_EXPR_TYPE[] = {T_FUN_SYS_DOC_ID, T_FUN_SYS_WORD_SEGMENT, T_FUN_SYS_WORD_COUNT, T_FUN_SYS_DOC_LENGTH};
 
+static constexpr int64_t FT_WORD_DOC_COL_CNT = 4;
+static constexpr int64_t FT_MIN_WORD_BUCKET_COUNT = 2;
+static constexpr int64_t FT_MAX_WORD_BUCKET_COUNT = 997;
+static constexpr int64_t FT_ESTIMATED_BYTES_PER_TOKEN = 10;
+
+static int64_t estimate_ft_word_bucket_count(const int64_t fulltext_length)
+{
+  return MIN(MAX(fulltext_length / FT_ESTIMATED_BYTES_PER_TOKEN,
+                 FT_MIN_WORD_BUCKET_COUNT),
+             FT_MAX_WORD_BUCKET_COUNT);
+}
+
 ObFTIndexRowCache::ObFTIndexRowCache()
-  : rows_(),
+  : words_(),
+    word_iter_(),
+    word_end_iter_(),
+    row_(),
+    doc_id_datum_(),
+    doc_length_(0),
     row_idx_(0),
     is_fts_index_aux_(true),
     helper_(),
@@ -70,6 +87,8 @@ int ObFTIndexRowCache::init(
     LOG_WARN("failed to create merge memctx", K(ret));
   } else if (OB_FAIL(helper_.init(&(merge_memctx_->get_arena_allocator()), parser_name, parser_properties))) {
     LOG_WARN("fail to init full-text parser helper", K(ret));
+  } else if (OB_FAIL(row_.init(FT_WORD_DOC_COL_CNT))) {
+    LOG_WARN("fail to init streaming fulltext index row", K(ret), K(FT_WORD_DOC_COL_CNT));
   } else {
     row_idx_ = 0;
     is_fts_index_aux_ = is_fts_index_aux;
@@ -86,58 +105,87 @@ int ObFTIndexRowCache::segment(const common::ObObjMeta &ft_obj_meta,
                                const ObString &fulltext)
 {
   int ret = OB_SUCCESS;
+  const int64_t ft_word_bkt_cnt = estimate_ft_word_bucket_count(fulltext.length());
   if (OB_UNLIKELY(!is_inited_)) {
     ret = OB_NOT_INIT;
     LOG_WARN("ObFTIndexRowCache hasn't be initialized", K(ret), K(is_inited_));
   } else if (FALSE_IT(reuse())) {
-  } else if (OB_FAIL(ObDASDomainUtils::generate_fulltext_word_rows(merge_memctx_->get_arena_allocator(),
-                                                                   &helper_,
-                                                                   ft_obj_meta,
-                                                                   doc_id_datum,
-                                                                   fulltext,
-                                                                   is_fts_index_aux_,
-                                                                   rows_))) {
-    LOG_WARN("fail to generate fulltext word rows", K(ret), K(helper_), K(is_fts_index_aux_));
+  } else if (0 == fulltext.length()) {
+    ret = OB_ITER_END;
+  } else if (OB_FAIL(words_.create(ft_word_bkt_cnt, common::ObMemAttr("FTWordMap")))) {
+    LOG_WARN("fail to create fulltext word map", K(ret), K(ft_word_bkt_cnt));
+  } else if (OB_FAIL(helper_.segment(ft_obj_meta,
+                                    fulltext.ptr(),
+                                    fulltext.length(),
+                                    doc_length_,
+                                    words_))) {
+    LOG_WARN("fail to segment fulltext", K(ret), K(helper_), K(ft_obj_meta), K(fulltext));
+  } else if (0 == words_.size()) {
+    ret = OB_ITER_END;
   } else {
+    doc_id_datum_ = doc_id_datum;
+    word_iter_ = words_.begin();
+    word_end_iter_ = words_.end();
     row_idx_ = 0;
   }
-  LOG_TRACE("word segment", K(ret), K(row_idx_), K(rows_.count()), K(doc_id_datum), K(fulltext));
+  LOG_TRACE("word segment", K(ret), K(row_idx_), K(words_.size()), K(doc_id_datum), K(fulltext));
   return ret;
 }
 
 int ObFTIndexRowCache::get_next_row(blocksstable::ObDatumRow *&row)
 {
   int ret = OB_SUCCESS;
+  row = nullptr;
   if (OB_UNLIKELY(!is_inited_)) {
     ret = OB_NOT_INIT;
     LOG_WARN("ObFTIndexRowCache hasn't be initialized", K(ret), K(is_inited_));
-  } else if (row_idx_ >= rows_.count()) {
+  } else if (!words_.created() || word_iter_ == word_end_iter_) {
     ret = OB_ITER_END;
   } else {
-    row = (rows_[row_idx_]);
+    const storage::ObFTWord &ft_word = word_iter_->first;
+    const int64_t word_cnt = word_iter_->second;
+    const int64_t word_idx = is_fts_index_aux_ ? 0 : 1;
+    const int64_t doc_id_idx = is_fts_index_aux_ ? 1 : 0;
+    row_.reuse();
+    row_.storage_datums_[word_idx].set_datum(ft_word.get_word());
+    row_.storage_datums_[doc_id_idx].shallow_copy_from_datum(doc_id_datum_);
+    row_.storage_datums_[2].set_uint(word_cnt);
+    row_.storage_datums_[3].set_uint(doc_length_);
+    row = &row_;
+    ++word_iter_;
     ++row_idx_;
   }
-  LOG_TRACE("get next row", K(ret), KPC(row), K(row_idx_), K(rows_.count()));
+  LOG_TRACE("get next row", K(ret), KPC(row), K(row_idx_), K(words_.size()));
   return ret;
 }
 
 void ObFTIndexRowCache::reset()
 {
-  rows_.reset();
-  row_idx_ = 0;
-  is_fts_index_aux_ = true;
+  reuse();
+  row_.reset();
   helper_.reset();
   if (OB_NOT_NULL(merge_memctx_)) {
     DESTROY_CONTEXT(merge_memctx_);
     merge_memctx_ = nullptr;
   }
+  is_fts_index_aux_ = true;
   is_inited_ = false;
 }
 
 void ObFTIndexRowCache::reuse()
 {
-  rows_.reuse();
+  if (words_.created()) {
+    const int tmp_ret = words_.destroy();
+    if (OB_SUCCESS != tmp_ret) {
+      LOG_WARN_RET(tmp_ret, "fail to destroy fulltext word map");
+    }
+  }
+  word_iter_ = storage::ObFTWordMap::const_iterator();
+  word_end_iter_ = storage::ObFTWordMap::const_iterator();
+  doc_id_datum_.reset();
+  doc_length_ = 0;
   row_idx_ = 0;
+  row_.reuse();
   if (OB_NOT_NULL(merge_memctx_)) {
     merge_memctx_->reset_remain_one_page();
   }

src/sql/das/ob_das_domain_utils.h

@@ -48,10 +48,16 @@ class ObFTIndexRowCache final
   int get_next_row(blocksstable::ObDatumRow *&row);
   void reset();
   void reuse();
-  TO_STRING_KV(K_(row_idx), K_(is_fts_index_aux), K_(helper), K_(is_inited), K_(rows));
+  TO_STRING_KV(K_(row_idx), K_(is_fts_index_aux), K_(helper), K_(is_inited),
+      K_(doc_length), "word_count", words_.size());
 private:
   lib::MemoryContext merge_memctx_;
-  ObDomainIndexRow rows_;
+  storage::ObFTWordMap words_;
+  storage::ObFTWordMap::const_iterator word_iter_;
+  storage::ObFTWordMap::const_iterator word_end_iter_;
+  blocksstable::ObDatumRow row_;
+  common::ObDatum doc_id_datum_;
+  int64_t doc_length_;
   uint64_t row_idx_;
   bool is_fts_index_aux_;
   storage::ObFTParseHelper helper_;

src/sql/engine/cmd/ob_alter_system_executor.cpp

@@ -35,6 +35,9 @@
 #include "sql/engine/cmd/ob_timezone_importer.h"
 #include "sql/engine/cmd/ob_srs_importer.h"
 #include "share/ob_internal_table_change_notifier.h"
+#include "share/schema/ob_schema_getter_guard.h"
+#include "storage/fts/ob_fts_plugin_helper.h"
+#include "storage/fts/dict/ob_ft_dict_hub.h"
 
 namespace oceanbase
 {
@@ -552,6 +555,46 @@ int ObClearMergeErrorExecutor::execute(ObExecContext &ctx, ObClearMergeErrorStmt
   return ret;
 }
 
+int ObRefreshFulltextDictExecutor::execute(ObExecContext &ctx,
+                                           ObRefreshFulltextDictStmt &stmt)
+{
+  int ret = OB_SUCCESS;
+  share::schema::ObSchemaGetterGuard schema_guard;
+  const share::schema::ObDatabaseSchema *database_schema = nullptr;
+  const share::schema::ObTableSchema *table_schema = nullptr;
+  storage::ObFTDictHub *dict_hub = nullptr;
+  if (OB_ISNULL(GCTX.schema_service_)) {
+    ret = OB_ERR_UNEXPECTED;
+  } else if (OB_FAIL(GCTX.schema_service_->get_tenant_schema_guard(schema_guard))) {
+    LOG_WARN("fail to get schema guard", K(ret));
+  } else if (OB_FAIL(schema_guard.get_database_schema(stmt.get_database_name(), database_schema))) {
+    LOG_WARN("fail to get dictionary database", K(ret), K(stmt));
+  } else if (OB_ISNULL(database_schema)) {
+    ret = OB_ERR_BAD_DATABASE;
+  } else if (OB_FAIL(schema_guard.get_table_schema(database_schema->get_database_id(),
+                                                   stmt.get_table_name(),
+                                                   false,
+                                                   table_schema))) {
+    LOG_WARN("fail to get dictionary table", K(ret), K(stmt));
+  } else if (OB_ISNULL(table_schema)) {
+    ret = OB_TABLE_NOT_EXIST;
+  } else if (!table_schema->is_fulltext_dict()) {
+    ret = OB_INVALID_ARGUMENT;
+    LOG_USER_ERROR(OB_INVALID_ARGUMENT, "table is not a FULLTEXT_DICT table");
+  } else if (OB_FAIL(table_schema->check_fulltext_dict_structure())) {
+    LOG_WARN("invalid fulltext dictionary structure", K(ret), KPC(table_schema));
+  } else if (OB_FAIL(storage::ObFTParsePluginData::instance().get_dict_hub(dict_hub))) {
+    LOG_WARN("fail to get fulltext dictionary hub", K(ret));
+  } else if (OB_ISNULL(dict_hub)) {
+    ret = OB_ERR_UNEXPECTED;
+  } else if (OB_FAIL(dict_hub->refresh_user_dict(stmt.get_database_name(),
+                                                stmt.get_table_name(),
+                                                table_schema->get_table_id()))) {
+    LOG_WARN("fail to refresh fulltext dictionary", K(ret), K(stmt));
+  }
+  return ret;
+}
+
 int ObUpgradeVirtualSchemaExecutor ::execute(
 		ObExecContext &ctx, ObUpgradeVirtualSchemaStmt &stmt)
 {

src/sql/engine/cmd/ob_alter_system_executor.h

@@ -64,6 +64,8 @@ DEF_SIMPLE_EXECUTOR(ObSetConfig);
 
 DEF_SIMPLE_EXECUTOR(ObClearMergeError);
 
+DEF_SIMPLE_EXECUTOR(ObRefreshFulltextDict);
+
 DEF_SIMPLE_EXECUTOR(ObUpgradeVirtualSchema);