[improvement](inverted index) Shrink SNII indexes on mostly NULL columns and use per-field BM25 document counts

be/src/common/config.cpp

@@ -1393,6 +1393,10 @@ DEFINE_mInt64(snii_forced_spill_min_arena_bytes, "67108864");
 // merge-compacted into one (bounds the k-way merge fan-in and its open fds;
 // every run is held open for the whole merge). 0 = uncapped. Default 64.
 DEFINE_mInt32(snii_spill_max_run_files_per_buffer, "64");
+// Sparse SNII norms: a norms section stores bytes only for the rows that carry a norm when that
+// is smaller than one byte per row. Off writes the legacy dense layout older BEs read; readers
+// accept both layouts either way.
+DEFINE_mBool(enable_snii_sparse_norms, "true");
 // dict path for chinese analyzer
 DEFINE_String(inverted_index_dict_path, "${DORIS_HOME}/dict");
 // The kuromoji (Japanese) analyzer
@@ -1408,10 +1412,11 @@ DEFINE_mBool(debug_inverted_index_compaction, "false");
 DEFINE_mBool(inverted_index_ram_dir_enable, "true");
 // wheather index by RAM directory when base compaction
 DEFINE_mBool(inverted_index_ram_dir_enable_when_base_compaction, "true");
-// Norms cost one byte per segment row, including rows that hold no value for the field. A segment
-// holds one index per variant path, so writing norms for them costs rows * paths bytes. Turn this on
-// to leave norms out of every index on a variant path, whatever its "norms" property says; BM25
-// scoring (score()) on those indexes then fails.
+// Norms cost one byte per segment row, including rows that hold no value for the field (SNII skips
+// those rows while enable_snii_sparse_norms is on). A segment holds one index per variant path, so
+// writing norms for them costs rows * paths bytes. Turn this on to leave norms out of every index
+// on a variant path, whatever its "norms" property says; BM25 scoring (score()) on those indexes
+// then fails.
 DEFINE_mBool(inverted_index_skip_norms_for_variant, "false");
 // use num_broadcast_buffer blocks as buffer to do broadcast
 DEFINE_Int32(num_broadcast_buffer, "32");

be/src/common/config.h

@@ -1447,6 +1447,15 @@ DECLARE_mInt64(snii_forced_spill_min_arena_bytes);
 // run counts across ~100 concurrent writers can exhaust the BE nofile rlimit
 // ("Too many open files" at run reopen). 0 disables the cap. Default 64.
 DECLARE_mInt32(snii_spill_max_run_files_per_buffer);
+// Lets SNII write BM25 norms in the sparse layout (section type kNormsSparse), which stores a
+// norm only for the rows that carry one, whenever that is smaller than the dense layout's one
+// byte per row; mostly NULL columns and VARIANT paths shrink the most. When off, every index
+// written from then on -- load, compaction output, schema change, BUILD INDEX -- uses the legacy
+// dense layout, which BEs without sparse-norms support can read: turn it off while such BEs may
+// read newly written segments (a rolling upgrade, or before a downgrade). Segments already
+// written keep their layout until they are rewritten. Read each time a logical index is
+// finished. Reading sparse sections is always supported, whatever the value.
+DECLARE_mBool(enable_snii_sparse_norms);
 // dict path for chinese analyzer
 DECLARE_String(inverted_index_dict_path);
 // The kuromoji (Japanese) analyzer
@@ -1462,10 +1471,11 @@ DECLARE_mBool(debug_inverted_index_compaction);
 DECLARE_mBool(inverted_index_ram_dir_enable);
 // wheather index by RAM directory when base compaction
 DECLARE_mBool(inverted_index_ram_dir_enable_when_base_compaction);
