Repository navigation
feat(iceberg): re-pin a view's Iceberg tables each time a query names it - #2784
Draft
NoahKusaba wants to merge 42 commits into
Draft
NoahKusaba wants to merge 42 commits into
NoahKusaba wants to merge 42 commits into
Conversation
register_iceberg(state) wraps the session's catalogs so that DataFusion's async table lookup, which runs once per table a query names, loads each catalog-backed Iceberg table and returns it at that version. Every scan of a table in one query reads the same version, and the logical codec no longer loads tables, so it no longer blocks the client or needs a multi-threaded runtime.
Flight SQL clients send SQL that the scheduler plans, so they hold no catalog. iceberg_session_builder builds the scheduler's sessions with configured catalogs mounted and readied with register_iceberg, so their tables are pinned while the scheduler plans. register_iceberg now wraps the codecs itself and leaves an already-ready session unchanged, so register_iceberg_codecs is gone. The docs now say a separately deployed scheduler and executors set the Iceberg codecs as their codec overrides, and that a view keeps the version from when it was created.
A metadata table such as t$snapshots reached the inner codec, which failed with an internal error suggesting a DataFusion bug. The logical codec now refuses it as not supported on Ballista yet.
iceberg_session_builder now takes the scheduler's existing session builder and wraps it, so a scheduler that sets up its sessions itself (for example to register object stores) keeps that setup. Also explains the dead_code allow on the shared test helpers.
iceberg_session_builder now takes and returns Arc<dyn Fn>, the shape SchedulerConfig stores, so an existing builder passes straight in and the result goes straight to with_override_session_builder. The Flight SQL docs now link to mounting Iceberg catalogs.
SessionStateBuilder::new_from_existing gives a session a new ID, so register_iceberg now keeps the original, as Ballista's own share_file_statistics_cache does. Also makes the SessionBuilder alias private: ballista-scheduler already exports one.
Payloads that aren't Iceberg's now pass to the inner codec untouched, so codecs for other table formats that follow the same convention can be stacked in any order. Document the convention in the user guide.
…lone-scans # Conflicts: # Cargo.lock
ballista-iceberg is now its own workspace, excluded from Ballista's like python/, with its own Cargo.lock and CI workflow. Ballista's root Cargo.lock matches main again, so a DataFusion upgrade that iceberg-rust doesn't support yet doesn't block Ballista's build or release. SchedulerConfigExt::with_iceberg (the `scheduler` feature) sets the Iceberg codecs and mounts catalogs on the scheduler's sessions, and ExecutorProcessConfigExt::with_iceberg (the `executor` feature) sets the codecs on an executor. Both wrap whatever the config already has. They replace iceberg_session_builder.
…nto feat/iceberg-standalone-scans
…nto feat/iceberg-standalone-scans
Reverts 9025298 so it can be reviewed on its own. Views keep the version from when they were created, documented as a limitation.
Depend on the Ballista crates at a commit that includes apache#2752 instead of the crates beside it, and stop running the workflow on changes under ballista/, so a dependency bump in Ballista can't break this build.
Ballista restricts each task to its partitions only for the scans it knows, so each task would read every partition of an Iceberg scan and return duplicate rows. Fail loudly once a dependency bump starts splitting scans.
github-actions
Bot
added
documentation
…nto feat/iceberg-standalone-scans
A table's metadata grows with each snapshot, and every scan of it carried a copy, so a self-join on a table with a long history could exceed the gRPC message limit. Send only the metadata file's location: the table cache keeps the extra read to once per version.
…views # Conflicts: # integrations/iceberg/src/lib.rs
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.
Which issue does this PR close?
Closes #2726.
Stacked on #2638, so draft until it merges. Only the last commit, "Re-pin a view's Iceberg tables each time a query names it", is new; the rest is #2638. Split out at andygrove's request so the plan rewriting gets its own review.
Rationale for this change
#2638 pins each catalog-backed Iceberg table to one version when a query looks it up. DataFusion stores a view as the plan it was created with, so a view over an Iceberg table keeps reading the version from when the view was created. In single-process DataFusion, the same view reads new commits on every run.
What changes are included in this PR?
When a query names a view, the pinning schema provider re-pins the view's stored plan: every scan of a pinned catalog-backed table, including scans inside subqueries, is replaced with the version its catalog serves now. Each distinct table is loaded once, and the loads run concurrently.
PinnedTablekeeps theIcebergCatalogTableProviderit was pinned from (never sent between nodes), so it can be pinned again.It re-pins the stored plan rather than re-planning the view's SQL.
SchemaProvider::tablegets no session state to plan with, and a view created from aDataFramehas no SQL. Nested views need no recursion, since DataFusion inlines a view into any plan built over it.Limitations:
DataFrameor Flight SQL prepared statement still reads the version from when it was planned. Neither looks its tables up again.Are these changes tested?
Yes.
should_read_a_views_latest_version(standalone Ballista over a memory catalog) creates a view over a table and a second view over the first with a subquery on the table, commits a row, and checks that both views return it. The test fails without the re-pinning and without the subquery handling.Are there any user-facing changes?
Yes: views over Iceberg tables now read the latest version. The docs' limitation for views is narrowed to reused
DataFrames and prepared statements.