perf: parallel token-range scan pipelines, PER PARTITION LIMIT pushdown and page prefetch for CQL scans

[porunov]

Removes the producer-side ceiling of single-node OLAP scans (SchemaAction.REINDEX and other scan jobs). Previously the whole table streamed through one set of per-slice-query data pullers - a vertex mixed-index reindex used 3 puller threads total, each a single paged coordinator SELECT - so raising the scan job's thread count only added idle consumers (observed in production: 100 reindex threads, ~10 ever busy, Elasticsearch at 15% CPU, ~20k vertices/min over a 35M-vertex graph).

True parallel token-range scan (scan jobs)

• New SplittableScanStore capability: a store whose unordered scan can be split into disjoint key-space partitions that tile the key space exactly. CQLKeyColumnValueStore implements it for the Murmur3 partitioner using the existing token-range tiling (CQLTokenRangeSplitter) and the token-bounded scan statement.
• New PartitionedRowsCollector: when storage.cql.parallel-scan-token-ranges > 1, StandardScannerExecutor runs one MultiThreadsRowsCollector pipeline per range (own data pullers + own merge thread), all feeding the shared processor queue. Per-split merge is safe because each split preserves the same key-iteration order across slice queries; scan jobs do not depend on cross-key global order. Non-scan-job getKeys(SliceQuery) callers keep the back-to-back token-ordered range concatenation.
• A split failure interrupts sibling splits and fails the scan (fail-fast).

PER PARTITION LIMIT pushdown (storage.cql.scan-per-partition-limit-enabled, default true)

• Scan statements now bind the slice query's per-key limit as PER PARTITION LIMIT. The scan framework's grounding (key-existence) query has limit 1; previously every cell of every row slice streamed to the client and was discarded there, so the grounding query alone transferred the whole table including adjacency data irrelevant to the job. Wide (edge-heavy) rows benefit the most. Disable for CQL services without PER PARTITION LIMIT support (needs Apache Cassandra 3.6+/ScyllaDB).

Scan-only page size (storage.cql.scan-page-size, default 0 = storage.page-size)

• Full scans can use pages several times larger than the OLTP-oriented storage.page-size without affecting transactional reads.

Page prefetch (always on)

• CQLPagingIterator now fetches the next page while the current one is being drained (one-page lookahead). Back-pressure permits are held only while a fetch is in flight, exactly as before; the lookahead just overlaps network wait with row processing.

Scan jobs fail loudly on data-puller errors

• A puller that died on a storage error used to signal end-of-data; the merge loop treated the query as exhausted and the scan completed "successfully" with silently missing rows - a reindex could ENABLE an incomplete index. MultiThreadsRowsCollector now records puller failures and fails the scan with TemporaryBackendException after the merge loop drains.

Observability

• StandardScannerExecutor logs scan start (job, queries, processors, collector type, queue capacity), a 30s progress line (produced/processed rows + rates, failure count, row-queue fill - near-empty queue = storage-bound, near-full = processing/index-bound) and a completion summary. Per-puller counters (rows, hand-off block time) log at DEBUG. IndexRepairJob tracks mixed-index bulk flushes via custom metrics mixed-index-flushes / mixed-index-flushed-docs / mixed-index-flush-time-ms.

Benchmarks (ReindexThroughputBenchmark, new: seeds a keyspace once and times REINDEX of a fresh 5-key ES mixed index; 100k vertices with 11 properties + 3 edges each; single-node Cassandra 4.1.3 + Elasticsearch 9.3.2; threads=100, storage.page-size=200, mixed-index-batch=1000; averages of 2-3 runs):

before this change                        17.3s   (~5.8k vertices/s)
+ page prefetch                           12.1s   (1.4x)
+ PER PARTITION LIMIT pushdown             8.3s   (2.1x)
+ parallel-scan-token-ranges=8,
  scan-page-size=2000                      2.0s   (8.7x, ~50k vertices/s)

500k vertices, same shape:
reference (prefetch only)               59.5s  (~8.4k vertices/s; implies ~85s for the pre-change code)
parallel config (8 ranges, page 2000)   7.6s   (~66k vertices/s, ~11x end-to-end)

Document counts were verified exact (docUpdates == vertex count) on every configuration; the split-scan suite (CQLExternalScanTest) asserts gap- and overlap-free tiling for 1-32 splits and runs the standard scan assertions with 2 and 5 parallel ranges and with the pushdown disabled; ScanPullerFailureTest covers the fail-loudly behavior.

In production the removed bottlenecks (per-page round trips, whole-row transfer for the grounding query, single-coordinator scan) are all network/cluster-bound, so the expected gain there exceeds the local numbers: with 16 ranges even per-range throughput equal to the OLD total implies ~16x.

Claude-Session: https://claude.ai/code/session_01TuY76PEVUyZXu6MMSkpHcw

---

Thank you for contributing to JanusGraph!

In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:

For all changes:

• Is there an issue associated with this PR? Is it referenced in the commit message?
• Does your PR body contain #xyz where xyz is the issue number you are trying to resolve?
• Has your PR been rebased against the latest commit within the target branch (typically master)?
• Is your initial contribution a single, squashed commit?

For code changes:

