Row extraction pages Elasticsearch sequentially and does not use shard parallelism

[fupelaqu]

Summary

Large row-returning extractions page Elasticsearch strictly sequentially, and the request shape
makes the cost grow with the shard count. On a multi-shard index the reader never uses the
parallelism the cluster is offering: neither the sidecar nor Elasticsearch is anywhere near
saturated, yet the wall clock does not improve when shards and nodes are added.

Where it is

pitSearchAfter (es8/java/.../JavaClientApi.scala:1517, and the es7/es9 equivalents) builds a
single Source.unfoldAsync: fetch a page, await it, use its sort values as search_after, fetch the
next. There is no slice and no shard preference anywhere in the client. A PIT covers the whole
index, so every page fans out to all shards and the coordinating node performs a sorted merge
before returning it.

For a 10M-row extraction at the default page size of 1,000, that is:

• 10,000 sequential round-trips, each one a fan-out + merge across every shard
• on a 6-shard index, roughly 60,000 shard-level operations for a single statement
• per-page cost that rises with shard count, because both the fan-out and the cross-node gather
  grow with it

Measurements

10M rows / 8 columns, no ORDER BY, no LIMIT; 6-shard index on a 3-node Elasticsearch cluster
(3 x 2 CPU); sidecar 4 CPU / 4 GB. Medians of 3 runs after 2 warm-ups.

metric	value
wall	34.6 s (≈ 3.46 ms per page)
sidecar CPU	19.3 s = 0.56 of 4 cores
Elasticsearch CPU	53.6 s = 1.55 of 6 cores
inter-node transport	2,293 MB

Neither side is CPU-bound: the pipeline is latency-bound, idle while waiting on round-trips.
The same statement against a 1-shard index on the same host takes essentially the same wall
(34.6 s), i.e. adding shards and nodes bought nothing — while the fan-out per page, the cluster
CPU and the cross-node transport all grew.

Proposal — one slice per shard, merged into one stream

Elasticsearch supports slicing on PIT + search_after. Open one PIT, start N readers sharing it,
each pinned to one slice, and merge them into the single Arrow/row stream the caller already
receives. No API or protocol change; callers still see one result.

The hard constraint

Slice count must equal shard count. With more slices than shards, Elasticsearch applies a
document-level hash filter and each slice re-scans its shard. Measured on this cluster: 5 slices over
a 1-shard index cost 84.9 s of cluster CPU, against 28.2 s for 5 slices over 5 shards
where each slice maps to exactly one shard. Choosing the slice count wrong turns this improvement
into a 3x regression in cluster CPU.

Expected effect (projection from the numbers above)

Pages per slice fall to ~1,667, so the Elasticsearch-side contribution drops to roughly 6 s and the
single-threaded row-to-Arrow conversion (19.3 s CPU) becomes the bottleneck — about 20 s wall.
Parallelising conversion per slice as well puts the floor near 10-12 s. Cluster CPU should also fall,
because the per-page fan-out and merge disappear.

Scope

