[Proposal] Scan I/O acceleration: node-local fragment cache, asynchronous prefetch, and cache-affinity scheduling

[schenksj]

Disclosure: this proposal was drafted with the help of an AI assistant. I have reviewed it, understand the proposed approach end-to-end, and stand behind it; a more detailed design document and a prototype direction exist to back it up. Sharing as a high-level proposal first, per the project's preference for a clear problem statement over an AI-generated code dump.

What is the problem the feature request solves?

When Comet reads Parquet (including via Delta and Iceberg native scans) from cloud object storage (S3/ADLS/GCS), I/O is the dominant cost for many workloads, and Comet currently leaves two well-known wins on the table:

1. No data caching. Comet does not cache file data across queries. Every scan re-fetches the same bytes from object storage, even for repeated/iterative workloads (BI dashboards, repeated MERGE/UPDATE source reads, re-scans within a session). The only caching today is per-scan Parquet footer metadata, which is in-memory and not reused across queries.

2. No read-ahead. Reads are strictly demand-driven — the decoder blocks on object-store latency rather than overlapping I/O with compute. (Notably, vanilla Spark does not prefetch Parquet either, so this is a net-new capability rather than parity work.)

These are the same two levers that make commercial engines substantially faster on the same data and hardware (e.g. Databricks markets analogous capabilities as Predictive I/O and the disk cache). Comet has the right architecture to add them in a clean, format-agnostic way.

Describe the potential solution

Add an opt-in "Scan I/O Acceleration" capability to Comet's native scans, composed of three cooperating pieces:

1. Node-local fragment cache — a bounded, persistent cache of raw file fragments on local disk/SSD, shared across queries on an executor. Because data files in Parquet/Delta/Iceberg are immutable, caching is coherence-free (no invalidation protocol needed).

2. Asynchronous prefetch — read-ahead that overlaps object-store I/O with decode. Where the query's predicate/projection/limit are known, prefetch is filter-aware (only fetch what the scan will actually read); otherwise it falls back to simple sequential read-ahead. Prefetch is budget-bounded and provides natural backpressure (e.g. a satisfied LIMIT stops it).

3. Cache-affinity scheduling — to make a node-local cache effective on cloud storage (where there are no block locations), tasks must prefer the executor/node that already holds the relevant data. Comet's native scans already use a Comet-owned RDD, so this can be implemented once and apply uniformly across formats.

Design intent:

• Format coverage: Parquet, Delta (native/kernel-read), and Iceberg (native). The bulk of the mechanism lives at the shared I/O layer, so most of it is implemented once and reused.
• Opt-in and safe: disabled by default, enabled per-scan via config. Correctness never depends on the cache or on locality — a cache miss or a busy/dead preferred host only affects latency, never results.
• Incremental: the three pieces deliver value independently and can land as a sequence of small, reviewable PRs. A reasonable order is: fragment cache → sequential prefetch → filter-aware prefetch → cache-affinity scheduling, with optional cross-cluster pre-warming later. Iceberg's filter-aware prefetch may depend on upstream iceberg-rust work and can come last.

This issue is intended as a proposal at a high level to gauge interest and direction. A detailed design document (architecture, injection points, scheduling model, configuration surface, metrics, and a full test/benchmark plan) is available and can be shared as a follow-up design doc or discussion if the community is interested.

Additional context

• Why Comet is well-positioned: Comet already owns the native scan path and its scan RDD for Parquet, Delta, and Iceberg, and reads (except Iceberg) already converge on a single object-store I/O layer — so a shared cache + prefetch layer can be injected at very few points and cover all three formats. Cache-affinity scheduling fits Comet's existing plan-rewrite model (a driver-side scheduling hint), with the actual caching/prefetch living below the scan at the I/O layer.
• Prior art / inspiration: the asynchronous-prefetch scheduling model can follow the well-tested bounded/credit-based design of Spark's shuffle block fetcher; node-local cache-affinity via preferredLocations is a proven pattern for cloud-backed scans.
• Out of scope (initially): caching decoded/Arrow data (we cache raw file bytes), any cross-node distributed cache or coherency protocol (unnecessary given immutable files), and the non-native fallback scan path.
• Open questions for discussion: appetite for an in-tree on-disk cache and any new dependency it implies; default block/cache sizing; interaction with dynamic allocation; and whether to start with a "buy" (existing caching library) vs. "build" cache backend.

Related issues

• [EPIC] native_datafusion scan improvements #3736 ([EPIC] native_datafusion scan improvements) — this proposal could live under, or alongside, that umbrella.
• native_datafusion doesn't use all available parallelism for scan #3817 (native_datafusion doesn't use all available parallelism for scan) — related I/O-throughput concern; asynchronous/parallel prefetch is in the same area.
• Only create one native plan for a query on an executor #1204 (Only create one native plan for a query on an executor) — related, since a node-local cache benefits from executor-scoped (rather than per-task) state.

Note on terminology: a previous "prefetch" toggle in Comet (see #918) referred to the legacy native_comet reader's parallel ranged-read behavior. The asynchronous prefetch proposed here is a distinct, new capability (read-ahead of file fragments ahead of demand, optionally filter-aware) and should not be confused with that flag.

I'm happy to break this into child issues per phase and to contribute the implementation. Feedback on scope, naming, and whether a design doc or a GitHub Discussion is the preferred next step would be very welcome.

[schenksj]

Update: initial implementation + benchmarks

We've built an initial implementation of two of the three pieces in this proposal:

• Node-local fragment cache — block-aligned, sitting behind the object_store API, with an
  in-memory tier (SIEVE eviction, single-flight miss dedup, coalesced fetches) and an on-disk SSD
  tier (region files, crc32c, wholesale region eviction, an admission gate that filters single-pass
  scan traffic). It relies on data-file immutability (ETag/version validation, no invalidation
  protocol). The cache core is storage-API-neutral (no object_store/opendal/engine deps) with a
  thin object_store adapter, so the same core can back the Delta/Iceberg paths later.
• Cache-affinity scheduling — driver-side sticky file→host assignment exposed through
  RDD.getPreferredLocations, so repeat reads of a file route back to the executor that already
  cached it (turning per-host private caches into a coherent cluster cache).

Both are opt-in / default-off, and correctness never depends on the cache or on locality, exactly
as framed above. Asynchronous prefetch (the third piece) is not implemented yet.

Below are cold-vs-warm numbers across scan / aggregation / join shapes, to gauge the effect.

Important setup caveat: these were run on a small MacBook Air (local[4]) reading from
S3 over the public internet on commercial broadband — so cold times include WAN round-trips and
are slower than a co-located, in-region executor would see. The absolute numbers and speedups are
therefore illustrative, not datacenter-representative; a real cluster with S3 in-region would
show smaller (but still meaningful) ratios. Each scenario forces a genuine cold read (the cache is
cleared), then times warm runs; ColdFetch/WarmFetch are bytes read from S3 per run (warm ≈ 0 =
served from cache).

Incompressible data (random doubles, ~0.5 GB on disk):

Category Scenario                  Cold(ms)  Warm(ms)  Speedup  ColdFetch  WarmFetch
------------------------------------------------------------------------------------
scan     full scan (16 cols)          30519       503    60.7x    514.8MB      0.0MB
scan     projection (1 col)            3594       147    24.4x     66.8MB      0.0MB
scan     projection (4 cols)           9647       115    83.9x    146.8MB      0.0MB
scan     filter (v2 < 0.25)            5410       119    45.5x     82.8MB      0.0MB
agg      group-by aggregation          5472       158    34.6x     98.8MB      0.0MB
agg      count distinct key            2885        77    37.5x     50.8MB      0.0MB
join     join fact-dim                 5150       149    34.6x     82.8MB      0.0MB
join     join + group-by               6319       176    35.9x     82.8MB      0.0MB
------------------------------------------------------------------------------------
mean warm speedup: 44.6x across 8 scenarios

Compressible data (id-derived; Parquet dictionary/RLE crushes it to ~8 MB):

Category Scenario                  Cold(ms)  Warm(ms)  Speedup  ColdFetch  WarmFetch
------------------------------------------------------------------------------------
scan     full scan (16 cols)           1991        81    24.6x      7.8MB      0.0MB
scan     projection (1 col)             677        38    17.8x      7.8MB      0.0MB
scan     projection (4 cols)           1264        48    26.3x      7.8MB      0.0MB
scan     filter (v2 < 0.25)             730        38    19.2x      7.8MB      0.0MB
agg      group-by aggregation           852       124     6.9x      7.8MB      0.0MB
agg      count distinct key             599        68     8.8x      7.8MB      0.0MB
join     join fact-dim                  736        66    11.2x      7.8MB      0.0MB
join     join + group-by                771       101     7.6x      7.8MB      0.0MB
------------------------------------------------------------------------------------
mean warm speedup: 15.3x across 8 scenarios

A few observations:

• The cache serves real data volume (full scan ~515 MB cold → ~0 warm) and only fetches the blocks
  a query touches (a 1-column projection reads ~67 MB vs ~515 MB for the full 16-column scan).
• Even the tiny compressed file still shows a large speedup — the cold cost over object storage is
  dominated by request/round-trip latency, so the cache helps for the compressed data real tables
  actually have, not just large files.
• Aggregation/join gains are lower than pure scans because shuffle/hash compute is a fixed cost the
  cache doesn't touch — a useful signal about where scan-side caching pays off most.

Happy to share the detailed design document (architecture, injection points, config surface,
metrics, test plan) and to break the work into per-phase child issues as suggested above.