support phrase query with block max (#1189)

### What problem does this PR solve?
* support phrase query with block max
* fix skiplist reader error in reading position

Issue link:#639

### Type of change

- [x] Bug Fix (non-breaking change which fixes an issue)
- [x] New Feature
- [x] Test cases

Author: Ma-cat

benchmark/local_infinity/fulltext/fulltext_benchmark.cpp

@@ -222,8 +222,8 @@ void BenchmarkOptimize(SharedPtr<Infinity> infinity, const String &db_name, cons
 
 void BenchmarkQuery(SharedPtr<Infinity> infinity, const String &db_name, const String &table_name) {
     std::string fields = "text";
-    //std::vector<std::string> query_vec = {"one of", "is", "a", "\"is a\"", "\"one of\""};// {"Animalia", "Algorithms", "Animalia Algorithms", "network space", "harmful chemical anarchism"};
-    std::vector<std::string> query_vec = {"harmful chemical anarchism", "\"harmful chemical\"", "\"one of\"", "harmful chemical"};
+    std::vector<std::string> query_vec = {"harmful \"social custom\"", "social custom \"harmful chemical\"", "\"annual American awards\"", "harmful chemical", "\"one of\""};
+
     for (auto match_text : query_vec) {
         BaseProfiler profiler;
         profiler.Begin();
@@ -314,7 +314,7 @@ int main(int argc, char *argv[]) {
     };
     Mode mode(Mode::kInsert);
     SizeT insert_batch = 500;
-    app.add_option("--mode", mode, "Bencmark mode, one of insert, import, merge, query")
+    app.add_option("--mode", mode, "Benchmark mode, one of insert, import, merge, query")
         ->required()
         ->transform(CLI::CheckedTransformer(mode_map, CLI::ignore_case));
     app.add_option("--insert-batch", insert_batch, "batch size of each insert, valid only at insert and merge mode, default value 500");
@@ -331,7 +331,7 @@ int main(int argc, char *argv[]) {
     String srcfile = test_data_path();
     srcfile += "/benchmark/dbpedia-entity/corpus.jsonl";
 
-// #define DEL_LOCAL_DATA
+//#define DEL_LOCAL_DATA
 #ifdef DEL_LOCAL_DATA
     system("rm -rf /var/infinity/data /var/infinity/wal /var/infinity/log /var/infinity/tmp");
 #endif
@@ -358,7 +358,7 @@ int main(int argc, char *argv[]) {
         case Mode::kQuery: {
             BenchmarkCreateIndex(infinity, db_name, table_name, index_name);
             BenchmarkInsert(infinity, db_name, table_name, srcfile, insert_batch);
-            // BenchmarkOptimize(infinity, db_name, table_name);
+            BenchmarkOptimize(infinity, db_name, table_name);
             sleep(10);
             BenchmarkMoreQuery(infinity, db_name, table_name, 1);
             break;

python/benchmark/configs/infinity_enwiki.json

@@ -9,7 +9,7 @@
     "query_link": "to_be_set",
     "mode": "fulltext",
     "topK": 10,
-    "use_import": false,
+    "use_import": true,
     "schema": {
         "doctitle": {"type": "varchar", "default":""},
         "docdate": {"type": "varchar", "default":""},

src/executor/operator/physical_match.cpp

@@ -494,45 +494,6 @@ void ASSERT_FLOAT_EQ(float bar, u32 i, float a, float b) {
     }
 }
 
-void AnalyzeFunc(const String &analyzer_name, String &&text, TermList &output_terms) {
-    UniquePtr<Analyzer> analyzer = AnalyzerPool::instance().Get(analyzer_name);
-    // (dynamic_cast<CommonLanguageAnalyzer*>(analyzer.get()))->SetExtractEngStem(false);
-    if (analyzer.get() == nullptr) {
-        RecoverableError(Status::UnexpectedError(fmt::format("Invalid analyzer: {}", analyzer_name)));
-    }
-    Term input_term;
-    input_term.text_ = std::move(text);
-    TermList temp_output_terms;
-    analyzer->Analyze(input_term, temp_output_terms);
-    if (analyzer_name == AnalyzerPool::STANDARD) {
-        // remove duplicates and only keep the root words for query
-        const u32 INVALID_TERM_OFFSET = -1;
-        Term last_term;
-        last_term.word_offset_ = INVALID_TERM_OFFSET;
-        for (const Term &term : temp_output_terms) {
-            if (last_term.word_offset_ != INVALID_TERM_OFFSET) {
-                assert(term.word_offset_ >= last_term.word_offset_);
-            }
-            if (last_term.word_offset_ != term.word_offset_) {
-                if (last_term.word_offset_ != INVALID_TERM_OFFSET) {
-                    output_terms.emplace_back(last_term);
-                }
-                last_term.text_ = term.text_;
-                last_term.word_offset_ = term.word_offset_;
-                last_term.stats_ = term.stats_;
-            } else {
-                if (term.text_.size() < last_term.text_.size()) {
-                    last_term.text_ = term.text_;
-                    last_term.stats_ = term.stats_;
-                }
-            }
-        }
-        if (last_term.word_offset_ != INVALID_TERM_OFFSET) {
-            output_terms.emplace_back(last_term);
-        }
-    }
-}
-
 void ExecuteFTSearch(UniquePtr<EarlyTerminateIterator> &et_iter, FullTextScoreResultHeap &result_heap, u32 &blockmax_loop_cnt) {
     if (et_iter) {
         while (true) {
@@ -585,12 +546,7 @@ bool PhysicalMatch::ExecuteInnerHomebrewed(QueryContext *query_context, Operator
     if (!query_tree) {
         RecoverableError(Status::ParseMatchExprFailed(match_expr_->fields_, match_expr_->matching_text_));
     }
-    if (query_tree->type_ == QueryNodeType::PHRASE) {
-        // TODO: make sure there is no problem with block max phrase and delete this code
-        // LOG_INFO(fmt::format("Block max phrase not supported, use ordinary iterator, query: {}", match_expr_->matching_text_));
-        use_block_max_iter = false;
-        use_ordinary_iter = true;
-    }
+
     auto finish_parse_query_tree_time = std::chrono::high_resolution_clock::now();
     TimeDurationType parse_query_tree_duration = finish_parse_query_tree_time - finish_init_query_builder_time;
     LOG_TRACE(fmt::format("PhysicalMatch Part 0.2: Parse QueryNode tree time: {} ms", parse_query_tree_duration.count()));
@@ -868,8 +824,9 @@ bool PhysicalMatch::Execute(QueryContext *query_context, OperatorState *operator
 SharedPtr<Vector<String>> PhysicalMatch::GetOutputNames() const {
     SharedPtr<Vector<String>> result_names = MakeShared<Vector<String>>();
     result_names->reserve(base_table_ref_->column_names_->size() + 2);
-    for (auto &name : *base_table_ref_->column_names_)
+    for (auto &name : *base_table_ref_->column_names_) {
         result_names->emplace_back(name);
+    }
     result_names->emplace_back(COLUMN_NAME_SCORE);
     result_names->emplace_back(COLUMN_NAME_ROW_ID);
     return result_names;

src/storage/invertedindex/format/position_list_decoder.cpp

@@ -47,8 +47,8 @@ void PositionListDecoder::InitPositionSkipList(const ByteSliceList *pos_list,
         state->SetRecordOffset(pos_skiplist_end);
     } else {
         pos_skiplist_reader_ = session_pool_ ? new ((session_pool_)->Allocate(sizeof(SkipListReaderByteSlice)))
-                                                   SkipListReaderByteSlice(option_.GetDocListFormatOption())
-                                             : new SkipListReaderByteSlice(option_.GetDocListFormatOption());
+                                                   SkipListReaderByteSlice(option_.GetPosListFormatOption())
+                                             : new SkipListReaderByteSlice(option_.GetPosListFormatOption());
         skiplist_reader_real_size_ = sizeof(SkipListReaderByteSlice);
         static_cast<SkipListReaderByteSlice *>(pos_skiplist_reader_)->Load(pos_list, pos_skiplist_start, pos_skiplist_end);
         decoded_pos_count_ = 0;
@@ -66,8 +66,8 @@ void PositionListDecoder::InitPositionSkipList(ByteSlice *pos_list,
         state->SetRecordOffset(pos_skiplist_end);
     } else {
         pos_skiplist_reader_ = session_pool_ ? new ((session_pool_)->Allocate(sizeof(SkipListReaderByteSlice)))
-                                                   SkipListReaderByteSlice(option_.GetDocListFormatOption())
-                                             : new SkipListReaderByteSlice(option_.GetDocListFormatOption());
+                                                   SkipListReaderByteSlice(option_.GetPosListFormatOption())
+                                             : new SkipListReaderByteSlice(option_.GetPosListFormatOption());
         skiplist_reader_real_size_ = sizeof(SkipListReaderByteSlice);
         static_cast<SkipListReaderByteSlice *>(pos_skiplist_reader_)->Load(pos_list, pos_skiplist_start, pos_skiplist_end);
         decoded_pos_count_ = 0;

src/storage/invertedindex/format/skiplist_reader.cppm

@@ -9,12 +9,15 @@ import doc_list_format_option;
 import memory_pool;
 import posting_byte_slice;
 import posting_byte_slice_reader;
+import position_list_format_option;
 
 namespace infinity {
 
 export class SkipListReader {
 public:
-    explicit SkipListReader(const DocListFormatOption &doc_list_format_option) : doc_list_format_option_(doc_list_format_option) {
+    explicit SkipListReader(const DocListFormatOption &doc_list_format_option)
+        : has_tf_list_(doc_list_format_option.HasTfList()),
+          has_block_max_(doc_list_format_option.HasBlockMax()) {
         if (has_tf_list_) {
             ttf_buffer_ = MakeUnique<u32[]>(SKIP_LIST_BUFFER_SIZE);
         }
@@ -24,6 +27,8 @@ public:
         }
     }
 
+    explicit SkipListReader(const PositionListFormatOption &doc_list_format_option) {}
+
     virtual ~SkipListReader() = default;
 
     bool SkipTo(u32 query_doc_id, u32 &doc_id, u32 &prev_doc_id, u32 &offset, u32 &delta);
@@ -55,9 +60,8 @@ public:
 protected:
     virtual Pair<int, bool> LoadBuffer() = 0;
 
-    DocListFormatOption doc_list_format_option_;
-    const bool has_tf_list_ = doc_list_format_option_.HasTfList();
-    const bool has_block_max_ = doc_list_format_option_.HasBlockMax();
+    const bool has_tf_list_ = false;
+    const bool has_block_max_ = false;
     i32 skipped_item_count_ = 0;
     u32 current_doc_id_ = 0;
     u32 current_offset_ = 0;
@@ -80,6 +84,8 @@ export class SkipListReaderByteSlice final : public SkipListReader {
 public:
     explicit SkipListReaderByteSlice(const DocListFormatOption &doc_list_format_option) : SkipListReader(doc_list_format_option) {}
 
+    explicit SkipListReaderByteSlice(const PositionListFormatOption &pos_list_format_option) : SkipListReader(pos_list_format_option) {}
+
     void Load(const ByteSliceList *byte_slice_list, u32 start, u32 end);
 
     void Load(ByteSlice *byteSlice, u32 start, u32 end);

src/storage/invertedindex/multi_posting_decoder.cpp

@@ -201,6 +201,7 @@ bool MultiPostingDecoder::DiskSegMoveToSegment(SegmentPosting &cur_segment_posti
     u32 doc_skiplist_end = doc_skiplist_start + doc_skiplist_size;
 
     index_decoder_->InitSkipList(doc_skiplist_start, doc_skiplist_end, posting_list, term_meta.GetDocFreq());
+
     if (format_option_.HasPositionList()) {
         u32 pos_list_begin = doc_list_reader.Tell() + doc_skiplist_size + doc_list_size;
         in_doc_state_keeper_.MoveToSegment(posting_list, term_meta.GetTotalTermFreq(), pos_list_begin, format_option_);

src/storage/invertedindex/multi_posting_decoder.cppm

@@ -71,9 +71,13 @@ private:
     }
 
     bool MoveToSegment(RowID start_row_id);
+
     bool MemSegMoveToSegment(const SharedPtr<PostingWriter> &posting_writer);
+
     bool DiskSegMoveToSegment(SegmentPosting &cur_segment_posting);
+
     IndexDecoder *CreateDocIndexDecoder(u32 doc_list_begin_pos);
+
 private:
     PostingFormatOption format_option_;
     bool need_decode_doc_id_ = false;

src/storage/invertedindex/search/bm25_ranker.cpp

@@ -20,6 +20,7 @@ module bm25_ranker;
 
 import stl;
 import index_defines;
+import third_party;
 
 namespace infinity {
 
@@ -34,23 +35,10 @@ void BM25Ranker::AddTermParam(u64 tf, u64 df, float avg_column_len, u32 column_l
     score_ += smooth_idf * smooth_tf * weight;
 }
 
-void BM25Ranker::AddPhraseParam(const Vector<tf_t>& all_tf,
-                                const Vector<u32>& all_df,
-                                float avg_column_len,
-                                u32 column_len,
-                                float weight,
-                                SizeT term_num) {
-    for (SizeT i = 0; i < term_num; ++i) {
-        u64 tf = 0;
-        u64 df = 0;
-        if (i < all_tf.size()) {
-            tf = all_tf[i];
-        }
-        if (i < all_df.size()) {
-            df = all_df[i];
-        }
-        AddTermParam(tf, df, avg_column_len, column_len, weight);
-    }
+void BM25Ranker::AddPhraseParam(tf_t tf, u64 df, float avg_column_len, u32 column_len, float weight) {
+    float smooth_idf = std::log(1.0F + (total_df_ - df + 0.5F) / (df + 0.5F));
+    float smooth_tf = (k1 + 1.0F) * tf / (tf + k1 * (1.0F - b + b * column_len / avg_column_len));
+    score_ += smooth_idf * smooth_tf * weight;
 }
 
 } // namespace infinity

src/storage/invertedindex/search/bm25_ranker.cppm

@@ -27,7 +27,7 @@ public:
 
     void AddTermParam(u64 tf, u64 df, float avg_column_len, u32 column_len, float weight);
 
-    void AddPhraseParam(const Vector<tf_t>& all_tf, const Vector<u32>& all_df, float avg_column_len, u32 column_len, float weight, SizeT term_num);
+    void AddPhraseParam(tf_t tf, u64 df, float avg_colum_len, u32 column_len, float weight);
 
     float GetScore() { return score_; }
 

src/storage/invertedindex/search/early_terminate_iterator/blockmax_and_iterator.cpp

@@ -21,6 +21,7 @@ import stl;
 import index_defines;
 import early_terminate_iterator;
 import internal_types;
+import third_party;
 
 namespace infinity {
 
@@ -79,7 +80,7 @@ bool BlockMaxAndIterator::BlockSkipTo(RowID doc_id, float threshold) {
                 sum_score += sorted_iterators_[j - 1]->BlockMaxBM25Score();
                 common_block_max_bm25_score_parts_[j - 1] = prev_sum_score;
             }
-            assert((sum_score <= bm25_score_upper_bound_));
+            // assert((sum_score <= bm25_score_upper_bound_));
             if (sum_score >= threshold) {
                 common_block_max_bm25_score_ = sum_score;
                 common_block_min_possible_doc_id_ = doc_id;

src/storage/invertedindex/search/early_terminate_iterator/blockmax_maxscore_iterator.cppm

@@ -90,4 +90,4 @@ private:
     Vector<Pair<u32, u64>> must_have_history_;
 };
 
-} // namespace infinity
+} // namespace infinity