Merge branch 'main' into feat/zstd-level-config

Author: zjw1111

src/paimon/global_index/lucene/jieba_analyzer.cpp

@@ -82,6 +82,11 @@ void JiebaTokenizer::CutWithMode(const std::string& tokenize_mode, const cppjieb
     }
 }
 
+bool JiebaTokenizer::IsWhitespaceOnly(const std::string& term) {
+    return term.empty() ||
+           std::all_of(term.begin(), term.end(), [](unsigned char c) { return std::isspace(c); });
+}
+
 void JiebaTokenizer::Normalize(const std::unordered_set<std::string>& stop_words,
                                std::vector<std::string>* input_ptr,
                                std::vector<std::string_view>* output_ptr) {
@@ -90,11 +95,13 @@ void JiebaTokenizer::Normalize(const std::unordered_set<std::string>& stop_words
     output.clear();
     output.reserve(input.size());
     for (auto& term : input) {
+        if (IsWhitespaceOnly(term)) {
+            continue;
+        }
         // remove stop words
         if (stop_words.find(term) != stop_words.end()) {
             continue;
         }
-
         // to lower case
         bool is_alphanumeric = true;
         for (const auto& c : term) {

src/paimon/global_index/lucene/jieba_analyzer.h

@@ -60,6 +60,8 @@ class JiebaTokenizer : public Lucene::Tokenizer {
  private:
     void InnerReset();
 
+    static bool IsWhitespaceOnly(const std::string& term);
+
  private:
     JiebaTokenizerContext context_;
     size_t term_index_ = 0;

src/paimon/global_index/lucene/jieba_analyzer_test.cpp

@@ -90,7 +90,7 @@ TEST_P(JiebaAnalyzerTest, TestWithPosition) {
 TEST_P(JiebaAnalyzerTest, TestNormalize) {
     auto tokenizer = CreateJiebaTokenizer(
         /*with_position=*/false,
-        L"由于购买了Iphone14，我越来越热爱网上学习了！Happy work, happy day!");
+        L"由于购买了Iphone14，我越来越热爱网上学习了！Happy work, happy day! \n\t");
 
     auto term_att = tokenizer->addAttribute<Lucene::TermAttribute>();
 

src/paimon/global_index/lucene/lucene_defs.h

@@ -23,7 +23,6 @@ namespace paimon::lucene {
 static inline const int32_t kVersion = 1;
 static inline const char kIdentifier[] = "lucene-fts";
 static inline const wchar_t kEmptyWstring[] = L"";
-static inline const wchar_t kRowIdFieldWstring[] = L"_ROW_ID";
 
 static inline const char kOptionKeyPrefix[] = "lucene-fts.";
 

src/paimon/global_index/lucene/lucene_global_index_reader.cpp

@@ -179,13 +179,7 @@ Result<std::shared_ptr<GlobalIndexResult>> LuceneGlobalIndexReader::SearchWithLi
     // prepare BitmapScoredGlobalIndexResult
     std::map<int64_t, float> id_to_score;
     for (auto score_doc : results->scoreDocs) {
-        Lucene::DocumentPtr result_doc = searcher_->doc(score_doc->doc);
-        std::string row_id_str = LuceneUtils::WstringToString(result_doc->get(kRowIdFieldWstring));
-        std::optional<int32_t> row_id = StringUtils::StringToValue<int32_t>(row_id_str);
-        if (!row_id) {
-            return Status::Invalid(fmt::format("parse row id str {} to int failed", row_id_str));
-        }
-        id_to_score[static_cast<int64_t>(row_id.value())] = static_cast<float>(score_doc->score);
+        id_to_score[static_cast<int64_t>(score_doc->doc)] = static_cast<float>(score_doc->score);
     }
     RoaringBitmap64 bitmap;
     std::vector<float> scores;

src/paimon/global_index/lucene/lucene_global_index_writer.cpp

@@ -40,13 +40,12 @@ namespace paimon::lucene {
 LuceneGlobalIndexWriter::LuceneWriteContext::LuceneWriteContext(
     const std::string& _tmp_index_path, const Lucene::FSDirectoryPtr& _lucene_dir,
     const Lucene::IndexWriterPtr& _index_writer, const Lucene::DocumentPtr& _doc,
-    const Lucene::FieldPtr& _field, const Lucene::FieldPtr& _row_id_field)
+    const Lucene::FieldPtr& _field)
     : tmp_index_path(_tmp_index_path),
       lucene_dir(_lucene_dir),
       index_writer(_index_writer),
       doc(_doc),
-      field(_field),
-      row_id_field(_row_id_field) {}
+      field(_field) {}
 
 Result<std::shared_ptr<LuceneGlobalIndexWriter>> LuceneGlobalIndexWriter::Create(
     const std::string& field_name, const std::shared_ptr<arrow::DataType>& arrow_type,
@@ -89,17 +88,11 @@ Result<std::shared_ptr<LuceneGlobalIndexWriter>> LuceneGlobalIndexWriter::Create
         auto field = Lucene::newLucene<Lucene::Field>(LuceneUtils::StringToWstring(field_name),
                                                       kEmptyWstring, Lucene::Field::STORE_NO,
                                                       Lucene::Field::INDEX_ANALYZED_NO_NORMS);
-        auto row_id_field = Lucene::newLucene<Lucene::Field>(
-            kRowIdFieldWstring, kEmptyWstring, Lucene::Field::STORE_YES,
-            Lucene::Field::INDEX_NOT_ANALYZED_NO_NORMS);
         field->setOmitTermFreqAndPositions(omit_term_freq_and_positions);
-        row_id_field->setOmitTermFreqAndPositions(true);
         doc->add(field);
-        doc->add(row_id_field);
         return std::shared_ptr<LuceneGlobalIndexWriter>(new LuceneGlobalIndexWriter(
-            field_name, arrow_type,
-            LuceneWriteContext(tmp_path, lucene_dir, writer, doc, field, row_id_field), file_writer,
-            options, pool));
+            field_name, arrow_type, LuceneWriteContext(tmp_path, lucene_dir, writer, doc, field),
+            file_writer, options, pool));
     } catch (const std::exception& e) {
         return Status::Invalid(
             fmt::format("create lucene global index writer failed, with {} error.", e.what()));
@@ -153,8 +146,7 @@ Status LuceneGlobalIndexWriter::AddBatch(::ArrowArray* arrow_array) {
                 auto view = string_array->Value(i);
                 write_context_.field->setValue(LuceneUtils::StringToWstring(view));
             }
-            write_context_.row_id_field->setValue(
-                LuceneUtils::StringToWstring(std::to_string(row_id_++)));
+            row_id_++;
             write_context_.index_writer->addDocument(write_context_.doc);
         }
     } catch (const std::exception& e) {
@@ -170,6 +162,11 @@ Status LuceneGlobalIndexWriter::AddBatch(::ArrowArray* arrow_array) {
 Result<std::string> LuceneGlobalIndexWriter::FlushIndexToFinal() {
     try {
         // flush index to tmp dir
+        if (write_context_.index_writer->numDocs() != row_id_) {
+            return Status::Invalid(
+                fmt::format("lucene writer row count {} mismatch paimon inner row count {}",
+                            write_context_.index_writer->numDocs(), row_id_));
+        }
         write_context_.index_writer->optimize();
         write_context_.index_writer->close();
 

src/paimon/global_index/lucene/lucene_global_index_writer.h

@@ -29,8 +29,7 @@ class LuceneGlobalIndexWriter : public GlobalIndexWriter {
         LuceneWriteContext(const std::string& _tmp_index_path,
                            const Lucene::FSDirectoryPtr& _lucene_dir,
                            const Lucene::IndexWriterPtr& _index_writer,
-                           const Lucene::DocumentPtr& _doc, const Lucene::FieldPtr& _field,
-                           const Lucene::FieldPtr& _row_id_field);
+                           const Lucene::DocumentPtr& _doc, const Lucene::FieldPtr& _field);
 
         LuceneWriteContext(LuceneWriteContext&&) = default;
         LuceneWriteContext& operator=(LuceneWriteContext&&) = default;
@@ -40,7 +39,6 @@ class LuceneGlobalIndexWriter : public GlobalIndexWriter {
         Lucene::IndexWriterPtr index_writer;
         Lucene::DocumentPtr doc;
         Lucene::FieldPtr field;
-        Lucene::FieldPtr row_id_field;
     };
 
     static Result<std::shared_ptr<LuceneGlobalIndexWriter>> Create(