NoahKusaba opened a new pull request, #2638:
URL: https://github.com/apache/datafusion-ballista/pull/2638
# Which issue does this PR close?
Part of #2174: the first step of its integration foundation, standalone
Iceberg scans and codec support. Split out of #2217 as its smallest reviewable
piece: reads only, built on merged upstream code. Related: #1241.
# Rationale for this change
datafusion-iceberg plans Iceberg scans, but its tables and scans hold live
state (a loaded `Table` and its `FileIO`), so Ballista can't send them to the
scheduler and executors. This PR adds codecs that can.
# What changes are included in this PR?
A new crate, `iceberg-ballista` (`integrations/iceberg`), with a logical and
a physical extension codec, registered with `register_iceberg_codecs(config)`.
No changes to Ballista core.
- **Each table and scan travels as one version of the table**: the location
of its metadata file plus its serialized `FileIO`. Metadata files are
immutable, so every node rebuilds exactly that version without a catalog.
Tables sent from the client also carry the metadata the client already loaded,
so the scheduler doesn't read the file. Scans sent to executors carry only the
location, keeping each task's plan small.
- **Catalog-backed tables** (`IcebergCatalogTableProvider`, including the
ones a mounted `IcebergCatalogProvider` serves) are loaded on the client when
the query is sent. Each query reads the latest version, and the scheduler and
executors need no catalog configuration.
- **Static tables** (`IcebergStaticTableProvider`) send the version they
hold, for time travel.
- **Framing**: each payload starts with a tag byte, Iceberg (`1`) or
delegated (`0`), which also versions the payload format. Everything else goes
to the codec the session already had, so other Ballista extensions keep
working. A driver for another table format, such as Delta, maintained outside
this repo, can wrap these codecs the same way, so one query can join tables
from both. It only has to be stacked in the same order on every node.
- **Rebuilt tables** run on a runtime that lives as long as the process, so
they outlive whatever runtime calls the codecs. Each process caches up to 16
table versions. The cap bounds memory, since a table with a long history can
hold megabytes of metadata. It isn't configurable yet: a session setting can't
govern a per-process cache, and a constructor option can be added later without
breaking anything.
Known limitations, also listed in the docs:
- Reads only. `INSERT INTO` fails with "isn't supported on Ballista yet".
- One task per scan, until Iceberg scans split into partitions.
- Catalog-backed tables need a multi-threaded tokio runtime on the client.
Sending the query blocks while it loads the table, which can hang a
single-threaded runtime, so the codec returns an error there instead. The
proper fix is an async plan-preparation hook in Ballista.
- A table read twice in one query, such as both sides of a self-join, can
see two versions.
- No metadata tables (`t$snapshots`).
- Storage credentials travel in plain text, so the links between nodes must
be trusted.
Dependencies: iceberg-rust and datafusion-iceberg are pinned to git
revisions until they release, with iceberg at the revision datafusion-iceberg
pins so the build resolves a single iceberg crate. The crate is `publish =
false` and needs Rust 1.95; the rest of the workspace still builds on 1.94.
Almost all of the `Cargo.lock` diff (about 90 new packages) is iceberg's own
dependency tree, plus opendal for S3 access.
# Are these changes tested?
Yes:
- **Codec tests**:
- a scan keeps its snapshot, projection, filter and fetch through encode
and decode;
- a catalog-backed table is pinned to the version loaded at encode time,
decodes without its metadata file (the test deletes it), and refuses `INSERT`;
- a single-threaded client runtime gets an error rather than a hang;
- non-Iceberg nodes are handed to the inner codec;
- malformed buffers (empty, an unknown tag, an invalid payload) are
rejected.
- **Standalone Ballista** over a local memory catalog: a mounted catalog
reads commits made after it was registered, and a static table keeps its
version.
- **The same checks against S3**, through an Iceberg REST catalog backed by
RustFS. These run with testcontainers behind the `testcontainers` feature,
which the existing `linux-test` job already enables.
# Are there any user-facing changes?
Yes, a new crate and a docs page:
- `docs/source/user-guide/iceberg.md`, covering usage, time travel, how it
works and the limitations;
- a runnable example: `cargo run -p iceberg-ballista --example
standalone-iceberg-scan`.
No existing APIs change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
