[ci] Add PR versus base vector index benchmarks

.github/workflows/benchmark-comment.yml

@@ -0,0 +1,90 @@
+# 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.
+
+name: PR benchmark comment
+
+on:
+  workflow_run:
+    workflows: ['PR benchmark']
+    types: [completed]
+
+permissions:
+  actions: read
+  pull-requests: write
+
+concurrency:
+  group: benchmark-comment-${{ github.event.workflow_run.head_repository.id }}-${{ github.event.workflow_run.head_branch }}
+  cancel-in-progress: false
+
+jobs:
+  comment:
+    if: ${{ github.event.workflow_run.event == 'pull_request' && github.event.workflow_run.conclusion != 'cancelled' }}
+    runs-on: ubuntu-24.04
+    timeout-minutes: 5
+    steps:
+      # Never check out PR code, restore its caches, or execute artifact contents.
+      - name: Download this run's comment input
+        uses: actions/download-artifact@v4
+        with:
+          name: benchmark-comment-${{ github.event.workflow_run.run_attempt }}
+          run-id: ${{ github.event.workflow_run.id }}
+          github-token: ${{ github.token }}
+          path: report
+
+      - name: Create or update benchmark comment
+        uses: actions/github-script@v7
+        with:
+          script: |
+            const fs = require('fs');
+            const { owner, repo } = context.repo;
+            const run = context.payload.workflow_run;
+            if (run.path !== '.github/workflows/benchmark-pr.yml') {
+              throw new Error('Unexpected source workflow');
+            }
+            // Artifact fields are untrusted: bind the requested PR to GitHub's run data.
+            const input = JSON.parse(fs.readFileSync('report/benchmark-context.json', 'utf8'));
+            if (!Number.isSafeInteger(input.number) || input.number <= 0) {
+              throw new Error('Invalid PR number');
+            }
+            const { data: pr } = await github.rest.pulls.get({ owner, repo, pull_number: input.number });
+            if (pr.state !== 'open' || pr.base.repo.full_name !== `${owner}/${repo}`
+                || pr.head.repo?.id !== run.head_repository.id
+                || pr.head.ref !== run.head_branch || pr.head.sha !== run.head_sha
+                || input.head !== pr.head.sha || input.base !== pr.base.sha) {
+              core.info('Skipping unrelated or stale benchmark result');
+              return;
+            }
+            // A delayed delivery from an earlier rerun must not replace the latest result.
+            const { data: current } = await github.rest.actions.getWorkflowRun({ owner, repo, run_id: run.id });
+            if (current.run_attempt !== run.run_attempt) return;
+            const marker = '<!-- paimon-pr-benchmark -->';
+            const path = 'report/results/summary.md';
+            const report = fs.existsSync(path)
+              ? fs.readFileSync(path, 'utf8')
+              : 'Benchmark did not produce a report. See the workflow logs.';
+            if (Buffer.byteLength(report, 'utf8') > 60000) throw new Error('Report too large');
+            const body = `${marker}\n[Benchmark run](${run.html_url}) · ${run.conclusion} · attempt ${run.run_attempt}\n\n${report}`;
+            const comments = await github.paginate(github.rest.issues.listComments, {
+              owner, repo, issue_number: pr.number, per_page: 100,
+            });
+            const existing = comments.find(comment =>
+              comment.user.login === 'github-actions[bot]' && comment.body?.startsWith(marker));
+            if (existing) {
+              await github.rest.issues.updateComment({ owner, repo, comment_id: existing.id, body });
+            } else {
+              await github.rest.issues.createComment({ owner, repo, issue_number: pr.number, body });
+            }

.github/workflows/benchmark-pr.yml

