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

docs/changelog.md

@@ -178,11 +178,63 @@ token-bounded queries instead of a single coordinator-funneled scan:
 storage.cql.parallel-scan-token-ranges=1
 ```
 The default value `1` preserves the previous single-query behavior. When set above `1` and the Murmur3
-partitioner is in use, the scan is split into that many disjoint token ranges streamed back in token
-order. Note: in this release the ranges are consumed back-to-back by a single scan thread (only the
-first page of each range is prefetched concurrently), so it replaces one unbounded coordinator scan
-with several bounded ones rather than scanning all ranges fully in parallel; setting very high values
-can overload the cluster. The option is ignored for non-Murmur3 partitioners.
+partitioner is in use, scan jobs (reindex and other jobs running through the scan-job framework) drain
+each token range on its own row-collection pipeline — its own data-puller threads and merge thread —
+so the storage scan runs fully in parallel across ranges and producer throughput scales with the range
+count until the cluster saturates. Non-scan-job callers of a whole-table scan receive the ranges as
+token-bounded queries streamed back-to-back in token order (bounded coordinator scans, without extra
+parallelism). Each range adds concurrent scan queries against the cluster (one per scan-job query,
+times the number of ranges), so very high values can overload the cluster; a small multiple of the
+cluster's node count is a sensible starting point. The option is ignored for non-Murmur3 partitioners.
+
+##### CQL scan-only page size
+
+Full-table scans can use their own CQL page size instead of the OLTP-oriented `storage.page-size`:
+```
+storage.cql.scan-page-size=0
+```
+The default `0` keeps using `storage.page-size`. Since a full scan streams many rows per request, a
+page size several times larger than the OLTP page size usually cuts scan round trips (and total scan
+time) substantially; a few thousand rows per page is a reasonable value. Scan pages are additionally
+fetched with a one-page lookahead, overlapping the network wait of the next page with client-side
+processing of the current one — this pipelining is always on and needs no configuration.
+
+##### PER PARTITION LIMIT pushdown for CQL scans
+
+Scan queries now push their per-key entry limit into the CQL query as `PER PARTITION LIMIT`
+(enabled by default):
+```
+storage.cql.scan-per-partition-limit-enabled=true
+```
+Scan jobs issue a grounding (key-existence) query with a per-key limit of 1; previously every cell of
+every row slice was streamed to the client and discarded there, so the grounding query alone
+transferred the whole table — including adjacency data irrelevant to the job. With the pushdown the
+server stops after the per-key limit (one cell per key for the grounding query), which substantially
+reduces scan time and network transfer on graphs with wide rows (many edges or properties per vertex).
+Requires `PER PARTITION LIMIT` support in the backend (Apache Cassandra 3.6+, ScyllaDB). A
+CQL-compatible service that rejects the clause (e.g. Amazon Keyspaces) automatically falls back to
+plain scan statements — store open logs a warning and continues instead of failing; set the option
+to `false` to skip the attempt entirely.
+
+##### Scan jobs fail loudly on data-puller errors
+
+Previously, when an internal scan data-puller thread died on a storage error, the scan completed
+"successfully" with silently missing rows — for a reindex this could ENABLE an incomplete index. A
+scan job now fails with a `TemporaryBackendException` describing the failed puller instead of
+returning a partial result, and it fails fast: the error surfaces as soon as the dead puller's
+end-of-data marker is observed rather than after the remaining key space has been streamed and
+discarded.
+
+##### Scan progress logging
+
+`StandardScannerExecutor` now logs a start line (job, query count, processor count, collector type,
+queue capacity), a progress line every 30 seconds (rows produced by the storage scan, rows processed
+by workers, current rates, row-queue fill) and a completion summary (total rows, elapsed, average
+rate). The row-queue fill discriminates the bottleneck at a glance: a near-empty queue means the scan
+is storage-bound, a near-full queue means processing/index writes are the bottleneck. Per-puller
+counters are logged at `DEBUG` level, and mixed-index reindex jobs additionally report bulk-flush
+count/size/time through the custom scan metrics `mixed-index-flushes`, `mixed-index-flushed-docs` and
+`mixed-index-flush-time-ms`.
 
 ##### Whole-row deletion on vertex removal (super-node tombstone reduction)
 

docs/configs/janusgraph-cfg.md

@@ -498,7 +498,7 @@ CQL storage backend options
 | storage.cql.metadata-schema-enabled | Whether schema metadata is enabled. | Boolean | (no default value) | MASKABLE |
 | storage.cql.metadata-token-map-enabled | Whether token metadata is enabled. If disabled, partitioner-name must be provided. | Boolean | (no default value) | MASKABLE |
 | storage.cql.only-use-local-consistency-for-system-operations | True to prevent any system queries from using QUORUM consistency and always use LOCAL_QUORUM instead | Boolean | false | MASKABLE |
-| storage.cql.parallel-scan-token-ranges | Number of disjoint Murmur3 token ranges a full-table scan (e.g. a mixed-index reindex or other OLAP job) is split into. With the default value of 1 the full scan is issued as a single query that funnels the whole table through one coordinator. When set above 1 and the Murmur3 partitioner is in use, the scan is split into this many token-bounded queries that are streamed back in token order. NOTE: the ranges are currently consumed back-to-back by a single scan thread (only the first page of each range is prefetched concurrently, at construction), so this replaces one unbounded coordinator scan with several bounded ones rather than scanning all ranges fully in parallel. Values that are too high can overload the cluster. | Integer | 1 | MASKABLE |
+| storage.cql.parallel-scan-token-ranges | Number of disjoint Murmur3 token ranges a full-table scan (e.g. a mixed-index reindex or other OLAP job) is split into. With the default value of 1 the full scan is issued as a single query that funnels the whole table through one coordinator. When set above 1 and the Murmur3 partitioner is in use: scan jobs (reindex and other OLAP jobs run through the scan-job framework) drain each token range on its own set of threads, scanning all ranges fully in parallel; other callers of a whole-table scan receive the ranges as token-bounded queries streamed back-to-back in token order (bounded coordinator scans, not extra parallelism). Each range adds concurrent scan queries against the cluster (one per scan-job query, times the number of ranges), so values that are too high can overload the cluster; a small multiple of the cluster's node count is a sensible starting point. | Integer | 1 | MASKABLE |
 | storage.cql.partitioner-name | The name of Cassandra cluster's partitioner. It will be retrieved by client if not provided. If provided, it must match the cluster's partitioner name. It can be the full class name such as `org.apache.cassandra.dht.ByteOrderedPartitioner` or the simple name such as `ByteOrderedPartitioner` | String | (no default value) | MASKABLE |
 | storage.cql.protocol-version | The protocol version used to connect to the Cassandra database.  If no value is supplied then the driver will negotiate with the server. | Integer | 0 | LOCAL |
 | storage.cql.read-consistency-level | The consistency level of read operations against Cassandra | String | QUORUM | MASKABLE |
@@ -507,6 +507,8 @@ CQL storage backend options
 | storage.cql.replication-strategy-class | The replication strategy to use for JanusGraph keyspace. Available strategies: SimpleStrategy,NetworkTopologyStrategy. | String | SimpleStrategy | FIXED |
 | storage.cql.replication-strategy-options | Replication strategy options, e.g. factor or replicas per datacenter.  This list is interpreted as a map.  It must have an even number of elements in [key,val,key,val,...] form. This options is used when storage.cql.replication-strategy-class is set to NetworkTopologyStrategy. `replication_factor` can be used to specify a replication factor. | String[] | (no default value) | FIXED |
 | storage.cql.request-timeout | Timeout for CQL requests in milliseconds. See DataStax Java Driver option `basic.request.timeout` for more information. | Long | 12000 | MASKABLE |
+| storage.cql.scan-page-size | Page size (CQL rows per backend round trip) used for scan queries: whole-table and token-range scans on the Murmur3 partitioner as well as key-range scans on ordered partitioners - i.e. every getKeys-style key iteration, which is the path OLAP jobs (such as a mixed-index reindex) read through. Transactional slice reads keep using storage.page-size. When 0 or negative, storage.page-size is used for scans as well. A scan streams many rows per request, so a page size several times larger than the OLTP-oriented storage.page-size usually cuts scan round trips (and total scan time) substantially at the cost of slightly larger response payloads held in memory per scan thread. | Integer | 0 | MASKABLE |
+| storage.cql.scan-per-partition-limit-enabled | Push each scan query's per-key entry limit into the CQL full-scan queries as PER PARTITION LIMIT. Scan jobs (e.g. reindex) issue a grounding (key-existence) query with a per-key limit of 1; without the pushdown every cell of every row slice is streamed to the client and discarded there, so the grounding query alone transfers the whole table (including adjacency data irrelevant to the job). With the pushdown the server stops after the per-key limit, e.g. one cell per key for the grounding query. Requires PER PARTITION LIMIT support in the backend (Apache Cassandra 3.6+, ScyllaDB); a CQL-compatible service that rejects the clause (e.g. Amazon Keyspaces) automatically falls back to plain scan statements with a warning at store open, so leaving the option enabled is safe everywhere; set it to false to skip the attempt. | Boolean | true | MASKABLE |
 | storage.cql.session-leak-threshold | The maximum number of live sessions that are allowed to coexist in a given VM until the warning starts to log for every new session. If the value is less than or equal to 0, the feature is disabled: no warning will be issued. See DataStax Java Driver option `advanced.session-leak.threshold` for more information. | Integer | (no default value) | MASKABLE |
 | storage.cql.session-name | Default name for the Cassandra session | String | JanusGraph Session | MASKABLE |
 | storage.cql.speculative-retry | The speculative retry policy. One of: NONE, ALWAYS, <X>percentile, <N>ms. | String | (no default value) | FIXED |

docs/index-backend/elasticsearch.md

@@ -389,6 +389,51 @@ is reached, so three settings determine reindex throughput:
     keep `bulk-refresh = false` and rely on the index becoming searchable at the
     next periodic refresh once the job completes.
 
+#### When the storage scan, not Elasticsearch, is the ceiling
+
+On the CQL backend the reindex pipeline is often **storage-scan-bound**, not
+Elasticsearch-bound: the scan job streams the whole table through a small,
+fixed number of scan queries (one per slice query — three for a vertex mixed
+index), and the reindex worker threads spend most of their time waiting for
+rows. The telltale signs are low Elasticsearch CPU, only a handful of reindex
+workers reporting activity regardless of the configured thread count, and the
+scan progress log (`StandardScannerExecutor`) showing a near-empty row queue.
+
+Three CQL options remove that ceiling:
+
+```properties
+# Scan each Murmur3 token range on its own pipeline (true parallel scan).
+storage.cql.parallel-scan-token-ranges = 8
+
+# Bigger pages for full scans only (OLTP reads keep storage.page-size).
+storage.cql.scan-page-size = 2000
+
+# Stop scan queries server-side at each row's per-key limit (default true).
+storage.cql.scan-per-partition-limit-enabled = true
+```
+
+-   **`parallel-scan-token-ranges`** splits the scan into disjoint token
+    ranges, each drained by its own data-puller threads and merged
+    independently — producer throughput scales roughly linearly until the
+    Cassandra cluster saturates. Start with a small multiple of the cluster's
+    node count and watch node load.
+-   **`scan-page-size`** decouples the scan fetch size from the OLTP-oriented
+    `storage.page-size`; a few thousand rows per page is usually a safe and
+    substantial improvement. Pages are additionally prefetched one ahead, so
+    the network wait overlaps row processing.
+-   **`scan-per-partition-limit-enabled`** pushes each scan query's per-key
+    limit into the query as `PER PARTITION LIMIT`. The scan's key-existence
+    (grounding) query has a per-key limit of 1, so on graphs with wide rows
+    (many edges or properties per vertex) this stops the grounding query from
+    streaming the entire table to the client just to prove each key exists.
+    On a CQL-compatible service without `PER PARTITION LIMIT` support the
+    pushdown auto-disables with a warning at store open.
+
+The scan-side and Elasticsearch-side settings compose: once the storage scan is
+parallel, raise the reindex thread count and, if Elasticsearch becomes the new
+bottleneck, scale its side (shards, replicas during rebuild, refresh interval)
+as described above.
+
 ### Further Reading
 
 -   Please refer to the [Elasticsearch homepage](https://www.elastic.co)

janusgraph-benchmark/pom.xml

@@ -73,6 +73,18 @@
             <version>${jna.version}</version>
         </dependency>
 
+        <!-- Manually-enabled large-scale benchmark tests -->
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
+
         <!-- needed by Cassandra ccm -->
         <dependency>
             <groupId>org.apache.cassandra</groupId>