In scope (first cut): the no-ORDER BY, no-LIMIT row-returning path — the one SearchApi
already routes through scroll (#209).

Out of scope initially, and why:

• ORDER BY — slices return independently, so global order needs a k-way merge across slices.
• explicit LIMIT / OFFSET — needs post-merge application and per-slice over-fetch; SELECT with an explicit LIMIT above index.max_result_window fails, while the same query with NO LIMIT succeeds #224's bounded
  path stays sequential.
• aggregation-shaped statements — pushed down already, unaffected.
• ES6 — uses scroll rather than PIT; sliced scroll exists there but is a separate implementation.

Risks

• Silent incompleteness is the failure mode of this whole area (PIT + search_after injects _shard_doc as primary sort, defeating doc-id skip on ES 8+ #197, GROUP BY without LIMIT silently returns only the top 10 groups (silent wrong answer) #205, Window PARTITION BY silently truncates to 10 partitions (same class as #205) #207, SELECT without LIMIT returns only 10 rows on the non-scroll search path #209 all
  returned wrong answers without raising anything). Any slicing change needs a multi-shard test
  asserting both the exact row count and the distinct id count, on real Elasticsearch.
• Memory: N concurrent readers with in-flight batches inside a 4 GB sidecar. The merge must be
  bounded and backpressured, not an unbounded buffer.
• PIT lifecycle: one PIT with N consumers must still be closed exactly once (PIT is closed twice on successful extraction — spurious 'PIT close reported failure' WARN on every clean run #202's
  watchTermination sole-owner rule), and a failing slice must fail the stream rather than truncate
  it.
• Version gates: PIT slicing needs 7.10+; the existing _shard_doc tiebreaker gate is 7.12.
• Shard-count discovery adds a metadata lookup; it can ride the existing schema cache.
• Licence quota (maxQueryResults) must be enforced on the merged total, not per slice.

Cheaper mitigation to measure first

Raising the page size (1,000 -> 5,000) cuts round-trips 5x with no architectural change and no
correctness risk beyond memory. Worth quantifying before committing to slicing, and the two compose.

Acceptance criteria

1. A no-ORDER BY extraction from an N-shard index opens N slices under one PIT and merges them.
2. Slice count is derived from the index's shard count, never configured above it.
3. Multi-shard completeness test: exact count and distinct ids, on real Elasticsearch, for
   1 / 3 / 6 shards.
4. Peak sidecar memory does not regress against the sequential path.
5. The PIT is closed exactly once on completion, failure and cancellation.
6. ORDER BY and explicit LIMIT statements continue to use the sequential path unchanged.

[fupelaqu]

Two corrections from the implementation-story research (story perf-238, ready-for-dev; target train 0.21.0):

1. PIT slicing requires Elasticsearch ≥ 7.15.0, not 7.10+. Support search slicing with point-in-time elastic/elasticsearch#74457 "Support search slicing with point-in-time" carries the labels v7.15.0 / v8.0.0-alpha1; the 7.14 Point-in-time API page has no "Search slicing" section, the 7.15 page does. The gate will be supportsPitSlicing = 7.15; supportsPit stays at 7.12 (PIT + search_after injects _shard_doc as primary sort, defeating doc-id skip on ES 8+ #197), and 7.12–7.14 keep today's sequential PIT path.

2. Under a PIT, slices are contiguous Lucene doc-id ranges, split first across shards — not the _id hash filter used by sliced scroll (the same PR also lifted index.max_slices_per_scroll for PIT). The 84.9 s vs 28.2 s cluster-CPU figure above was measured with the benchmark's sliced-scroll floor (run_es.py, _search?scroll= + slice), so it bounds scroll; for PIT the more-slices-than-shards penalty is expected to be much smaller but is unmeasured. The constraint "slice count ≤ primary shard count" is kept regardless — it is the cheap, safe rule under both models.

Also noted while scoping: there is no configuration surface for scroll today (ScrollConfig() is hard-coded at the three production call sites — gateway, licence cap, scrollRows), so the "page size 1,000 → 5,000" mitigation cannot be set on the sidecar / JDBC / REPL path without a code change. The story adds elastic.scroll { size, max-slices } (+ ELASTIC_SCROLL_SIZE / ELASTIC_SCROLL_MAX_SLICES), which also turns the before/after into an env-only A/B on one image. Default ceiling max-slices = 8 (one slice per primary shard up to the REST client's 10-connections-per-route pool, leaving headroom for PIT open/close and the settings lookup; 1 = sequential opt-out).

Measured table to follow on the same image: sequential vs sliced, page 1,000 vs 5,000 — wall, sidecar CPU + peak memory, cluster CPU, transport.

[fupelaqu]

Measurement — sliced PIT paging (PR #239)

Same setup as the 3-node probe that opened this issue: bench_events_10m (10,000,000 docs, 6 primary shards, ~974 MB) on 3 × elasticsearch:8.18.3 (2 CPU / 4 GB / 2 GB heap each), read through the Flight SQL sidecar (4 CPU / 4 GB, -Xmx2g, ARROW_BATCH_SIZE=1000, log level WARN) by run_flight.py S1 (SELECT … FROM bench_events_10m, no ORDER BY, no LIMIT). One overlay image (released 0.2.5.1 sidecar + the branch's core/client jars), arms toggled by environment only (ELASTIC_SCROLL_MAX_SLICES, ELASTIC_SCROLL_SIZE), a fresh sidecar container per arm so cgroup memory.peak covers that arm alone, 2 warm-ups + 3 measured runs, exact row-count assertion armed. Medians of 3; every run returned exactly 10,000,000 rows.

Arm	wall (s)	sidecar CPU (s)	sidecar memory.peak	ES CPU Σ 3 nodes (s)	transport (MB)
A′ released 0.2.5.1 — blocking sequential, page 1000	34.12	20.01	1,764 MiB	54.06	2,293
A2 branch, max-slices = 1 — async sequential, page 1000	34.92 (+2.3 %)	21.07 (+5.3 %)	1,769 MiB (+0.3 %)	53.73 (−0.6 %)	2,293
B2 branch, default — sliced × 6, page 1000	9.71 (3.5× vs A′, 3.6× vs A2)	23.68	1,770 MiB (+0.0 %)	36.12	1,537 (−33 %)
C branch, max-slices = 1, page 5000	27.31	21.22	2,086 MiB (+18.5 %)	36.86	2,215
D branch, sliced × 6, page 5000	8.58	25.62	2,167 MiB (+23 % vs A, +3.9 % vs C)	28.49	1,478
E-A k = 4 concurrent S1 clients, max-slices = 1	4/4 complete, per-client walls 48.3–48.4	—	1,789 MiB	_cat/thread_pool/search rejected = 0 on all 3 nodes	—
E-B k = 4 concurrent S1 clients, default	4/4 complete, per-client walls 34.1	—	1,804 MiB (+0.8 %)	rejected = 0	—

took at depth (the #197 method, 1,400 pages of 1,000): unsliced PIT took = 1 ms at pages 10 / 100 / 300 / 600 / 1000 / 1400; slice 0/6 and slice 3/6 = 1 ms at every mark — flat in depth, so the PIT slice filter does not defeat the _doc doc-id skip (#197 stays fixed).

Reading it

• The win is the one the issue predicted: the sequential path was latency-bound (34.6 s on 6 shards/3 nodes = 34.6 s on 1 shard); slicing makes the wall track the shard count — 34.1 s → 9.7 s, with less ES CPU (54 → 36 s: a slice request touches one shard, no coordinating merge across nodes) and 33 % less transport.
• Memory: the default (B2) is flat. The only arm that costs memory is the page-size knob (C, +18.5 % at 5,000 rows/page); slicing on top of it adds 3.9 %.
• The async move itself (A2 vs A′) costs ~0.1 ms per page of machinery: +2.3 % wall (within the session's own A′ drift of +3.8 %), +5 % sidecar CPU, +0.3 % memory — disclosed, not a behaviour change. The first async build was 7.5 % slower because the per-page request build (withJson re-parse) ran on the stream's interpreter thread; moved back onto a dispatcher thread.
• Concurrency (E): four simultaneous extractions, 0 search-pool rejections, p95 48.4 → 34.1 s.
• Predictions scored (recorded before the run): B was predicted 20–24 s → measured 10 s (miss on the good side — the merged stream keeps the row→Arrow conversion parallel); B's ES CPU < A ✓; memory ≤ +10 % ✓; transport predicted ~unchanged → −33 %; C ✓ 25–28 s; D ≈ B ✓.

Two corrections to the issue text, for the record: PIT slicing needs ES ≥ 7.15 (elastic/elasticsearch#74457), gated as supportsPitSlicing; and under PIT a slice is a doc-id range (the 84.9 s _id-hash penalty measured earlier was on sliced scroll, not sliced PIT — the took probe above confirms no per-page filter cost).

Default-on (max-slices = 8, capped by the primary shard count) ships in #239; elastic.scroll.max-slices = 1 / ELASTIC_SCROLL_MAX_SLICES=1 is the opt-out.