[refactor](be) Give the ANN index a storage interface of its own

be/benchmark/analyzer_ab.py

@@ -0,0 +1,101 @@
+#!/usr/bin/env python3
+# 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.
+
+"""Paired throughput comparison of two benchmark_test binaries built with
+benchmark_analyzer.hpp, as analyzer_benchmark.md describes.
+
+Each round runs both binaries once, alternating which goes first, so drift in the host's load
+hits both sides alike. For every benchmark the candidate/baseline throughput ratio is taken per
+round and the median over the rounds is reported, with the geometric mean over the corpora of
+each analyzer. Give the same binary as both sides for the A/A check: every ratio should stay
+within 5% of 1.
+
+usage: analyzer_ab.py --baseline BIN --candidate BIN --filter REGEX --out DIR [--rounds N]
+
+The environment the binaries need (ANALYZER_BENCH_CORPUS_DIR, ANALYZER_BENCH_DICT_DIR, ...) is
+passed through.
+"""
+
+import argparse
+import json
+import math
+import os
+import statistics
+import subprocess
+import sys
+
+
+def run(binary, benchmark_filter, out):
+    subprocess.run([binary, f"--benchmark_filter={benchmark_filter}", f"--benchmark_out={out}",
+                    "--benchmark_out_format=json"], check=True, stdout=subprocess.DEVNULL,
+                   stderr=subprocess.DEVNULL)
+
+
+def throughput(path):
+    """Bytes per second by (path, analyzer, corpus), from a benchmark JSON file."""
+    result = {}
+    with open(path) as handle:
+        for benchmark in json.load(handle)["benchmarks"]:
+            name = benchmark["name"].split("/min_time")[0]
+            result[tuple(name.split("/")[:3])] = benchmark["bytes_per_second"]
+    return result
+
+
+def geomean(values):
+    return math.exp(sum(math.log(v) for v in values) / len(values))
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__,
+                                     formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("--baseline", required=True)
+    parser.add_argument("--candidate", required=True)
+    parser.add_argument("--filter", required=True, help="--benchmark_filter for both binaries")
+    parser.add_argument("--out", required=True, help="directory for the JSON of every run")
+    parser.add_argument("--rounds", type=int, default=5)
+    args = parser.parse_args()
+    os.makedirs(args.out, exist_ok=True)
+
+    ratios = {}
+    for round_index in range(args.rounds):
+        sides = [("baseline", args.baseline), ("candidate", args.candidate)]
+        if round_index % 2 == 1:
+            sides.reverse()
+        results = {}
+        for side, binary in sides:
+            path = os.path.join(args.out, f"{side}{round_index}.json")
+            run(binary, args.filter, path)
+            results[side] = throughput(path)
+        for key, baseline in results["baseline"].items():
+            ratios.setdefault(key, []).append(results["candidate"][key] / baseline)
+
+    medians = {key: statistics.median(values) for key, values in ratios.items()}
+    analyzers = sorted({(path, analyzer) for path, analyzer, _ in medians})
+    print(f"{'benchmark':40} {'geomean':>8}  per corpus: median (min-max) of the round ratios")
+    for path, analyzer in analyzers:
+        cells = sorted((corpus, medians[(p, a, corpus)], ratios[(p, a, corpus)])
+                       for p, a, corpus in medians if (p, a) == (path, analyzer))
+        mean = geomean([median for _, median, _ in cells])
+        detail = "  ".join(f"{corpus} {median:.2f} ({min(values):.2f}-{max(values):.2f})"
+                           for corpus, median, values in cells)
+        print(f"{path + '/' + analyzer:40} {mean:8.2f}  {detail}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

be/benchmark/analyzer_benchmark.md

@@ -0,0 +1,98 @@
+<!--
+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.
+-->
+
+# Analyzer benchmark
+
+`benchmark_analyzer.hpp` measures the text analyzers on the path the SNII writer takes for one row,
+and writes a digest of every row's tokens, so a change to an analyzer can be checked for speed and
+for identical output against a baseline binary.
+
+## Build
+
+Build a release benchmark binary of only this header:
+
+```shell
+BUILD_TYPE=RELEASE DORIS_BENCHMARK_HEADER=benchmark_analyzer.hpp ./build.sh --benchmark
+```
+
+The binary is `be/output/lib/benchmark_test`; copy it away before building the other side. The
+baseline is the commit that added this benchmark, on the branch `snii-analysis-bench-baseline`:
+build it the same way from that commit.
+
+## Corpora
+
+The corpora are `<name>.rows` files, each row a little-endian u32 length and that many bytes.
+They are not in the repository; `analyzer_corpus.py` builds them and lists the sources and
+licenses in its header. The comparisons use at least the mixed corpora `http_request`,
+`otel_body`, `agent_input` and `prose_en`, and the real Chinese and Japanese text `zh_wiki` and
+`ja_wiki`:
+
+```shell
+be/benchmark/analyzer_corpus.py wikipedia zh corpus/zh_wiki.rows 8
+be/benchmark/analyzer_corpus.py wikipedia ja corpus/ja_wiki.rows 8
+be/benchmark/analyzer_corpus.py ndjson documents-171998.json request corpus/http_request.rows
+```
+
+## Environment
+
+```shell
+export DORIS_HOME=<source tree>
+export LD_LIBRARY_PATH=${JAVA_HOME}/lib/server:${LD_LIBRARY_PATH}
+export ANALYZER_BENCH_CORPUS_DIR=<directory of .rows files>
+export ANALYZER_BENCH_CORPUS_MB=8           # bytes used from each corpus file
+export ANALYZER_BENCH_DICT_DIR=<output/be/dict of a BE build>
+```
+
+The dictionary directory has the layout of a BE package: the jieba files at the top, and `icu`,
+`ik`, `pinyin` and `kuromoji` below. Without `ANALYZER_BENCH_CORPUS_DIR` the header registers
+nothing.
+
+The benchmarks are named `row/<analyzer>/<corpus>`, a new stream per row as the SNII writer does,
+and `reuse/<analyzer>/<corpus>`, one stream across the rows.
+
+## Throughput
+
+Host load moves single measurements by 10 to 20 percent, so the two binaries are compared in
+paired rounds and only ratios are reported:
+
+```shell
+be/benchmark/analyzer_ab.py --baseline baseline/benchmark_test --candidate new/benchmark_test \
+    --filter '^row/' --out ab --rounds 5
+```
+
+Each round runs both binaries, alternating which goes first; the report gives, for every analyzer,
+the median over the rounds of the candidate/baseline ratio per corpus and their geometric mean.
+Run the baseline against itself the same way first (A/A): a ratio further than 5% from 1 means the
+host is too noisy for the comparison.
+
+## Identical output
+
+With `ANALYZER_BENCH_DIGEST` set, the binary registers only the benchmark `digest`, which writes one
+line per analyzer, corpus and row: the number of terms and a digest of their bytes and positions,
+or `threw`. Run it with both binaries over the same corpora and compare the files:
+
+```shell
+ANALYZER_BENCH_DIGEST=baseline.digest baseline/benchmark_test --benchmark_filter=digest
+ANALYZER_BENCH_DIGEST=new.digest new/benchmark_test --benchmark_filter=digest
+diff baseline.digest new.digest
+```
+
+Every line of the diff is a row whose tokens changed. Unlike the golden tests in
+`be/test/storage/index/inverted/analysis/golden/`, this covers real text and the real kuromoji
+dictionary.

be/benchmark/analyzer_corpus.py

@@ -0,0 +1,149 @@
+#!/usr/bin/env python3
+# 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.
+
+"""Builds corpora for benchmark_analyzer.hpp: `<name>.rows` files, each row a little-endian u32
+length and that many bytes. The corpora are not in the repository; this script makes them.
+
+Sources and licenses of the corpora analyzer_benchmark.md uses:
+
+  zh_wiki, ja_wiki  The plain text of the first articles of the Chinese and Japanese Wikipedia
+                    CirrusSearch content dumps of 2025-12-29,
+                    https://dumps.wikimedia.org/other/cirrussearch/20251229/, cut at sentence ends
+                    into rows of up to 512 bytes. The dump is read as a stream and closed once the
+                    corpus is full, so it is never downloaded whole, and the same date gives the
+                    same corpus. Wikipedia text is licensed CC BY-SA 4.0,
+                    https://creativecommons.org/licenses/by-sa/4.0/.
+  http_request      The "request" field of documents-171998.json from the Elastic Rally http_logs
+                    track, https://github.com/elastic/rally-tracks/tree/master/http_logs, which is
+                    derived from the 1998 World Cup web server logs of the Internet Traffic Archive.
+  otel_body         The "Body" field of an OpenTelemetry log sample of the ClickHouse TextBench
+                    benchmark data.
+  agent_input       The "input" field of an AgentLogs observation file.
+  prose_en          Paragraphs of the English ClickHouse documentation (Apache License 2.0),
+                    https://github.com/ClickHouse/ClickHouse/tree/master/docs/en.
+
+Usage:
+  analyzer_corpus.py wikipedia <zh|ja> <out.rows> [megabytes, default 8]
+  analyzer_corpus.py ndjson <file> <field> <out.rows> [megabytes, default 64]
+  analyzer_corpus.py paragraphs <glob> <out.rows> [megabytes, default 64]
+"""
+
+import glob
+import gzip
+import json
+import re
+import struct
+import sys
+import urllib.request
+
+# A generic client name, as Wikimedia asks every client to send one.
+USER_AGENT = "doris-analyzer-benchmark/1.0 (corpus builder for tokenizer benchmarks)"
+DUMP_URL = ("https://dumps.wikimedia.org/other/cirrussearch/20251229/"
+            "{lang}wiki-20251229-cirrussearch-content.json.gz")
+# Sentence ends: CJK full stops and marks, or Latin ones before white space.
+SENTENCE_END = re.compile(r"(?<=[。！？])|(?<=[.!?])\s+")
+
+
+def write_rows(path, rows, limit_bytes):
+    """Writes rows until `limit_bytes` of row bytes are written; returns (rows, bytes)."""
+    count = 0
+    total = 0
+    with open(path, "wb") as out:
+        for row in rows:
+            data = row if isinstance(row, bytes) else row.encode("utf-8", "surrogatepass")
+            if not data:
+                continue
+            out.write(struct.pack("<I", len(data)))
+            out.write(data)
+            count += 1
+            total += len(data)
+            if total >= limit_bytes:
+                break
+    print(f"{path}: {count} rows, {total} bytes")
+    return count, total
+
+
+def sentence_rows(text, max_bytes=512):
+    """Packs the sentences of `text` into rows of up to `max_bytes`; a longer sentence is a row."""
+    row = ""
+    for sentence in SENTENCE_END.split(text):
+        if not sentence:
+            continue
+        if row and len((row + sentence).encode("utf-8")) > max_bytes:
+            yield row
+            row = ""
+        row += sentence
+    if row:
+        yield row
+
+
+def wikipedia_rows(lang):
+    """Rows of the article texts of a CirrusSearch dump, in the dump's order. Every other line of
+    the dump is an index action without text."""
+    request = urllib.request.Request(DUMP_URL.format(lang=lang),
+                                     headers={"User-Agent": USER_AGENT})
+    with urllib.request.urlopen(request, timeout=60) as response, \
+            gzip.GzipFile(fileobj=response) as lines:
+        for line in lines:
+            text = json.loads(line).get("text")
+            if isinstance(text, str):
+                yield from sentence_rows(text)
+
+
+def ndjson_rows(path, field):
+    with open(path, encoding="utf-8", errors="surrogateescape") as lines:
+        for line in lines:
+            try:
+                value = json.loads(line)[field]
+            except (ValueError, KeyError):
+                continue
+            if isinstance(value, str):
+                yield value
+
+
+def paragraph_rows(pattern):
+    for path in sorted(glob.glob(pattern, recursive=True)):
+        with open(path, encoding="utf-8", errors="ignore") as text:
+            for paragraph in text.read().split("\n\n"):
+                paragraph = paragraph.strip()
+                if len(paragraph) > 80 and not paragraph.startswith(("```", "|", "<", "---")):
+                    yield paragraph
+
+
+def main(argv):
+    if len(argv) < 2:
+        print(__doc__)
+        return 1
+    command = argv[1]
+    if command == "wikipedia" and len(argv) in (4, 5):
+        megabytes = int(argv[4]) if len(argv) == 5 else 8
+        write_rows(argv[3], wikipedia_rows(argv[2]), megabytes << 20)
+    elif command == "ndjson" and len(argv) in (5, 6):
+        megabytes = int(argv[5]) if len(argv) == 6 else 64
+        write_rows(argv[4], ndjson_rows(argv[2], argv[3]), megabytes << 20)
+    elif command == "paragraphs" and len(argv) in (4, 5):
+        megabytes = int(argv[4]) if len(argv) == 5 else 64
+        write_rows(argv[3], paragraph_rows(argv[2]), megabytes << 20)
+    else:
+        print(__doc__)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv))