Repository navigation
Conversation
Implements Phase 3a of SCAN_PREFETCH_DESIGN.md: a default-off,
background prefetcher that warms the object-store data cache ahead of a
native scan's demand reads, removing the cold-scan first-read penalty by
overlapping object-store I/O with decode. Correctness never depends on
it — it writes only through the same cache/fetch path demand reads use.
native/block-cache:
- PrefetchLedger: per-plan credit pool (charge/release/refund) bounding
in-flight + fetched-but-unconsumed prefetched bytes (§2.4)
- Intent-tagged blocks; first demand hit consumes (releases credit, keeps
visited, stays single-pass so SSD admission is unaffected — §2.5)
- BlockCache::prefetch_ranges: single-flight composition with demand,
charge-before-claim (no in-flight held while parked on budget), a
cancellation drop-guard, and a process-wide fetch semaphore
- Waiter retry-once + a final cache re-probe so a demand read never errors
when a racing reader has since cached the block (§2.6)
- prefetch_* metrics
native/object-store-cache:
- Inherent CachingObjectStore::prefetch_ranges wrapper (not on the trait)
native/core:
- execution/prefetch.rs ScanPrefetcher: footer parse -> split-range +
best-effort stats row-group pruning -> projected leaf column-chunk
ranges -> paced fetch; CSV whole-split; encrypted = footer-only (§2.3)
- Planner accumulates one PrefetchSpec per Parquet/CSV scan; jni_api spawns
one task per scan after create_plan and cancels cooperatively in
releasePlan (§2.2)
- Credential-free cache namespace so STS/IRSA rotation keeps cache hits
(§4.3); shared store-identity computation
- getDataCacheStats extended with prefetch counters
Scala:
- CometConf: prefetch.{enabled,aheadBudget,maxConcurrentRequests,
filterAware}, force-added to the native config map (§2.8)
Default off (spark.comet.scan.dataCache.prefetch.enabled=false); requires
the data cache. Phases 3b (host warm-set), 3c, 4 (Iceberg), 5 (Delta)
follow separately.
Co-Authored-By: Claude Opus 4.8 <[email protected]>
Closes the Phase 3a Rust test gaps: Prefetcher-unit (compute_ranges): - split-range pruning matches the opener's first-page rule - statistics pruning drops a row group with filterAware=true, keeps both with filterAware=false - nested struct projection maps to the right leaf column chunks - encrypted spec degrades to footer-only - CSV spec covers exactly the split (bounded / whole-file / empty) - footer-parse failure returns Err and run() counts prefetch_files_skipped Integration (ProbeStore counting/delaying/failing inner store): - prefetch drives concurrent upstream requests (max in-flight > 1 — the apache#3817 single-outstanding-I/O fix) - a second scan of the warmed ranges issues zero upstream requests - an injected upstream failure is counted (prefetch_errors) while the demand read still succeeds (errors are never cached) Refactors the existing warming test onto shared helpers (write_parquet / store_over / parquet_spec / lcg). All native gates stay green. Co-Authored-By: Claude Opus 4.8 <[email protected]>
Adds PrefetchFromS3Suite (extends the existing MinIO testcontainers harness) exercising the full JVM -> JNI -> native prefetch path over a remote (s3a) store — the scheme the data cache actually wraps: - prefetch ON: a native Comet scan warms the cache ahead of the decoder; getDataCacheStats shows bytes_fetched > 0 and blocks_consumed > 0 (observed 15 MiB fetched, all 15 blocks consumed, 0 wasted/errors), results byte-identical to the expected aggregate. - prefetch OFF (probe): the same cold scan still fetches through the cache but issues zero prefetch requests / bytes — proving the prefetch counters are attributable to the prefetcher, not the cache. Also folds in a spotless reformat of the CometConf prefetch entries. Co-Authored-By: Claude Opus 4.8 <[email protected]>
The demand path (`fill_pass`) fetched coalesced runs in a sequential loop, so with the data cache in the path a scan's independent column-chunk fetches were serialized — a regression versus vanilla `object_store::get_ranges`, which coalesces and fetches a request's ranges in parallel. Drive the runs through `buffer_unordered` (cap DEMAND_FETCH_CONCURRENCY) and publish each as it lands, so waiters still wake promptly and the mutations stay single-threaded. Independent of prefetch: this restores parallelism the cache dropped. With it, the scan prefetcher's contribution is purely fetch<->decode overlap rather than also compensating for a serial demand path. Co-Authored-By: Claude Opus 4.8 <[email protected]>
Adds a cache+prefetch cold-scan column beside the existing cold/warm arms (§5/§6). Per scenario it now reports Cold (cache-only), Cold+PF (cache+prefetch, same ColdFetch bytes), PF-Speedup, Warm, and prefetch Coverage (consumed / (consumed+wasted)). Because the demand path now parallelizes coalesced fetches, both cold arms parallelize, so Cold vs Cold+PF isolates the fetch<->decode overlap specifically — the number that classifies a workload: Cold+PF ≈ Cold => bandwidth-bound (no overlap to reclaim); large PF-Speedup => fetch<->decode-serialized (decode/latency-bound). Prefetch is toggled per-query via the SQLConf (read per-plan). Runs against the same env-configured s3a base path as the rest of the benchmark. Co-Authored-By: Claude Opus 4.8 <[email protected]>
|
Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days. |
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
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.
Implements Phase 3a (the default-off MVP) of the scan-prefetch design. Stacks on the object-store data cache + SSD tier (#16 /
feat/object-store-cache-ssd— the PR base).Design doc:
docs/design/SCAN_PREFETCH_DESIGN.mdon the reference-onlydocs/scan-prefetch-designbranch (never merged; cited by section below). Companion toOBJECT_STORE_CACHE_DESIGN.mdand the parent proposal apache#4695.What it does
A background prefetcher runs entirely on Comet's tokio runtime (never the Spark task thread) and warms the block cache ahead of a native scan's demand reads, overlapping object-store I/O with decode. Correctness never depends on it — every byte it writes goes through the same cache/fetch path demand reads use; a missing, failed, or cancelled prefetch only costs latency.
Layers
native/block-cache—PrefetchLedgercredit pool (charge/release/refund, §2.4); intent-tagged blocks whose first demand hit consumes without touching the SSD single-pass gate (§2.5);prefetch_rangeswith single-flight composition, charge-before-claim (no in-flight held while parked on the budget), a cancellation drop-guard, and a process-wide fetch semaphore; waiter-retry-once + a final cache re-probe so a demand read never errors when a racing reader has since cached the block (§2.6);prefetch_*metrics.native/object-store-cache— inherentCachingObjectStore::prefetch_ranges(deliberately not on theObjectStoretrait).native/core—execution/prefetch.rsScanPrefetcher(footer → split-range + best-effort stats row-group pruning → projected leaf column-chunk ranges → paced fetch; CSV whole-split; encrypted = footer-only, §2.3); planner spec accumulation +jni_apispawn/cooperative-cancel lifecycle (§2.2); credential-free cache namespace so STS/IRSA rotation keeps cache hits (§4.3);getDataCacheStatsextended with prefetch counters.CometConfprefetch.{enabled,aheadBudget,maxConcurrentRequests,filterAware}, force-added to the native config map (§2.8).Config
Default off (
spark.comet.scan.dataCache.prefetch.enabled=false); requiresspark.comet.scan.dataCache.enabled(the cache is the prefetch buffer; warn-and-ignored otherwise).Testing
native/block-cache: ledger charge/await/release, tag lifecycle + single-pass SSD gate, waste-on-eviction, waiter-retry-once (owner error and drop), cancellation leak-freedom.native/core: parquet range computation (projected-chunk selection) + cache warming + consumption; credential-free namespace survives a simulated credential rotation.clippy -D warnings, cargo-machete); Scala compiles (spark-3.5).Reviewed
Two review passes (inline + workflow-backed). Notable catch:
aheadBudgetwas initiallyByteUnit.MiB, which would have serialized the MiB count and been read natively as bytes — collapsing the budget; fixed toByteUnit.BYTE. Also fixed via review: a demand-read error when a racing reader had cached the block (final re-probe), and an in-flight-entry leak on prefetch cancellation (drop guard).Not included
Phase 3b (host warm-set), 3c (adaptive throttle), 4 (Iceberg), 5 (Delta) — separate follow-ups. The ledger + paced fetcher are format/storage-neutral so those add only front-ends.
🤖 Generated with Claude Code