Skip to content

feat: asynchronous scan prefetch (Phase 3a MVP) - #17

Closed
schenksj wants to merge 5 commits into
feat/object-store-cache-ssdfrom
feat/scan-prefetch
Closed

schenksj wants to merge 5 commits into
feat/object-store-cache-ssdfrom
feat/scan-prefetch

Conversation

@schenksj

@schenksj schenksj commented Jul 5, 2026 •

Copy link
Copy Markdown
Owner

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.md on the reference-only docs/scan-prefetch-design branch (never merged; cited by section below). Companion to OBJECT_STORE_CACHE_DESIGN.md and 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 — PrefetchLedger credit pool (charge/release/refund, §2.4); intent-tagged blocks whose first demand hit consumes without touching the SSD single-pass gate (§2.5); prefetch_ranges with 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 — inherent CachingObjectStore::prefetch_ranges (deliberately not on the ObjectStore trait).
  • native/core — execution/prefetch.rs ScanPrefetcher (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_api spawn/cooperative-cancel lifecycle (§2.2); credential-free cache namespace so STS/IRSA rotation keeps cache hits (§4.3); getDataCacheStats extended with prefetch counters.
  • Scala — CometConf prefetch.{enabled,aheadBudget,maxConcurrentRequests,filterAware}, force-added to the native config map (§2.8).

Config

Default off (spark.comet.scan.dataCache.prefetch.enabled=false); requires spark.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.
  • All native CI gates pass (fmt, clippy -D warnings, cargo-machete); Scala compiles (spark-3.5).

Reviewed

Two review passes (inline + workflow-backed). Notable catch: aheadBudget was initially ByteUnit.MiB, which would have serialized the MiB count and been read natively as bytes — collapsing the budget; fixed to ByteUnit.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

schenksj and others added 5 commits July 5, 2026 16:17
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]>
@github-actions

github-actions Bot commented Sep 4, 2026

Copy link
Copy Markdown

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.

@github-actions github-actions Bot added the Stale label Sep 4, 2026
@github-actions github-actions Bot closed this Sep 11, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant