nordisk/skade

Full-scan decode — the read_all fast path

Class: design · Function: read · Uses: read.rs, warehouse, znippy-zoomies/gatling_forkjoin, parquet. Only the shape that runs. This is how Table::read (read_all → read_all_gatling) turns a plain append-only snapshot into Arrow batches without paying the Iceberg engine's per-file async to_arrow pipeline — and, as of the local-fd change, without slurping the whole dataset into memory.

The three-way decode

read_all picks the cheapest correct decode for the current snapshot:

1. Fast gatling path (read_all_gatling, the common case) — a plain append-only, single-schema table with no delete files. The data-file set is resolved from the manifest list, then decoded straight to Arrow across all cores via the ONE fork-join engine (gatling), one worker per file, batches kept in file order. Bypasses the engine's per-file async to_arrow.
2. Engine fallback — schema evolution (field-id remap) or any delete file (row-level merge-on-read). read_all_gatling returns None and read_all falls back to scan().select_all().to_arrow(). Correctness first: the fast path never drops a deleted row or mis-maps an evolved schema.
3. Empty — no current snapshot ⇒ zero batches.

Inside the fast path: local-fd streaming vs whole-Bytes slurp

Once the data-file set is known, the decode itself has two byte/row-identical strategies — same Parquet reader, same default batch size, same per-file gatling fan-out, same file order — chosen by whether the FileIO backend is local:

read_all_gatling
  ├─ resolve data files (walk manifest list + parse manifests)
  ├─ all_local_paths(files)?              # every file a file:// / absolute path that exists
  │     ├─ Some(paths) → decode_data_local(paths)     # LOCAL  — stream from fd
  │     └─ None        → decode_data_bytes(file_io, …) # REMOTE — whole-Bytes via FileIO
  └─ batches (file order)

• decode_data_local (local warehouse — the common Warehouse::open case, and the oxr/mashup read). Each file is opened as a std::fs::File and decoded straight from the fd — the Parquet reader pulls pages lazily. Nothing is pre-slurped. Peak memory is the decoded batches + per-file page buffers, and each file's I/O overlaps its own decode.
• decode_data_bytes (any non-local FileIO — S3, memory). The backend-neutral fallback: read each file's whole bytes through the Iceberg FileIO, then decode from the in-memory Bytes. This is what every backend can serve.

The gate is all-or-nothing (all_local_paths returns None if any file isn't a local file that exists), so a single scan never mixes the two strategies.

Why the local path exists (the waste it removes)

The pre-change fast path did decode_data_bytes for every backend: it read all data files' whole bytes into a Vec<Bytes> before decoding a single row — so peak RSS was the entire dataset, and the I/O was a serial pre-read in front of the decode. Against raw Parquet (which streams pages lazily from the fd) this both lost throughput and spiked memory on a full scan of a big local table. For a local warehouse — where the files are right there on disk — reading whole bytes into the heap first is pure waste. decode_data_local removes it while staying byte/row-identical to the fallback, so the S3 path is untouched.

The data-file resolution (manifest-list walk + Avro manifest parse) is inherent — a full scan must learn which files the snapshot points at — and is unchanged.

Correctness / equivalence

The two decoders emit the identical batch sequence (schema + values + order), guarded RED-when-broken by local_streamed_decode_is_byte_identical_to_bytes (streamed decode == the Bytes reference over a multi-file, multi-row-group fixture; reversing the file order fails it) and read_all_gatling_uses_local_streamed_path (a local warehouse takes the streamed path — LOCAL_STREAMED_READS counter — with no Bytes-slurp fallback, all rows intact). The read_all ↔ engine-fallback equivalence is covered by the existing roundtrip / data-skip suites.

Reuse (L5)

The file:///absolute → local-path rule is warehouse::local_path_of (pub(crate)), consumed by both the compaction data-dir logic and this read fast path — never re-derived. The all-core decode is the ONE fork-join engine (gatling), the same one read_delta / read_row_groups use — no private thread pool.