• Have you written and/or updated unit tests to verify your changes?
• If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under ASF 2.0?
• If applicable, have you updated the LICENSE.txt file, including the main LICENSE.txt file in the root of this repository?
• If applicable, have you updated the NOTICE.txt file, including the main NOTICE.txt file found in the root of this repository?

For documentation related changes:

• Have you ensured that format looks appropriate for the output in which it is rendered?

[Copilot]

Pull request overview

Note

Copilot couldn't run its full agentic review because no GitHub Actions runner was available. Make sure your repository has a runner available to run Copilot's review, or add a copilot-setup-steps.yml file specifying one with the runs-on attribute. See the docs for more details.

This PR significantly improves JanusGraph CQL-backed OLAP scan throughput and reliability by enabling true parallel token-range scanning, pushing down per-key limits via PER PARTITION LIMIT, adding scan-only paging controls with page prefetch, and improving observability/metrics around scan progress and mixed-index flushing.

Changes:

• Add split-parallel scan capability (SplittableScanStore) and a PartitionedRowsCollector to run independent scan pipelines per token-range split.
• Add PER PARTITION LIMIT pushdown + scan-only page size, and implement one-page lookahead prefetch in the CQL paging iterator.
• Add tests for split tiling and “fail loudly” behavior; improve scan progress logging and reindex flush metrics; add a standalone reindex throughput benchmark.

Reviewed changes

Copilot reviewed 14 out of 14 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
janusgraph-test/src/test/java/org/janusgraph/diskstorage/keycolumnvalue/scan/ScanPullerFailureTest.java	New test verifying scans fail loudly when a data puller dies mid-scan
janusgraph-cql/src/test/java/org/janusgraph/diskstorage/cql/CQLExternalScanTest.java	Adds external-Cassandra scan suite coverage for split scans and PER PARTITION LIMIT toggle
janusgraph-cql/src/main/java/org/janusgraph/diskstorage/cql/CQLKeyColumnValueStore.java	Implements SplittableScanStore, scan-only paging/limit pushdown, and page prefetch
janusgraph-cql/src/main/java/org/janusgraph/diskstorage/cql/CQLConfigOptions.java	Adds config options for scan-only page size and PER PARTITION LIMIT pushdown
janusgraph-core/src/main/java/org/janusgraph/graphdb/olap/job/IndexRepairJob.java	Adds custom metrics for mixed-index bulk flush count/size/time
janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/scan/StandardScannerExecutor.java	Adds split collector selection + periodic progress logging
janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/scan/RowsCollector.java	Adds produced-row counting + optional puller progress reporting
janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/scan/PartitionedRowsCollector.java	New collector coordinating per-split scan pipelines
janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/scan/MultiThreadsRowsCollector.java	Adds split iterator injection, puller progress counters, and “fail loudly” detection
janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/SplittableScanStore.java	New API for backends that can tile unordered scans into disjoint splits
janusgraph-benchmark/src/main/java/org/janusgraph/ReindexThroughputBenchmark.java	Adds standalone benchmark to measure REINDEX throughput
docs/index-backend/elasticsearch.md	Documents tuning when storage scan is the bottleneck
docs/configs/janusgraph-cfg.md	Documents new CQL scan options
docs/changelog.md	Adds changelog entries for parallel scans, paging, limit pushdown, logging, and failure behavior

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[porunov]

Added a manually-enabled large-scale benchmark test (LargeScaleReindexBenchmarkTest, disabled by default via bench.large.enabled) in fe82a21 and ran it at full scale on a single-node Cassandra 4.1.3 + Elasticsearch 9.3.2:

Data: 10,000,000 vertices — 5 mixed-index properties + 6 non-indexed each, real vertex label, long-tailed out-degree (most vertices 0–3 edges, ~0.01% supernodes with 1000–5000; 42,124,612 edges total, ~10 GB in Cassandra).

Reindex of a fresh 5-key ES mixed index (threads=100, storage.page-size=200, mixed-index-batch-size=1000):

configuration	time	rate
single pipeline (parallel-scan-token-ranges=1, pushdown off)	1714.4 s (28.6 min)	~5.8k vertices/s
parallel scan (parallel-scan-token-ranges=8, scan-page-size=2000, pushdown on)	233.8 s (3.9 min)	~42.8k vertices/s (7.3×)

docUpdates was exactly 10,000,000 with 0 failures in both runs. Notably, the new 30-second scan progress log shows the parallel run's row queue saturated (rowQueue=60000/60000) — at this scale the bottleneck moved from the storage scan to the single-node Elasticsearch consumer side, so clustered ES deployments should scale further.

[Copilot]

Pull request overview

Copilot reviewed 17 out of 17 changed files in this pull request and generated 2 comments.

Comments suppressed due to low confidence (1)

janusgraph-cql/src/test/java/org/janusgraph/diskstorage/cql/CQLExternalScanTest.java:203

• loadScanSuiteData() hard-codes keys=1000 and columns=40, and runScanSuite() independently hard-codes the same values. Keeping these in sync is easy to miss if the test data shape changes. Use shared constants so future changes remain consistent.

    private void loadScanSuiteData() throws Exception {
        final int keys = 1000;
        final int columns = 40;
        final String[][] values = KeyValueStoreUtil.generateData(keys, columns);
        // Give every second key only half its columns, matching the in-memory scanTestWithSimpleJob.