feat(scroll): sliced PIT row extraction — one reader per primary shard, merged into one stream (#238) - #239
Open
fupelaqu wants to merge 3 commits into
Open
feat(scroll): sliced PIT row extraction — one reader per primary shard, merged into one stream (#238)#239fupelaqu wants to merge 3 commits into
fupelaqu wants to merge 3 commits into
Conversation
…nc PIT paging
Story perf-238: Sliced PIT row extraction — one reader per primary shard, merged into one stream.
Commit 1 of 2 — everything that is inert for slicing, landed and gated BEFORE the mechanism:
- `0.21.0-SNAPSHOT`; `elastic.scroll { size, max-slices }` (+ `ELASTIC_SCROLL_SIZE` /
`ELASTIC_SCROLL_MAX_SLICES`) read into `ElasticConfig.scroll: ScrollSettings` (validated,
owns the REST pool arithmetic); `ScrollConfig.maxSlices: Option[Int] = None` (None = inherit
the client ceiling, so the HOCON/env opt-out reaches explicit configs) + internal `slices`,
`ScrollConfig.DefaultMaxSlices = 8`, `ScrollMetrics.slices`, `SliceMerge` helper.
- `ScrollApi.defaultScrollConfig` (a def) / `configuredMaxSlices`, applied by `ElasticClientApi`
from HOCON and used by `scroll` / `scrollAsUnchecked` defaults, `cappedScroll` and
`scrollRows` (which clamps `maxSlices = Some(1)` on an explicit LIMIT). `scrollAs` keeps
`ScrollConfig()`: macro applications reject ANY omitted argument.
- `ScrollApi` self-type widened to `SettingsApi` (+ the five client scroll traits).
- `ElasticsearchVersion.supportsPitSlicing` (7.15 — PIT slicing is 7.15+, not 7.10);
`SettingsApi.primaryShardCount` (sum of `number_of_shards` over the concrete indices, dedup,
CCS skipped, failure as ElasticResult).
- Async PIT page fetch in es7 (`executeSearchPageAsync`) and es8/es9 (`async().search` via
`fromCompletableFuture`, which now unwraps CompletionException so retry stays alive);
`extractHitsOnly` fails the page instead of ending the stream on a parse failure; es8/es9
gain the "hits without sort values" guard; `close()` closes the async transport; REST pool
sized to the slice ceiling with ONE `setHttpClientConfigCallback` per builder.
Tests: ScrollSettingsSpec (13, both Scala legs), SettingsApiSpec +11, ElasticsearchVersionSpec
+3, SliceMergeSpec (5), JavaClientCompletionUnwrapSpec (6); Scroll/Select/Limit/WindowPartition/
GroupBy completeness + HitMetadata suites green on es6 rest+jest / es7 / es8 / es9 (5 x 28).
Refs #238
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
… shard, merged into one stream Story perf-238 — commit 2 of 2: the policy and the mechanism, landed together. Core policy (ScrollApi): - `resolveSlices`: 1 unless UsePIT, no sort (AST + JSON), ceiling > 1 and ES >= 7.15; then `max(1, min(Σ number_of_shards, ceiling))` via `SettingsApi.primaryShardCount` — guard order keeps ORDER BY / LIMIT / opt-out / ES6 / classic-scroll streams off the `_settings` round-trip; a lookup failure degrades to sequential with one WARN naming the remedy. - `scrollWithMetrics` resolves strategy + slices ONCE per stream, lazily and off the caller's thread (`lazyFutureSource` + `blocking`); every downstream read uses the resolved config; `ScrollMetrics.slices` reports the count. Explicit LIMIT clamps `maxSlices = Some(1)` BEFORE the window-enrichment branch (and in `scrollRows`, which strips the LIMIT first). Mechanism (es7 / es8 / es9 `pitSearchAfter`): - One PIT opened lazily (a cancel racing the open closes it as soon as the id arrives), N `pageSource(slice)` readers (`slice` after `withJson` + `trackTotalHits`; none when N = 1), merged page-granular by core `SliceMerge` with ONE `mapConcat`; `watchTermination` is the single PIT owner (completion / failure / cancellation); `terminated` makes in-flight steps short-circuit instead of racing the close; query parse hoisted above `openPit` (#202 rule); end-of-slice decided on RAW hits; page request built inside `Future {}` (not on the interpreter thread — that alone was a 7 % sequential-path regression). Tests: `ScrollSlicingSpec` (18), `SliceMergeSpec` (5); `SlicedScrollCompletenessSpec` (testkit template + 5 concrete clients — 1 / 3 / 6 shards + `sliced_*`, exact count + distinct ids, `maxSlices` Some(2)/Some(1), quota on the merged total, `open_contexts == 0` with a positive control, HOCON `max-slices = 1` and `size = 250` end to end); `ScrollCompletenessSpec` gains a `maxSlices = Some(1)` guard. Green on real ES 6.8 rest+jest / 7.17 / 8.18 / 9.0 (5 x 41), core 810 tests on both Scala legs. Measured (10M rows, 6 shards, 3 nodes, Flight sidecar overlay, medians of 3): sequential 34.1 s -> sliced 9.7 s wall (3.5x), ES CPU 54 -> 36 s, transport 2,293 -> 1,537 MB, sidecar memory.peak +0.0 %, `took` flat in depth, 0 search-pool rejections at k = 4 concurrent clients. Docs: scroll.md (strategy matrix, ScrollConfig/ScrollMetrics, "Sliced PIT paging"), common_principles.md (HOCON block + env vars), README scrollAs example. Refs #238 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…sliced PIT paging (#238) Story perf-238 — lead decisions 2a / 3c on the code-review findings, plus review pass #3 on this diff. ScrollApi.cachedPrimaryShardCount: a 5-minute TTL cache of the slice shard count, keyed by the sorted distinct index set, resolved under ConcurrentHashMap.compute (K concurrent cold extractions share ONE _settings round-trip; a clear() cannot be resurrected by an in-flight put; timestamp taken after the lookup). Remembered: a positive count, and a privilege failure (HTTP 401/403 — one WARN per TTL naming the privilege, DEBUG replay, cause stripped). NOT remembered: a transient failure (timeout, 503, 404, unparseable payload — WARN on every extraction, worded "retried on the next extraction") and an expression that matches no index (SettingsApi.primaryShardCount now returns the raw sum, 0 = nothing matched — a table created right after is seen at once). Expired entries are purged on a miss above 256 (keys are index SETS). invalidateShardCounts() clears everything — keys are FROM expressions, a wildcard or alias key cannot be matched by index name — and is called from createIndex (success), updateSchema, invalidateSchema (DROP TABLE, REPL refresh) and invalidateAllSchemas. Staleness is performance-only. SlicedScrollCompletenessSpec (m): the CLIENT really opens one PIT reader per resolved slice (ScrollMetrics.slices is stamped by core's policy, so the existing assertions would pass if a client ignored config.slices). A logback ListAppender on the app.softnetwork.elastic.client package logger, DEBUG for the window, asserts the distinct (slice, of) pairs — distinct because the start line is logged inside the page retry — and exactly one un-suffixed reader under maxSlices = Some(1); detach before reading, copy under the appender lock (the stream's onTerminate still logs "closing PIT" after Await.result — ConcurrentModificationException seen on ES 9.0.3); cancel on a non-logback binding; positive control so a mis-bound capture fails loudly. logback-classic is now declared on the testkit projects (it reached Compile only transitively). Tests: core ScrollSlicingSpec (+9 cache cases incl. an 8-thread cold-key race) + SettingsApiSpec green on 2.12.20 and 2.13.16 (116 each); SlicedScrollCompletenessSpec 13/13 on real ES 7.17.29 / 8.18.3 / 9.0.3, 12 + (m) cancelled on ES 6.8.23 (rest, jest); headerCheck, scalafmtCheckAll green. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes #238
What
On Elasticsearch ≥ 7.15, a row extraction with no
ORDER BYand noLIMITnow opens one PIT and readsmin(primary shards, elastic.scroll.max-slices)slices of it concurrently (default ceiling 8), merged page by page into the single stream the caller already consumes. The sequential pipeline was latency-bound (one round-trip per page, every page fanning out to every shard); this is what lets the wall clock improve when shards and nodes are added.ScrollApi.resolveSlices): 1 slice unless the strategy is PIT +search_after, no sort (AST or JSON), no explicitLIMIT, the effective ceiling is > 1 and the cluster is ≥ 7.15; otherwisemax(1, min(Σ number_of_shards of the resolved indices, ceiling)). The shard count comes fromGET <indices>/_settings(SettingsApi.primaryShardCount, wildcards/aliases summed and deduplicated), behind a 5-minute TTL cache (see below).ScrollMetrics.slicesreports the resolved count; one INFO line per sliced extraction.pitSearchAfter: lazy PIT open, N page readers (SlicedScroll/SliceBuilder(id, max)— under PIT a slice is a doc-id range, not an_idhash filter), merged by a coreSliceMerge(flatMapMerge), onewatchTerminationowner closing the PIT exactly once,terminated/outerDoneguards so a cancel never leaks the PIT. PIT page fetches are asynchronous on all three clients (no dispatcher thread blocks on the wire;CompletionExceptionunwrapped so retry stays alive; a page that cannot be parsed now FAILS the stream instead of ending it silently).elastic.scroll { size = 1000, max-slices = 8 }+ELASTIC_SCROLL_SIZE/ELASTIC_SCROLL_MAX_SLICES, validated at load;ScrollConfig.maxSlices: Option[Int](Noneinherits the client ceiling — so the HOCON/envmax-slices = 1is a real kill switch even for callers that build their ownScrollConfig);ScrollApi.defaultScrollConfig; REST pools on es7/es8/es9 sized frommax-slices.ElasticsearchVersion.supportsPitSlicing= 7.15 (PIT slicing is Support search slicing with point-in-time elastic/elasticsearch#74457, not 7.10 as the issue says). ES 7.12–7.14 keep sequential PIT paging; ES 6 / classic scroll /preferSearchAfter = false/ORDER BY/ explicitLIMITare unchanged.Measured (10M rows, 6 primary shards, 3 × ES 8.18.3 nodes, Flight sidecar overlay, env-only toggles, fresh container per arm, medians of 3 — every run returned exactly 10,000,000 rows)
memory.peakmax-slices = 1(async sequential)max-slices = 1, page 5000max-slices = 1→ default_cat/thread_pool/searchrejected = 0 on all 3 nodestook-at-depth probe (1,400 pages of 1,000): 1 ms flat for unsliced PIT and for slices 0/6 and 3/6 — the slice filter does not defeat the_docskip of #197. Default-on stands: gate (ii) passes (3.5×, memory +0.0 %, ES CPU 36 < 54), gate (iv) passes (0 rejections, p95 better); arm D's +23 % vs A is carried by the page-size knob (+18.5 % sequentially), not by slicing — lead decision recorded in the story.Commits
edb94ea5— config surface, 7.15 gate,primaryShardCount,ScrollConfig.maxSlices/slices,ScrollMetrics.slices,SliceMerge,defaultScrollConfig, async PIT paging on es7/es8/es9 (own commit, gated by all Docker suites before slicing was added).547526d5— the slice policy in core, the slicedpitSearchAfteron es7/es8/es9,SlicedScrollCompletenessSpec× 5 clients, docs, review-pass-follow elasticsearch versions #2 patches.be3bcf1e— shard-count TTL cache (privilege-only negative cache,compute-atomic, clear-all on every schema-cache write/invalidation, raw-sumprimaryShardCount),SlicedScrollCompletenessSpec(m) reader-count assertion (logback capture; aConcurrentModificationExceptionreproduced on ES 9.0.3 and fixed), logback declared on the testkit projects, review-pass-V8.x #3 patches.Tests
ScrollSlicingSpec/SettingsApiSpecpair re-run on both legs after commit 3), sql 500/500, macros 18/18,+ compile,headerCheck,scalafmtCheckAll.SlicedScrollCompletenessSpec13/13 on 6.8 (rest + jest: 12 + (m) cancelled — no PIT on ES 6) / 7.17 / 8.18 / 9.0 — 1/3/6-shard indices + a 10-shard wildcard, exact row count AND distinct ids on every run,slices == N, sequential guarantees, quota binding on the merged total, no open search context left behind (probe with a positive control), HOCON opt-out/page size through the SPI, and (m) the client really opens one PIT reader per slice; the existingScrollCompletenessSpec/GroupByCompletenessSpec/WindowPartitionCompletenessSpec/SelectCompletenessSpec/LimitCompletenessSpec/ REPL suites stayed green across the sweep (5 clients × 41); 0PIT close reported failureWARNs throughout.Behaviour changes to carry into the 0.21.0 release notes
max-slices = 8);elastic.scroll.max-slices = 1/ELASTIC_SCROLL_MAX_SLICES=1is a complete opt-out.elastic.scroll.size/elastic.scroll.max-slices(+ env), validated at load.ScrollConfiggainsmaxSlices: Option[Int] = Noneand the internalslices;ScrollMetricsgainsslices;ElasticConfiggainsscroll: ScrollSettings;ScrollApi.defaultScrollConfig/configuredMaxSlices;ElasticsearchVersion.supportsPitSlicing;SettingsApi.primaryShardCount. Source-compatible, binary-incompatible (apply/copy/unapplyarity changed) — downstream artifacts rebuild on the 0.21.0 train.scrollAsstill needs an explicit config (macro applications cannot take defaults) — passclient.defaultScrollConfig._doc-ordered); addORDER BYwhen order matters (ordered statements stay sequential). Audit of jdbc / arrow / extensions test fixtures: no at-risk test (all 1-shard).maxDocuments) are an arbitrary subset, not a stable prefix.SELECT *first-row-derived column metadata (JDBC, Arrow) may vary between runs on heterogeneous multi-shard indices.max-slices.close()also closes the async transport.view_index_metadataprivilege; without it the extraction logs one WARN per 5 min naming the remedy and pages sequentially.shardCountCacheTtlMs); privilege failures are remembered, transient ones are not;CREATE TABLE/ALTER TABLE/DROP TABLE/ REPLrefreshon the same client clear it.Follow-ups (not in this PR)
retryWithBackoff(N slices retry in lockstep on a 429/503 burst).search_afterstill turn a page parse failure into a silent end-of-stream (extractAllResults/ ES 6extractHitsOnly).🤖 Generated with Claude Code