airborne12 commented on code in PR #67538: URL: https://github.com/apache/doris/pull/67538#discussion_r3985302094
########## be/test/storage/index/snii/writer/high_df_digest_cost_test.cpp: ########## @@ -0,0 +1,178 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// What does building the high-df digest cost the writer? +// +// The digest earns its keep on the query side -- it lets the cost gate bound a node without +// the remote dictionary reads that were measured at 40 seconds on a cold segment. That is +// only a good trade if writing it is close to free, and "close to free" has to be a number +// rather than an assertion. +// +// The two builds below index the identical corpus with the identical settings and differ in +// exactly one field: whether a gram scheme is present, which is what decides that a digest +// is built at all. Everything else -- postings, dictionary blocks, term hashes, statistics +// -- is byte-for-byte the same work, so the difference in elapsed time is the digest. +// +// The shape of the corpus matters to what is measured. Term frequencies follow a Zipf-like +// curve, as a real vocabulary does, so most terms fall below the digest's floor and cost +// only the one comparison that rejects them, while a small head clears it and pays for a +// bounded-heap insertion. Measuring on a flat corpus would either exercise no heap at all +// or exercise nothing but the heap; neither resembles a real index. + +#include <gtest/gtest.h> + +#include <algorithm> +#include <chrono> +#include <cmath> +#include <cstdint> +#include <string> +#include <vector> + +#include "storage/index/snii/format/format_constants.h" +#include "storage/index/snii/reader/logical_index_reader.h" +#include "storage/index/snii/reader/snii_segment_reader.h" +#include "storage/index/snii/writer/logical_index_writer.h" +#include "storage/index/snii/writer/snii_compound_writer.h" +#include "storage/index/snii_query_test_util.h" + +namespace doris::snii::writer { +namespace { + +using snii_test::assert_ok; +using snii_test::make_term; +using snii_test::MemoryFile; +using snii_test::PostingDoc; + +constexpr uint32_t kDocCount = 200000; +constexpr uint32_t kTermCount = 40000; +constexpr int kRounds = 5; + +// A Zipf-like vocabulary over kDocCount documents: term i appears in roughly +// kTermCount / (i + 1) documents, so a few hundred terms clear the digest floor +// (kDocCount / kHighDfDigestDivisor = 100) and the rest sit far below it. +std::vector<TermPostings> BuildCorpus() { + std::vector<TermPostings> terms; + terms.reserve(kTermCount); + for (uint32_t i = 0; i < kTermCount; i++) { + const uint32_t df = std::max<uint32_t>(1, kTermCount / (i + 1)); + const uint32_t stride = std::max<uint32_t>(1, kDocCount / df); + std::vector<PostingDoc> docs; + docs.reserve(df); + for (uint32_t docid = 0; docid < kDocCount && docs.size() < df; docid += stride) { + docs.push_back(PostingDoc {.docid = docid, .positions = {}}); + } + // Zero-padded so the terms are already in lexicographic order. + char name[32]; + snprintf(name, sizeof(name), "term_%06u", i); + terms.push_back(make_term(name, std::move(docs))); + } + return terms; +} + +// One build of the whole index. with_digest selects it by supplying a gram scheme, which is +// the only thing that differs between the two arms. +double BuildOnceMs(const std::vector<TermPostings>& corpus, bool with_digest, + uint64_t* digest_entries) { + SniiIndexInput input; + input.index_id = 31; + input.index_suffix = "body"; + input.config = format::IndexConfig::kDocsOnly; + input.doc_count = kDocCount; + input.terms = corpus; + if (with_digest) { + segment_v2::gram::GramScheme scheme; + scheme.mode = segment_v2::gram::GramMode::SPARSE; + scheme.min_len = 3; + scheme.max_len = 4; + scheme.density_permille = 250; + input.gram_scheme = scheme; + } + + MemoryFile file; + SniiCompoundWriter writer(&file); + const auto start = std::chrono::steady_clock::now(); + assert_ok(writer.add_logical_index(input)); + assert_ok(writer.finish()); + const auto end = std::chrono::steady_clock::now(); + + if (digest_entries != nullptr) { + reader::SniiSegmentReader segment; + assert_ok(reader::SniiSegmentReader::open(&file, &segment)); + reader::LogicalIndexReader index; + assert_ok(segment.open_index(31, "body", &index)); + *digest_entries = index.high_df_terms().term_hash.size(); + } + return std::chrono::duration<double, std::milli>(end - start).count(); +} + +double Median(std::vector<double> values) { + std::ranges::sort(values); + return values[values.size() / 2]; +} + +} // namespace + +// Reports the cost; asserts only that it is not a large fraction of the build, so the case +// records a number without becoming a tripwire for ordinary timing noise. +TEST(SniiHighDfDigestCost, BuildingTheDigestIsANegligibleShareOfTheWrite) { + const std::vector<TermPostings> corpus = BuildCorpus(); + + uint64_t entries = 0; + std::vector<double> with_digest; + std::vector<double> without; + for (int round = 0; round < kRounds; round++) { + // Alternate the order so a warming effect cannot favour one arm systematically. + if (round % 2 == 0) { + with_digest.push_back(BuildOnceMs(corpus, true, &entries)); + without.push_back(BuildOnceMs(corpus, false, nullptr)); + } else { + without.push_back(BuildOnceMs(corpus, false, nullptr)); + with_digest.push_back(BuildOnceMs(corpus, true, &entries)); + } + } + + const double on = Median(with_digest); + const double off = Median(without); + const double overhead_pct = 100.0 * (on - off) / off; + // Spread of the two arms, so a reader can see whether the difference above is signal. + const double on_spread = + *std::ranges::max_element(with_digest) - *std::ranges::min_element(with_digest); + const double off_spread = + *std::ranges::max_element(without) - *std::ranges::min_element(without); + + printf("\n%u docs, %u terms, digest holds %llu entries\n", kDocCount, kTermCount, + static_cast<unsigned long long>(entries)); + printf("build with digest : %8.1f ms (median of %d, spread %.1f ms)\n", on, kRounds, + on_spread); + printf("build without digest : %8.1f ms (median of %d, spread %.1f ms)\n", off, kRounds, + off_spread); + printf("difference : %+8.1f ms (%+.2f%%)\n", on - off, overhead_pct); + printf("per term : %+8.3f us\n", 1000.0 * (on - off) / kTermCount); + // The work itself, which is deterministic and does not depend on the machine: one + // comparison for every term, and a bounded-heap insertion only for those above the + // floor. Reported alongside the timing because it is the number that stays true. + printf("work : %u comparisons + at most %llu heap ops (log2 K <= %.1f)\n", + kTermCount, static_cast<unsigned long long>(entries), + entries > 0 ? std::log2(static_cast<double>(entries)) : 0.0); + + EXPECT_GT(entries, 0U) << "the corpus must actually populate the digest"; + EXPECT_LT(overhead_pct, 5.0) Review Comment: Fixed in 94d3ca3a1e9: the 5% wall-clock gate is gone. The test pins the deterministic invariant (the digest never exceeds `kMaxHighDfDigestTerms` entries) and keeps the timing diagnostic-only. ########## regression-test/suites/inverted_index_p0/gram/test_gram_stop_gram.groovy: ########## @@ -0,0 +1,192 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// stop-gram end to end: an index built with the posting lists of very common grams dropped +// must still answer LIKE/REGEXP exactly, and must actually have dropped them. +// +// A dropped gram matches every document, so it can only ever widen the candidate set the +// index proposes, and the predicate is re-evaluated on those candidates. The observable +// contract is therefore equality: every pattern must return exactly what a full scan +// returns. The rows below are built so that the common grams really are common -- a shared +// prefix on every row -- while the patterns being searched for are rare, which is the shape +// where dropping matters. +// +// There is no switch to turn the feature off, so the control arm is the row floor instead: +// a segment under kHighDfDigestDivisor (2000) rows never drops a posting. The same 4000 rows +// are therefore loaded twice, once as a single segment (dropping active, threshold +// 4000 / 2000 * 3 = df 6) and once as five segments of 800 (dropping inactive, every +// posting kept). The two tables must agree on every answer, and the single-segment index +// must be much smaller -- that difference is the dropped postings, and it also pins that +// the floor holds. +suite("test_gram_stop_gram", "p0") { + def waitAnalyzerInstalled = { String name -> + def deadline = System.currentTimeMillis() + 180_000 + Exception lastNotFound = null + while (System.currentTimeMillis() < deadline) { + try { + sql """SELECT TOKENIZE('probe', '"analyzer"="${name}"')""" + return + } catch (Exception e) { + if (!e.message.contains("Policy not found")) { + throw e + } + lastNotFound = e + sleep(1000) + } + } + throw new IllegalStateException("analyzer ${name} was not installed on BE", lastNotFound) + } + + def backendId_to_backendIP = [:] + def backendId_to_backendHttpPort = [:] + getBackendIpHttpPort(backendId_to_backendIP, backendId_to_backendHttpPort) + def set_be_config = { key, value -> + for (String backend_id : backendId_to_backendIP.keySet()) { + def (code, out, err) = update_be_config(backendId_to_backendIP.get(backend_id), + backendId_to_backendHttpPort.get(backend_id), key, value) + logger.info("update ${key}=${value}: code=${code}, out=${out}, err=${err}") + } + } + + // The tables go first: a policy still referenced by a table left behind by an earlier run + // cannot be dropped. + sql "DROP TABLE IF EXISTS test_gram_stop_gram_one" + sql "DROP TABLE IF EXISTS test_gram_stop_gram_five" + sql "DROP INVERTED INDEX ANALYZER IF EXISTS gram_stop_ana" + sql "DROP INVERTED INDEX TOKENIZER IF EXISTS gram_stop_tok" + sql """CREATE INVERTED INDEX TOKENIZER gram_stop_tok PROPERTIES ( + "type"="ngram", "mode"="sparse", "min_gram"="3", "max_gram"="8", "density"="0.5")""" + sql """CREATE INVERTED INDEX ANALYZER gram_stop_ana + PROPERTIES ("tokenizer"="gram_stop_tok")""" + waitAnalyzerInstalled("gram_stop_ana") + + def rows = 4000 + def values = [] + for (int i = 0; i < rows; i++) { + // Every row shares "shared_prefix_", making its grams as common as a gram gets. Four + // rows carry a rare marker (df 4, under the single segment's threshold of 6, so its + // postings survive and the rare queries still filter); one row in seven carries a + // mid-frequency tail whose grams sit above the line and are dropped. + def tail = (i % 1000 == 0) ? "rare_marker_${i}" : "filler_${i}" + values.add("(${i}, 'shared_prefix_common_text ${tail} tail_${i % 7}')") + } + + def runAll = { String table, String label -> + def out = [:] + out["like_rare"] = sql "SELECT COUNT(*) FROM ${table} WHERE msg LIKE '%rare_marker_%'" + out["like_common"] = sql "SELECT COUNT(*) FROM ${table} WHERE msg LIKE '%shared_prefix_common%'" + out["regexp_rare"] = sql "SELECT COUNT(*) FROM ${table} WHERE msg REGEXP 'rare_marker_[0-9]+'" + out["regexp_common"] = sql "SELECT COUNT(*) FROM ${table} WHERE msg REGEXP 'shared_prefix_[a-z]+'" + out["regexp_mid"] = sql "SELECT COUNT(*) FROM ${table} WHERE msg REGEXP 'tail_3\$'" + out["regexp_alt"] = sql "SELECT COUNT(*) FROM ${table} WHERE msg REGEXP 'rare_marker_(0|1000|2000)\\\\b'" + out["regexp_absent"] = sql "SELECT COUNT(*) FROM ${table} WHERE msg REGEXP 'no_such_token_anywhere'" + out["like_tail"] = sql "SELECT COUNT(*) FROM ${table} WHERE msg LIKE '%tail_3'" + logger.info("${label}: ${out}") + return out + } + + def createTable = { String table -> + sql "DROP TABLE IF EXISTS ${table}" + sql """CREATE TABLE ${table} ( + `id` bigint NULL, + `msg` text NULL, + INDEX idx_msg (`msg`) USING INVERTED + PROPERTIES('analyzer'='gram_stop_ana', 'support_phrase'='false') + ) ENGINE=OLAP DUPLICATE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES('replication_num'='1', 'inverted_index_storage_format'='SNII', + 'disable_auto_compaction'='true')""" + } + + // information_schema reports a fresh table's sizes with a delay, and 0 there means "not + // yet" rather than "empty". + def indexBytes = { String table -> + def deadline = System.currentTimeMillis() + 180_000 + while (System.currentTimeMillis() < deadline) { + def r = sql """SELECT INDEX_LENGTH FROM information_schema.tables + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '${table}'""" + def bytes = r.isEmpty() ? 0L : (r[0][0] as long) + if (bytes > 0) { + return bytes + } + sleep(3000) + } + throw new IllegalStateException("index size of ${table} never reported") + } + + def oneSegment = "test_gram_stop_gram_one" + def fiveSegments = "test_gram_stop_gram_five" + try { + sql "SET enable_sql_cache=false" + sql "SET enable_condition_cache=false" + + // One batch keeps all 4000 rows in a single segment: above the row floor, dropping + // is in effect. + createTable(oneSegment) + sql "INSERT INTO ${oneSegment} VALUES ${values.join(',')}" + + // The same rows in five batches of 800: each segment is under the floor, so every + // posting list is kept. This is what the index looked like before dropping existed. + createTable(fiveSegments) + for (int b = 0; b < 5; b++) { + sql "INSERT INTO ${fiveSegments} VALUES ${values.subList(b * 800, (b + 1) * 800).join(',')}" + } + + // Ground truth, taken with the index out of the picture entirely. + sql "SET enable_inverted_index_query=false" + def scanned = runAll(oneSegment, "full scan") + sql "SET enable_inverted_index_query=true" + + def dropped = runAll(oneSegment, "one segment, postings dropped") + def kept = runAll(fiveSegments, "five segments, postings kept") + scanned.each { name, value -> + assertEquals(value[0][0], dropped[name][0][0], + "index and scan disagree on ${name} with postings dropped") + assertEquals(value[0][0], kept[name][0][0], + "index and scan disagree on ${name} with postings kept") + } + + // The postings really were dropped. Measured: the single-segment index is 0.36 of + // the five-segment one. Had nothing been dropped it would be about 0.9 -- the only + // difference left would be four fewer per-segment dictionaries -- so a factor of two + // separates the two outcomes with margin on both sides. + def droppedBytes = indexBytes(oneSegment) + def keptBytes = indexBytes(fiveSegments) + logger.info("index bytes: one segment ${droppedBytes}, five segments ${keptBytes}") + assertTrue(droppedBytes * 2 < keptBytes, + "the single-segment index (${droppedBytes} bytes) should be far smaller than " + + "the five-segment one (${keptBytes} bytes): its common postings were not dropped") + + // The gate could otherwise mask a broken dropped-gram path by giving up first, so + // check again with its fallback ratio disabled. The gate's primary budget comes from + // the segment itself and cannot be switched off -- that is the point of deriving it + // rather than configuring it -- but zeroing the ratio removes the one arm that a + // configuration could have been hiding behind. + set_be_config("gram_index_max_candidate_ratio_bp", "0") Review Comment: Fixed in 94d3ca3a1e9: the suite sets `enable_inverted_index_query_cache=false` for every phase, asserts each `update_be_config` call succeeded (exit 0 and an OK response), and runs a pattern whose grams straddle the line -- `common_text rare_marker_1000 `: the shared prefix's grams were dropped, the rare marker's grams were kept -- checking in the profile that the dictionary was resolved: `GramIndexGateGaveUp == 0`, `1 <= GramIndexCandidateRows <= 4` (bounded by the kept df-4 posting), `RowsGramIndexFiltered >= rows - 4`. ########## be/src/storage/index/inverted/gram/gram_density.cpp: ########## @@ -0,0 +1,125 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "storage/index/inverted/gram/gram_density.h" + +#include <algorithm> + +#include "storage/index/inverted/gram/gram_extractor.h" + +namespace doris::segment_v2::gram { + +DensitySolver::DensitySolver(size_t min_literal_len, size_t max_gram_len) + : _min_literal_len(min_literal_len), + _max_gram_len(max_gram_len), + _histogram(kHashValues, 0) {} + +void DensitySolver::observe(std::string_view value) { + if (_min_literal_len < _max_gram_len || _max_gram_len < 2) { + return; + } + // Boundaries with a whole gram's room after them inside a window: a boundary at k yields + // [k, k+max_gram_len) at worst, so k may not exceed len - max_gram_len. That leaves + // `per_window` candidate positions in a window of the promised length. + const size_t per_window = _min_literal_len - _max_gram_len + 1; + + size_t i = 0; + const size_t n = value.size(); + while (i < n) { + // Only ASCII runs: they are the only bytes the extractor indexes, so a window + // spanning anything else could never produce a gram. + if (static_cast<unsigned char>(value[i]) >= 0x80) { + ++i; + continue; + } + size_t j = i; + while (j < n && static_cast<unsigned char>(value[j]) < 0x80) { + ++j; + } + const std::string_view run = value.substr(i, j - i); + i = j; + if (run.size() < _min_literal_len) { + continue; + } + // Sliding minimum over the run's pair hashes, one window per start position. The + // monotonic queue keeps this linear in the run rather than quadratic in the window, + // which is what makes it affordable on the write path. + const size_t pairs = run.size() - 1; + _mono.clear(); + size_t head = 0; + for (size_t p = 0; p < pairs; ++p) { + const uint16_t h = + boundary_hash16(static_cast<uint8_t>(run[p]), static_cast<uint8_t>(run[p + 1])); + while (_mono.size() > head && + boundary_hash16(static_cast<uint8_t>(run[_mono.back()]), + static_cast<uint8_t>(run[_mono.back() + 1])) >= h) { + _mono.pop_back(); + } + _mono.push_back(static_cast<uint32_t>(p)); + if (p + 1 < per_window) { + continue; + } + const size_t window_start = p + 1 - per_window; + while (_mono[head] < window_start) { + ++head; + } + const uint16_t min_hash = boundary_hash16(static_cast<uint8_t>(run[_mono[head]]), + static_cast<uint8_t>(run[_mono[head] + 1])); + ++_histogram[min_hash]; + ++_windows; + } + } +} + +uint16_t DensitySolver::solve(uint32_t coverage_permille) const { + if (_windows == 0 || coverage_permille == 0) { + // No evidence, or nothing asked for. Stay where a configured default would have been + // rather than inventing a rate from nothing. + return kMaxSolvedDensityPermille; + } + // A window is covered exactly when its minimum is below the threshold, so the prefix sum + // of the histogram IS the coverage curve: walk it until the requested share is reached and + // the threshold that achieves it -- and nothing larger -- is one past that hash. + const uint64_t need = (static_cast<uint64_t>(coverage_permille) * _windows + 999) / 1000; + uint64_t seen = 0; + uint32_t threshold = kHashValues; + for (size_t h = 0; h < kHashValues; ++h) { + seen += _histogram[h]; + if (seen >= need) { + // `is_boundary` compares strictly, so the threshold has to clear this hash. + threshold = static_cast<uint32_t>(h) + 1; + break; + } + } + // GramExtractor derives its threshold as density_permille * 65536 / 1000 with integer + // division; round up or the resolved density can land a notch below what was solved for. + uint32_t permille = + static_cast<uint32_t>((static_cast<uint64_t>(threshold) * 1000 + 65535) / 65536); + permille = std::clamp<uint32_t>(permille, kMinSolvedDensityPermille, kMaxSolvedDensityPermille); Review Comment: Fixed in 2cc0776c891: `DensitySolver::solve_detailed` reports the unclamped `required_permille`, whether the answer was clamped, and the share the clamped rate really keeps; the writer logs a WARNING naming the unmet target when that share is below the requested coverage instead of recording the ceiling as if it met it. The bounds themselves stay: a required rate above 500 permille says the corpus has no boundary structure a sparse scheme can use, and the honest outcome is to report it. Tests: `GramDensityTest.ReportsAClampedSolveAndItsRealCoverage`, and the calibration test drives a `%`-only corpus to the ceiling. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