@@ -0,0 +1,137 @@
+# 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.
+
+name: PR benchmark
+
+on:
+  pull_request:
+    branches: [main, 'release-*']
+    paths:
+      - 'core/**'
+      - 'Cargo.toml'
+      - 'Cargo.lock'
+      - '.cargo/**'
+      - 'tools/benchmark_pr.py'
+      - 'tools/tests/test_benchmark_pr.py'
+      - 'tools/tests/test_benchmark_comment.cjs'
+      - '.github/workflows/benchmark-pr.yml'
+      - '.github/workflows/benchmark-comment.yml'
+
+permissions:
+  contents: read
+
+concurrency:
+  group: benchmark-pr-${{ github.event.pull_request.number }}
+  cancel-in-progress: true
+
+jobs:
+  compare:
+    name: Compare base and PR
+    runs-on: ubuntu-24.04
+    timeout-minutes: 25
+    steps:
+      - name: Record comparison revisions
+        uses: actions/github-script@v7
+        with:
+          script: |
+            const fs = require('fs');
+            const pr = context.payload.pull_request;
+            fs.writeFileSync('benchmark-context.json', JSON.stringify({
+              number: pr.number, head: pr.head.sha, base: pr.base.sha,
+            }));
+
+      - name: Check out PR merge result
+        uses: actions/checkout@v4
+        with:
+          ref: ${{ github.sha }}
+          path: candidate
+          persist-credentials: false
+
+      - name: Check out exact base revision
+        uses: actions/checkout@v4
+        with:
+          ref: ${{ github.event.pull_request.base.sha }}
+          path: base
+          persist-credentials: false
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+
+      - name: Set up stable Rust toolchain
+        run: |
+          rustup toolchain install stable --profile minimal
+          rustup default stable
+        env:
+          RUSTUP_TOOLCHAIN: stable
+
+      - name: Cache Rust downloads
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cargo/registry
+            ~/.cargo/git
+          key: benchmark-${{ runner.os }}-rust-stable-${{ hashFiles('candidate/Cargo.lock', 'base/Cargo.lock') }}
+          restore-keys: |
+            benchmark-${{ runner.os }}-rust-stable-
+
+      - name: Test comparison and failure handling
+        run: |
+          python3 -m unittest discover -s candidate/tools/tests -p 'test_benchmark_pr.py' -v
+          node candidate/tools/tests/test_benchmark_comment.cjs
+
+      - name: Build and compare both revisions
+        # Stop the script before the job deadline, leaving time to publish failures.
+        run: python3 candidate/tools/benchmark_pr.py --base base --candidate candidate --output results --total-timeout 1200
+        env:
+          RUSTUP_TOOLCHAIN: stable
+
+      - name: Publish comparison summary
+        if: always()
+        run: |
+          if [ -f results/summary.md ]; then
+            cat results/summary.md >> "$GITHUB_STEP_SUMMARY"
+          else
+            echo 'Benchmark did not produce a report. Check the setup and build steps.' >> "$GITHUB_STEP_SUMMARY"
+          fi
+
+      - name: Upload report and raw samples
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: pr-${{ github.event.pull_request.number }}-benchmark
+          overwrite: true
+          path: |
+            results/summary.md
+            results/summary.json
+            results/metadata.json
+            results/build-*.log
+            results/build-*.jsonl
+            results/raw/
+          if-no-files-found: warn
+          retention-days: 30
+
+      - name: Upload comment input
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: benchmark-comment-${{ github.run_attempt }}
+          path: |
+            benchmark-context.json
+            results/summary.md
+          retention-days: 7

core/benches/ann_bench.rs

@@ -999,8 +999,51 @@ 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)?);
+                // Keep one complete warm sweep for P95; bound storage to nq samples.
+                if steady_sequential_queries == 0 {
+                    steady_latencies.push(query_started.elapsed());
+                }
+            }
+            steady_sequential_queries += config.nq;
+        }
+        steady_sequential_elapsed = started.elapsed();
+        let started = Instant::now();
+        while started.elapsed() < minimum {
+            std::hint::black_box(batch_reader.search_batch(&dataset.queries, config.nq, search)?);
+            steady_batch_queries += config.nq;
+        }
+        steady_batch_elapsed = started.elapsed();
+    }
+    let steady_sequential_qps = if steady_sequential_queries == 0 {
+        0.0
+    } else {
+        steady_sequential_queries as f64 / steady_sequential_elapsed.as_secs_f64()
+    };
+    let steady_batch_qps = if steady_batch_queries == 0 {
+        0.0
+    } else {
+        steady_batch_queries as f64 / steady_batch_elapsed.as_secs_f64()
+    };
     println!(
-        "{dataset},{index},{storage},{n},{train_n},{raw_dataset_bytes},{nq},{d},{k},{nlist},{nprobe},{pq_m},{rq_bits},{build_distance},{raw_vector_encoding},{l_search},{build_ms},{train_ms},{add_ms},{write_ms},{peak_rss_bytes},{optimize_ms},{optimize_rounds},{optimize_ranges},{optimize_bytes},{file_bytes},{recall:.4},{first_us},{p50_us},{p95_us},{sequential_qps:.2},{seq_rounds},{seq_ranges},{seq_bytes},{batch_ms},{batch_qps:.2},{batch_rounds},{batch_ranges},{batch_bytes},{rq_seq_scanned},{rq_seq_refined},{rq_seq_final},{rq_seq_seeded_lists},{rq_seq_parallel_list_tasks},{rq_scanned},{rq_eligible},{rq_refined},{rq_refine_ratio:.6},{rq_final},{rq_refined_coarse_lookups},{rq_extra_plane_lookups},{rq_fastscan_blocks},{rq_scalar_blocks},{rq_seeded_lists},{rq_parallel_list_tasks}",
+        "{dataset},{index},{storage},{n},{train_n},{raw_dataset_bytes},{nq},{d},{k},{nlist},{nprobe},{pq_m},{rq_bits},{build_distance},{raw_vector_encoding},{l_search},{build_ms},{train_ms},{add_ms},{write_ms},{peak_rss_bytes},{optimize_ms},{optimize_rounds},{optimize_ranges},{optimize_bytes},{file_bytes},{recall:.4},{first_us},{p50_us},{p95_us},{sequential_qps:.2},{seq_rounds},{seq_ranges},{seq_bytes},{batch_ms},{batch_qps:.2},{batch_rounds},{batch_ranges},{batch_bytes},{rq_seq_scanned},{rq_seq_refined},{rq_seq_final},{rq_seq_seeded_lists},{rq_seq_parallel_list_tasks},{rq_scanned},{rq_eligible},{rq_refined},{rq_refine_ratio:.6},{rq_final},{rq_refined_coarse_lookups},{rq_extra_plane_lookups},{rq_fastscan_blocks},{rq_scalar_blocks},{rq_seeded_lists},{rq_parallel_list_tasks},{steady_min_ms},{steady_sequential_queries},{steady_batch_queries},{steady_sequential_ms},{steady_batch_ms},{steady_sequential_qps:.2},{steady_batch_qps:.2},{steady_sequential_p95_us:.3}",
         dataset = config.dataset_name,
         index = index.name,
         storage = storage.name,
@@ -1059,6 +1102,9 @@ fn run_query_case(
         rq_scalar_blocks = rq_stats.scalar_blocks,
         rq_seeded_lists = rq_stats.seeded_lists,
         rq_parallel_list_tasks = rq_stats.parallel_list_tasks,
+        steady_sequential_ms = steady_sequential_elapsed.as_millis(),
+        steady_batch_ms = steady_batch_elapsed.as_millis(),
+        steady_sequential_p95_us = percentile(&steady_latencies, 95).as_secs_f64() * 1_000_000.0,
     );
     Ok(())
 }