-// Norms cost one byte per segment row, including rows that hold no value for the field. A segment
-// holds one index per variant path, so writing norms for them costs rows * paths bytes. Turn this on
-// to leave norms out of every index on a variant path, whatever its "norms" property says; BM25
-// scoring (score()) on those indexes then fails.
+// Norms cost one byte per segment row, including rows that hold no value for the field (SNII skips
+// those rows while enable_snii_sparse_norms is on). A segment holds one index per variant path, so
+// writing norms for them costs rows * paths bytes. Turn this on to leave norms out of every index
+// on a variant path, whatever its "norms" property says; BM25 scoring (score()) on those indexes
+// then fails.
 DECLARE_mBool(inverted_index_skip_norms_for_variant);
 // use num_broadcast_buffer blocks as buffer to do broadcast
 DECLARE_Int32(num_broadcast_buffer);

be/src/storage/index/index_file_writer.cpp

@@ -267,10 +267,12 @@ Status IndexFileWriter::add_snii_index(const TabletIndex* index_meta, uint32_t d
     input.config = index_config;
     input.doc_count = doc_count;
     input.null_docids = std::move(null_docids);
+    input.write_norms = options.write_norms;
     input.encoded_norms = std::move(options.encoded_norms);
+    input.null_docids_with_norms = std::move(options.null_docids_with_norms);
     input.term_source = term_buffer;
     input.mem_reporter = mem_reporter;
-    snii_resolve_index_write_params(options.is_direct_load, !input.encoded_norms.empty(), &input);
+    snii_resolve_index_write_params(options.is_direct_load, input.write_norms, &input);
     RETURN_IF_ERROR(_snii_compound_writer->add_logical_index(input));
     ++_snii_index_count;
     return Status::OK();

be/src/storage/index/index_file_writer.h

@@ -110,9 +110,13 @@ class IndexFileWriter {
         // the prx region compresses at snii_prx_zstd_level_direct_load;
         // compaction / schema change / ADD INDEX keep snii_prx_zstd_level.
         bool is_direct_load = false;
-        // One byte of BM25 norms per document; empty for keyword or positionless indexes.
-        // If nonempty, its size must equal doc_count, and postings retain frequencies for scoring.
+        // The index writes BM25 norms (analyzed indexes with positions, unless the norms
+        // policy turns them off); postings then retain frequencies for scoring.
+        bool write_norms = false;
+        // One encoded norm per document that carries one, in docid order: every document
+        // outside null_docids plus null_docids_with_norms. See SniiIndexInput.
         std::vector<uint8_t> encoded_norms;
+        std::vector<uint32_t> null_docids_with_norms;
     };
     Status add_snii_index(const TabletIndex* index_meta, uint32_t doc_count,
                           std::vector<uint32_t> null_docids,

be/src/storage/index/inverted/similarity/collection_statistics.cpp

@@ -17,6 +17,7 @@
 
 #include "storage/index/inverted/similarity/collection_statistics.h"
 
+#include <algorithm>
 #include <exception>
 #include <set>
 #include <sstream>
@@ -41,7 +42,7 @@
 namespace doris {
 namespace collection_statistics_detail {
 
-Result<SniiScoringSegmentStats> resolve_snii_scoring_segment(uint64_t index_doc_count,
+Result<SniiScoringSegmentStats> resolve_snii_scoring_segment(uint64_t indexed_doc_count,
                                                              uint64_t sum_total_term_freq,
                                                              bool has_positions, bool has_norms) {
     if (!has_positions || !has_norms) {
@@ -51,7 +52,7 @@ Result<SniiScoringSegmentStats> resolve_snii_scoring_segment(uint64_t index_doc_
                 "\"false\" or, for a variant path, when inverted_index_skip_norms_for_variant is "
                 "on"));
     }
-    return SniiScoringSegmentStats {.doc_count = index_doc_count,
+    return SniiScoringSegmentStats {.doc_count = indexed_doc_count,
                                     .token_count = sum_total_term_freq};
 }
 
@@ -133,10 +134,10 @@ Status CollectionStatistics::collect_full_collection(
         }
         oss << "]";
 
-        oss << ", total_num_docs=" << _total_num_docs;
-
         for (const auto& [ws_field_name, num_tokens] : _total_num_tokens) {
+            const auto num_docs = _total_num_docs.find(ws_field_name);
             oss << ", {field=" << StringHelper::to_string(ws_field_name)
+                << ", num_docs=" << (num_docs == _total_num_docs.end() ? 0 : num_docs->second)
                 << ", num_tokens=" << num_tokens << ", terms=[";
 
             auto field_term_doc_freqs = _term_doc_freqs.find(ws_field_name);
@@ -245,11 +246,19 @@ Status CollectionStatistics::process_segment(const RowsetSharedPtr& rowset,
                 return status;
             }
             auto logical_reader = std::move(logical_reader_result.value());
-            const uint64_t segment_doc_count = logical_reader->stats().doc_count;
+            const auto& logical_stats = logical_reader->stats();
+            const uint64_t segment_doc_count = logical_stats.doc_count;
+            // Every SNII writer stores indexed_doc_count = doc_count - null_count, and the
+            // metadata decoder rejects a stats block without it.
+            if (logical_stats.indexed_doc_count > segment_doc_count) {
+                return Status::Error<ErrorCode::INVERTED_INDEX_FILE_CORRUPTED>(
+                        "SNII indexed document count {} exceeds segment document count {}",
+                        logical_stats.indexed_doc_count, segment_doc_count);
+            }
             RETURN_IF_ERROR(admit_snii_scoring_segment(
-                    ws_field_name, segment_doc_count, logical_reader->stats().sum_total_term_freq,
-                    logical_reader->has_positions(), logical_reader->has_norms(),
-                    &segment_accumulator));
+                    ws_field_name, logical_stats.indexed_doc_count,
+                    logical_stats.sum_total_term_freq, logical_reader->has_positions(),
+                    logical_reader->has_norms(), &segment_accumulator));
 
             ::doris::snii::reader::DictBlockCache dict_block_cache;
             for (const auto& logical_term_bytes : collect_info.unique_terms) {
@@ -342,37 +351,35 @@ Status CollectionStatistics::process_segment(const RowsetSharedPtr& rowset,
         }
     }
 
-    _total_num_docs += static_cast<uint64_t>(total_segment_docs);
+    for (const auto& [ws_field_name, collect_info] : collect_infos) {
+        _total_num_docs[ws_field_name] += static_cast<uint64_t>(total_segment_docs);
+    }
     _avg_dl_by_col.clear();
     _idf_by_col_term.clear();
 
     return Status::OK();
 }
 
 Status CollectionStatistics::admit_snii_scoring_segment(
-        const std::wstring& field_name, uint64_t index_doc_count, uint64_t sum_total_term_freq,
+        const std::wstring& field_name, uint64_t indexed_doc_count, uint64_t sum_total_term_freq,
         bool has_positions, bool has_norms, SniiScoringSegmentAccumulator* segment_accumulator) {
     DORIS_CHECK(segment_accumulator != nullptr);
     auto segment_stats = collection_statistics_detail::resolve_snii_scoring_segment(
-            index_doc_count, sum_total_term_freq, has_positions, has_norms);
+            indexed_doc_count, sum_total_term_freq, has_positions, has_norms);
     if (!segment_stats.has_value()) {
         clear();
         return segment_stats.error();
     }
-    if (!segment_accumulator->token_counts.empty() &&
-        segment_accumulator->doc_count != segment_stats->doc_count) {
-        clear();
-        return Status::Error<ErrorCode::INVERTED_INDEX_NOT_SUPPORTED>(
-                "SNII scoring fields in one segment have different document counts: {} and {}",
-                segment_accumulator->doc_count, segment_stats->doc_count);
-    }
-    segment_accumulator->doc_count = segment_stats->doc_count;
+    segment_accumulator->doc_counts[field_name] += segment_stats->doc_count;
     segment_accumulator->token_counts[field_name] += segment_stats->token_count;
     return Status::OK();
 }
 
 void CollectionStatistics::commit_snii_scoring_segment(
         SniiScoringSegmentAccumulator&& segment_accumulator) {
+    for (const auto& [field_name, doc_count] : segment_accumulator.doc_counts) {
+        _total_num_docs[field_name] += doc_count;
+    }
     for (const auto& [field_name, token_count] : segment_accumulator.token_counts) {
         _total_num_tokens[field_name] += token_count;
     }
@@ -381,13 +388,12 @@ void CollectionStatistics::commit_snii_scoring_segment(
             _term_doc_freqs[field_name][term] += doc_freq;
         }
     }
-    _total_num_docs += segment_accumulator.doc_count;
     _avg_dl_by_col.clear();
     _idf_by_col_term.clear();
 }
 
 void CollectionStatistics::clear() {
-    _total_num_docs = 0;
+    _total_num_docs.clear();
     _total_num_tokens.clear();
     _term_doc_freqs.clear();
     _avg_dl_by_col.clear();
@@ -424,14 +430,15 @@ uint64_t CollectionStatistics::get_total_term_cnt_by_col(const std::wstring& luc
     return token_count->second;
 }
 
-uint64_t CollectionStatistics::get_doc_num() const {
-    if (_total_num_docs == 0) {
-        throw Exception(
-                ErrorCode::INVERTED_INDEX_CLUCENE_ERROR,
-                "Index statistics collection failed: No data available for SimilarityCollector");
+uint64_t CollectionStatistics::get_doc_num(const std::wstring& lucene_col_name) const {
+    const auto doc_count = _total_num_docs.find(lucene_col_name);
+    if (doc_count == _total_num_docs.end()) {
+        throw Exception(ErrorCode::INVERTED_INDEX_CLUCENE_ERROR,
+                        "Index statistics collection failed: Not such column {}",
+                        StringHelper::to_string(lucene_col_name));
     }
 
-    return _total_num_docs;
+    return doc_count->second;
 }
 
 float CollectionStatistics::get_or_calculate_avg_dl(const std::wstring& lucene_col_name) {
@@ -441,8 +448,12 @@ float CollectionStatistics::get_or_calculate_avg_dl(const std::wstring& lucene_c
     }
 
     const uint64_t total_term_cnt = get_total_term_cnt_by_col(lucene_col_name);
-    const uint64_t total_doc_cnt = get_doc_num();
-    float avg_dl = total_doc_cnt > 0 ? float((double)total_term_cnt / (double)total_doc_cnt) : 0.0F;
+    // A field without documents has no tokens either, except for NULL ARRAY rows that kept
+    // their tokens (they sit in postings but not in the indexed count). Dividing by at least
+    // one keeps avgdl finite, and positive whenever a posting exists.
+    const uint64_t total_doc_cnt = std::max<uint64_t>(get_doc_num(lucene_col_name), 1);
+    const auto avg_dl = static_cast<float>(static_cast<double>(total_term_cnt) /
+                                           static_cast<double>(total_doc_cnt));
     _avg_dl_by_col[lucene_col_name] = avg_dl;
     return avg_dl;
 }
@@ -457,10 +468,13 @@ float CollectionStatistics::get_or_calculate_idf(const std::wstring& lucene_col_
         }
     }
 
-    const uint64_t doc_num = get_doc_num();
     const uint64_t doc_freq = get_term_doc_freq_by_col(lucene_col_name, term);
-    auto idf = (float)std::log(1 + ((double)doc_num - (double)doc_freq + (double)0.5) /
-                                           ((double)doc_freq + (double)0.5));
+    // doc_freq never exceeds the document count, except on SNII fields whose NULL ARRAY rows
+    // kept tokens: those rows count in doc_freq but not in the indexed document count. Using
+    // the larger value keeps idf positive; it changes nothing for any other collection.
+    const uint64_t doc_num = std::max(get_doc_num(lucene_col_name), doc_freq);
+    auto idf = (float)std::log(1 + ((double)doc_num - (double)doc_freq + 0.5) /
+                                           ((double)doc_freq + 0.5));
     _idf_by_col_term[lucene_col_name][term] = idf;
     return idf;
 }

be/src/storage/index/inverted/similarity/collection_statistics.h

@@ -71,7 +71,7 @@ class CollectionStatistics {
 
 private:
     struct SniiScoringSegmentAccumulator {
-        uint64_t doc_count = 0;
+        std::unordered_map<std::wstring, uint64_t> doc_counts;
         std::unordered_map<std::wstring, uint64_t> token_counts;
         std::unordered_map<std::wstring, std::unordered_map<std::wstring, uint64_t>> term_doc_freqs;
     };
@@ -83,7 +83,7 @@ class CollectionStatistics {
     Status process_segment(const RowsetSharedPtr& rowset, const RowsetSegmentView& seg,
                            const TabletSchema* tablet_schema, const CollectInfoMap& collect_infos,
                            io::IOContext* io_ctx);
-    Status admit_snii_scoring_segment(const std::wstring& field_name, uint64_t index_doc_count,
+    Status admit_snii_scoring_segment(const std::wstring& field_name, uint64_t indexed_doc_count,
                                       uint64_t sum_total_term_freq, bool has_positions,
                                       bool has_norms,
                                       SniiScoringSegmentAccumulator* segment_accumulator);
@@ -93,9 +93,15 @@ class CollectionStatistics {
     uint64_t get_term_doc_freq_by_col(const std::wstring& lucene_col_name,
                                       const std::wstring& term);
     uint64_t get_total_term_cnt_by_col(const std::wstring& lucene_col_name);
-    uint64_t get_doc_num() const;
-
-    uint64_t _total_num_docs = 0;
+    uint64_t get_doc_num(const std::wstring& lucene_col_name) const;
+
+    // Per field, the document count BM25 uses as N in idf and as the avgdl denominator.
+    // CLucene segments add their document count (the largest maxDoc among the segment's
+    // scored fields) to every field, as before per-field counts existed. SNII segments add
+    // the field's indexed (non-NULL) document count, which is what Lucene's per-field
+    // docCount means; a mostly NULL field would otherwise get a far too small avgdl and
+    // flattened idf. A field may count 0 documents.
+    std::unordered_map<std::wstring, uint64_t> _total_num_docs;
     std::unordered_map<std::wstring, uint64_t> _total_num_tokens;
     std::unordered_map<std::wstring, std::unordered_map<std::wstring, uint64_t>> _term_doc_freqs;
 
@@ -121,8 +127,9 @@ struct SniiScoringSegmentStats {
 
 // SNII scoring requires positions (which provide term frequencies) and norms. The current writer
 // emits norms for every analyzed index with positions. Older segments without norms return
-// NOT_SUPPORTED until an index rebuild or compaction supplies them.
-Result<SniiScoringSegmentStats> resolve_snii_scoring_segment(uint64_t index_doc_count,
+// NOT_SUPPORTED until an index rebuild or compaction supplies them. The field's document count is
+// its indexed (non-NULL) document count.
+Result<SniiScoringSegmentStats> resolve_snii_scoring_segment(uint64_t indexed_doc_count,
                                                              uint64_t sum_total_term_freq,
                                                              bool has_positions, bool has_norms);
 

be/src/storage/index/snii/bkd/bkd_format.h

@@ -62,7 +62,7 @@ inline constexpr uint32_t kSupportedVersion = 1;
 // no shared SectionType enum for blob logical indexes, so -- exactly as
 // format::kNullBitmapSectionType (0x20) does -- this is a documented literal
 // picked outside the ranges already taken by the inverted-index sections
-// (format::SectionType, currently 1..14) and the null-bitmap POD (0x20).
+// (format::SectionType, currently 1..15) and the null-bitmap POD (0x20).
 // Framing the payload is what gives bkd_index its checksum; no section here
 // hand-rolls a crc.
 inline constexpr uint8_t kBkdIndexSectionType = 0x30;