Copilot commented on code in PR #95:
URL:
https://github.com/apache/paimon-vector-index/pull/95#discussion_r3951092938
##########
core/benches/ann_bench.rs:
##########
@@ -999,8 +999,48 @@ fn run_query_case(
("DISKANN", DiskAnnRawVectorEncoding::F16) => "f16",
_ => "none",
};
+ // The original passes above retain their first-pass I/O and recall
semantics.
+ // Optional CI timing starts after those complete sweeps have warmed each
reader.
+ let steady_min_ms: u64 = read_env("ANN_STEADY_MIN_MS", 0)?;
+ if steady_min_ms > 0 && !storage.latency.is_zero() {
+ return Err("ANN_STEADY_MIN_MS requires local_ssd_warm_cache".into());
+ }
+ let mut steady_sequential_queries = 0usize;
+ let mut steady_batch_queries = 0usize;
+ let mut steady_latencies = Vec::new();
+ let mut steady_sequential_elapsed = Duration::ZERO;
+ let mut steady_batch_elapsed = Duration::ZERO;
+ if steady_min_ms > 0 {
+ let minimum = Duration::from_millis(steady_min_ms);
+ let started = Instant::now();
+ while started.elapsed() < minimum {
+ for query in dataset.queries.chunks_exact(config.d) {
+ let query_started = Instant::now();
+ std::hint::black_box(reader.search(query, search)?);
+ steady_latencies.push(query_started.elapsed());
+ }
Review Comment:
The steady-state loop stores a latency `Duration` for every sequential query
across all iterations until `ANN_STEADY_MIN_MS` elapses. If searches are very
fast, this vector can grow large (potentially hundreds of thousands to millions
of entries), increasing memory usage and pushing extra work into the percentile
calculation. Consider bounding memory by collecting latencies for only one
sweep, sampling (e.g., every Nth query), or using a fixed-size
histogram/streaming quantile approach to estimate P95 without retaining all
measurements.
##########
tools/benchmark_pr.py:
##########
@@ -0,0 +1,382 @@
+#!/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.
+
+"""Compare two disposable checkouts using the candidate's ANN benchmark
driver."""
+
+import argparse
+import csv
+import hashlib
+import json
+import math
+import os
+from pathlib import Path
+import platform
+import shutil
+import signal
+import statistics
+import subprocess
+import time
+
+
+INDEXES = ("IVF_FLAT", "IVF_SQ", "IVF_PQ", "IVF_RQ", "DISKANN")
+WORKLOAD = {
+ "ANN_DATASET_NAME": "ci-clustered-v1",
+ "ANN_N": "10000",
+ "ANN_TRAIN_N": "4096",
+ "ANN_NQ": "2048",
+ "ANN_D": "64",
+ "ANN_K": "10",
+ "ANN_NLIST": "64",
+ "ANN_NPROBE": "8",
+ "ANN_PQ_M": "8",
+ "ANN_RQ_BITS": "4",
+ "ANN_CLUSTERS": "32",
+ "ANN_NOISE_DIMENSIONS": "64",
+ "ANN_SEED": "42",
+ "ANN_DISKANN_L_SEARCHES": "100",
+ "ANN_DISKANN_MEMORY_BUDGET_BYTES": str(256 * 1024 * 1024),
+ "ANN_DISKANN_BUILD_DISTANCE": "full_precision",
+ "ANN_DISKANN_RAW_VECTOR_ENCODING": "f32",
+ "ANN_STORAGE_CASES": "local_ssd_warm_cache",
+ "RAYON_NUM_THREADS": "2",
+ "ANN_STEADY_MIN_MS": "1000",
+}
+# These identify the workload, not measured results. Never compare unlike
cases.
+CASE_FIELDS = (
+ "dataset", "index", "storage", "n", "train_n", "nq", "d", "k", "nlist",
+ "nprobe", "pq_m", "rq_bits", "diskann_build_distance",
+ "diskann_raw_vector_encoding", "l_search", "steady_min_ms",
+)
+METRICS = (
+ ("recall_at_10", "Batch Recall@10", "recall"),
+ ("steady_sequential_qps", "Warm sequential QPS", "higher"),
+ ("steady_batch_qps", "Warm batch QPS", "higher"),
+ ("steady_sequential_p95_us", "Warm sequential P95 (µs)", "lower"),
+ ("build_ms", "Build (ms)", "lower"),
+ ("peak_rss_bytes", "Process peak RSS up to build completion (MiB)",
"lower"),
+ ("file_bytes", "Index size (MiB)", "lower"),
+)
+
+
+def command_output(command, cwd=None):
+ return subprocess.check_output(command, cwd=cwd, text=True,
timeout=30).strip()
+
+
+def run_checked(command, *, deadline, timeout, **kwargs):
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ raise TimeoutError("Benchmark total time budget exhausted; stopping
before next process")
+ limit = min(timeout, remaining)
+ process = subprocess.Popen(command, start_new_session=os.name == "posix",
**kwargs)
+ try:
+ returncode = process.wait(timeout=limit)
+ except subprocess.TimeoutExpired as error:
+ # Cargo can leave rustc children alive if only the parent is killed.
+ if os.name == "posix":
+ os.killpg(process.pid, signal.SIGKILL)
+ else:
+ process.kill()
+ process.wait()
+ raise TimeoutError(
+ f"{command[0]} exceeded {limit:.1f}s process/remaining total
budget"
+ ) from error
+ if returncode:
+ raise subprocess.CalledProcessError(returncode, command)
+
+
+def numeric_field(row, field, path, *, integer=False):
+ raw = row.get(field)
+ if raw is None or not raw.strip():
+ raise ValueError(f"{path}: missing or empty metric/parameter {field}")
+ try:
+ value = int(raw) if integer else float(raw)
+ except (ValueError, OverflowError) as error:
+ raise ValueError(f"{path}: invalid {field}={raw!r}; expected a
number") from error
+ if not math.isfinite(value) or value < 0:
+ raise ValueError(f"{path}: invalid {field}={raw!r}; expected finite
nonnegative value")
+ return value
+
+
+def read_sample(path, index):
+ with path.open(newline="") as source:
+ rows = list(csv.DictReader(source))
+ if len(rows) != 1:
+ raise ValueError(f"{path}: expected exactly one case, found
{len(rows)}")
+ row = rows[0]
+ if row.get("index") != index:
+ raise ValueError(f"{path}: expected index {index}")
+ for field in CASE_FIELDS:
+ if not row.get(field):
+ raise ValueError(f"{path}: missing workload field {field}")
+ for field, _, _ in METRICS:
+ numeric_field(row, field, path)
+ if not 0 <= float(row["recall_at_10"]) <= 1 or numeric_field(row, "nq",
path, integer=True) <= 0:
+ raise ValueError(f"{path}: invalid recall or query count")
+ if float(row["steady_sequential_qps"]) <= 0 or
float(row["steady_batch_qps"]) <= 0:
+ raise ValueError(f"{path}: throughput must be positive")
+ minimum = numeric_field(row, "steady_min_ms", path, integer=True)
+ if minimum <= 0:
+ raise ValueError(f"{path}: steady_min_ms must be positive")
+ for mode in ("sequential", "batch"):
+ elapsed = numeric_field(row, f"steady_{mode}_ms", path)
+ queries = numeric_field(row, f"steady_{mode}_queries", path,
integer=True)
+ if elapsed < minimum or queries < int(row["nq"]) or queries %
int(row["nq"]):
+ raise ValueError(f"{path}: incomplete steady_{mode} measurement")
+ return row
+
+
+def metric_value(row, field):
+ value = float(row[field])
+ if field in ("peak_rss_bytes", "file_bytes"):
+ return value / (1024 * 1024)
+ return value
+
+
+def delta_text(base, candidate, direction):
+ if direction == "recall":
+ return f"{(candidate - base) * 100:+.2f} pp"
+ if base == 0:
+ return "0.0%" if candidate == 0 else "n/a (base=0)"
+ return f"{(candidate / base - 1) * 100:+.1f}%"
+
+
+def summarize(samples, rounds):
+ results = {}
+ for index in INDEXES:
+ sides = samples[index]
+ if any(len(sides[side]) != rounds for side in ("base", "candidate")):
+ raise ValueError(f"{index}: incomplete samples")
+ cases = {
+ tuple(row[field] for field in CASE_FIELDS)
+ for rows in sides.values() for row in rows
+ }
+ if len(cases) != 1:
+ raise ValueError(f"{index}: workload differs between samples")
+ result = {}
+ for field, label, direction in METRICS:
+ entry = {"label": label, "direction": direction}
+ for side in ("base", "candidate"):
+ values = [metric_value(row, field) for row in sides[side]]
+ entry[side] = {
+ "median": statistics.median(values),
+ "min": min(values), "max": max(values), "samples": values,
+ }
+ entry["delta"] = delta_text(
+ entry["base"]["median"], entry["candidate"]["median"],
direction
+ )
+ result[field] = entry
+ results[index] = result
+ return results
+
+
+def alert_level(entry):
+ base, candidate = entry["base"]["median"], entry["candidate"]["median"]
+ if entry["direction"] == "recall":
+ loss = round((base - candidate) * 100, 6)
+ yellow, red = 1, 3
+ else:
+ loss = round((1 - candidate / base) * 100, 6) if base else 0
Review Comment:
`alert_level()` currently treats any drop vs base as a regression for all
non-recall metrics, but it ignores the `direction` value (`higher` vs `lower`).
For `lower` metrics (e.g., build time, RSS, size), regressions are increases,
not decreases, so the current formula will fail to flag worse results. Update
the computation to branch on `entry['direction']` so that `higher` flags
decreases and `lower` flags increases (while still keeping the recall
percentage-point logic).
##########
core/benches/ann_bench.rs:
##########
@@ -1096,7 +1139,9 @@ fn exact_ground_truth(dataset: &Dataset, k: usize) ->
Vec<Vec<i64>> {
.then_with(|| left.1.cmp(&right.1))
});
}
- distances.into_iter().map(|(_, row)| row).collect()
+ // Borrow the retained top-k slice: consuming the Vec can reuse its
+ // original N-vector allocation for every ground-truth row.
Review Comment:
The comment describes allocation reuse behavior that doesn’t match what this
code does: `collect()` here still allocates a fresh `Vec<i64>` for the
ground-truth row, and iterating (rather than consuming) doesn’t by itself reuse
the original `distances` allocation. Please update the comment to reflect the
actual intent/behavior (e.g., avoiding an extra move vs borrowing), or adjust
the implementation if allocation reuse is the goal.
--
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]