@@ -1067,7 +1113,7 @@ struct CsvRow;
 
 impl CsvRow {
     fn header() -> &'static str {
-        "dataset,index,storage,n,train_n,raw_dataset_bytes,nq,d,k,nlist,nprobe,pq_m,rq_bits,diskann_build_distance,diskann_raw_vector_encoding,l_search,build_ms,train_ms,add_ms,write_ms,peak_rss_bytes,optimize_ms,optimize_pread_rounds,optimize_pread_ranges,optimize_pread_bytes,file_bytes,recall_at_10,first_query_us,p50_query_us,p95_query_us,sequential_qps,sequential_pread_rounds,sequential_pread_ranges,sequential_pread_bytes,batch_ms,batch_qps,batch_pread_rounds,batch_pread_ranges,batch_pread_bytes,rq_sequential_scanned_vectors,rq_sequential_refined_vectors,rq_sequential_final_distance_evaluations,rq_sequential_seeded_lists,rq_sequential_parallel_list_tasks,rq_scanned_vectors,rq_eligible_vectors,rq_refined_vectors,rq_refine_ratio,rq_final_distance_evaluations,rq_refined_coarse_byte_lookups,rq_extra_plane_byte_lookups,rq_fastscan_blocks,rq_scalar_blocks,rq_seeded_lists,rq_parallel_list_tasks"
+        "dataset,index,storage,n,train_n,raw_dataset_bytes,nq,d,k,nlist,nprobe,pq_m,rq_bits,diskann_build_distance,diskann_raw_vector_encoding,l_search,build_ms,train_ms,add_ms,write_ms,peak_rss_bytes,optimize_ms,optimize_pread_rounds,optimize_pread_ranges,optimize_pread_bytes,file_bytes,recall_at_10,first_query_us,p50_query_us,p95_query_us,sequential_qps,sequential_pread_rounds,sequential_pread_ranges,sequential_pread_bytes,batch_ms,batch_qps,batch_pread_rounds,batch_pread_ranges,batch_pread_bytes,rq_sequential_scanned_vectors,rq_sequential_refined_vectors,rq_sequential_final_distance_evaluations,rq_sequential_seeded_lists,rq_sequential_parallel_list_tasks,rq_scanned_vectors,rq_eligible_vectors,rq_refined_vectors,rq_refine_ratio,rq_final_distance_evaluations,rq_refined_coarse_byte_lookups,rq_extra_plane_byte_lookups,rq_fastscan_blocks,rq_scalar_blocks,rq_seeded_lists,rq_parallel_list_tasks,steady_min_ms,steady_sequential_queries,steady_batch_queries,steady_sequential_ms,steady_batch_ms,steady_sequential_qps,steady_batch_qps,steady_sequential_p95_us"
     }
 }
 
@@ -1096,7 +1142,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 to collect a fresh top-k Vec. Consuming the iterator can
+            // reuse the much larger N-vector allocation via in-place collect.
+            distances.iter().map(|&(_, row)| row).collect()
         })
         .collect()
 }