feat(datafusion) Add snapshot_id parameter to pin table reads

[NoahKusaba]

Which issue does this PR close?

• Closes: Add snapshot_id parameter for Datafusion IcebergTableProvider (Distributed reads) #2861

Related to : #2613

What changes are included in this PR?

Optional Snapshot ID parameter to IcebergTableProvider to pin table reads

Are these changes tested?

• test_pinned_snapshot_reads_historical_data: Tests that pinning and unpinning snapshot Id's for a table provider with multiple snapshots, correctly reads from the correct snapshot by asserting against expected output.

[NoahKusaba]

@CTTY do you have time to look at this?
Tried to make this change as simple as possible.
7 lines of functional code change, everything else is comments or test.

[xanderbailey]

Hey, thanks for the PR, just wondering what the benefit of this is vs using the StaticTableProvider?

If we decide this is the right path forward, I wonder if we can make it generic such that this works for incremental append scans etc also? Maybe with a ScanRange enum or something? Happy to hear thoughts here!

[NoahKusaba]

what the benefit of this is vs using the StaticTableProvider?

If we decide this is the right path forward, I wonder if we can make it generic such that this works for incremental append scans etc also? Maybe with a ScanRange enum or something? Happy to hear thoughts here!

My idea is to register tables to datafusion using IcebergTableProvider, so that the same registration also allows for inserts/deletes/merges, etc. Whereas StaticTableProvider is strictly scan only.
Also a staticTableProvider can never go back to current_state as it never refreshes, with my PR unpinning
with_snapshot_id(None)
will return back to current state.

There's also a distributed angle (my motivating use case is a Ballista integration): the flow is Client -> LogicalCodec serializes the provider's recipe -> Scheduler rebuilds the provider and does physical planning -> PhysicalCodec -> Executors. The provider needs to hold everything required to recreate the table connection on remote executors: snapshot_id here, and catalog config (catalog_type, name, storage_properties) in a follow-up PR.
This is split out from #2727, where the fuller picture is visible.

I think the ScanRange suggestion is great, and since the current snapshot_id field is private we can replace it with ScanRange in the future without breaking anything. This would also require some plumbing to IcebergTableScan to work, which I think should live in a separate PR to keep this minimal.
We can make a stub field too for the time being, and put snapshot_id in it?

[xanderbailey]

Something seems a little off here to me. Read/write skew seems like a footgun. A single registered table that reads snapshot A but writes to HEAD is a confusing object. INSERT INTO t; SELECT * FROM t won't show the row you just inserted. That's a strange contract to hand a user in my opinion.

I also think schema isn't being handled here correctly for pinned snapshots, we should resolve against the snapshot's schema rather than current, the static table provider handles this correctly.

[NoahKusaba]

Something seems a little off here to me. Read/write skew seems like a footgun. A single registered table that reads snapshot A but writes to HEAD is a confusing object. INSERT INTO t; SELECT * FROM t won't show the row you just inserted. That's a strange contract to hand a user in my opinion.

I also think schema isn't being handled here correctly for pinned snapshots, we should resolve against the snapshot's schema rather than current, the static table provider handles this correctly.

insert-read problem:
Sorry for that, I understand what you are saying on careful review. I was so tunnel visioned on the ballista integration, that I had missed actual users directly calling snapshot_id.

The way I have it wired in my Ballista integration is that snapshot_id is resolved by the scheduler when it encodes the physical plan to executors, so a read -> insert -> read would properly show the new data as the snapshot_id would be reset to head by the scheduler as queries are sequentially made.

I think it makes sense to add a write guard, if the snapshot_id is pinned (not None), to constrain confusing behavior? I added it in a new commit, let me know how you think this should be handled.

[xanderbailey]

Thanks for iterating with me! What are we gaining here over using the StaticTableProvider now?

[NoahKusaba]

Thanks for iterating with me! What are we gaining here over using the StaticTableProvider now?

Thanks for helping me with this too (still fixing the schema update with snapshot id problem), and sorry for responding late.

To answer your question, anyone not using the iceberg-ballista integration likely won't benefit from these changes, and they should keep using IcebergStaticTableProvider for time-travel reads. After this PR the two resolve schemas through the same helper and both refuse writes when pinned, so read semantics are identical.

Ballista requires you instantiate one TableProvider, through which queries are routed (explained above). This creates two requirements:

1. The TableProvider needs to carry all the information required to rebuild it on the scheduler, which de-serializes the logical plan and re-plans it (this PR + another one add the missing requisite information: snapshot-id + catalog-configuration). StaticTableProvider is built from an already-loaded Table and carries no catalog identity, so there is nothing a codec can rebuild it from.
2. If I want the same instantiated TableProvider to allow for both Scanning + Insert + etc, it needs functional .scan() and .insert_into() implementations. StaticTableProvider has the methods (the trait requires them), but its .insert_into() unconditionally errors, so writes are structurally impossible for it.

[NoahKusaba]

Schema Problem:
Following StaticTableProvider's approach (per your suggestion), pinning now re-derives the exposed schema from the pinned snapshot.

**Also now re-loads the table to reload the schema.

While addressing this, I found a second issue: with_snapshot_id cached the Arrow schema at construction time (the current schema). Pinning across a schema evolution would then push down non-existent column names from the old schema.

Fix: Schema resolution is now centralized in a shared snapshot_arrow_schema helper that looks up the snapshot’s own schema via snapshot.schema(metadata). Both with_snapshot_id and IcebergStaticTableProvider constructors use this helper, ensuring a single resolution path. with_snapshot_id is now fallible, as the snapshot ID is validated during pinning.

[NoahKusaba]

Hey @xanderbailey can you take a look at this again when you have time?
Hopefully it is in a more acceptable state.

[xanderbailey]

Run with me here for a moment if you will. Does something like this not work for ballista?

  pub struct IcebergSerializableTableProvider {
      table_ident: TableIdent,
      metadata: TableMetadata,
      storage_props: HashMap<String, String>,
      storage_factory: Arc<dyn StorageFactory>,
      snapshot_id: Option<i64>,
      inner: IcebergStaticTableProvider,
  }

If we were to add something like this as the table provider, everything here apart from inner is serialisable which means you can fully construct it on the "executor side" (sorry not sure what ballista calls these). This keeps the current public API clear of any runtime errors that may happen as a result of people trying to pin for scans and fail on read. My understanding is that the IcebergStaticTableProvider is designed for exactly the use-case you're describing.

[NoahKusaba]

This keeps the current public API clear of any runtime errors that may happen as a result of people trying to pin for scans and fail on read

( Did you mean pin for scans and fail on inserts?)

Thanks for the response. Let me give a clearer description of what Ballista is doing
here, since the execution model isn't obvious.

To register a table with a Ballista context, the client provides a TableProvider:

pub async fn register_iceberg_table(
    ctx: &SessionContext,
    register_name: &str,
    config: IcebergCatalogConfig,
    namespace: NamespaceIdent,
    table: impl Into<String>,
) -> Result<(), DataFusionError> {
    let catalog = bridge::build_catalog(&config).await?;
    let provider = iceberg_datafusion::IcebergTableProvider::try_new_with_config(
        catalog, config, namespace, table,
    )
    .await
    .map_err(bridge::to_df_err)?;
    ctx.register_table(register_name, Arc::new(provider))?;
    Ok(())
}

From there:

Client     -> LogicalCodec::try_encode     (extract serializable fields from the provider)
Scheduler  -> LogicalCodec::try_decode     (rebuild the provider)
Scheduler  -> create_physical_plan()
                -> provider.scan() / .insert_into()
                -> IcebergTableScan  /  (IcebergWriteExec -> IcebergCommitExec)
Scheduler  -> physical optimizer rules
Scheduler  -> split into stages at shuffle boundaries
Scheduler  -> per stage: PhysicalCodec::try_encode -> ship to executors
Executor   -> decode, run one partition

The provider never reaches an executor. It's rebuilt on the scheduler, used for a single
create_physical_plan() call, and stays there. Executors only ever receive encoded
ExecutionPlan nodes.

At try_decode the codec can't tell a read from a write. datafusion-proto encodes DML
write targets as synthetic TableScan nodes, so scans and inserts arrive in an identical
shape. That rules out having the codec build an IcebergStaticTableProvider for reads and
an IcebergTableProvider for writes, since the information isn't there.

On the struct as sketched: storage_factory is Arc<dyn StorageFactory>, a trait object,
so it can't be serialized. And StorageConfig is a props map with no scheme field, so
storage_props alone doesn't say which backend to rebuild. So "everything apart from
inner is serialisable" unfortunately doesn't hold as written.

That said, I've been investigating your suggestion in this shape:

pub struct IcebergSerializableTableProvider {
    snapshot_id: Option<i64>,
    catalog_config: IcebergCatalogConfig,   // from the other PR
    schema: SchemaRef,
    inner: IcebergTableProvider,
}

Most methods delegate to inner. scan() and schema() are overridden to apply the
pinned snapshot. This keeps the distributed concerns in the Ballista crate, which I think
is what you're after.

It will still require some of the upstream change listed in NoahKusaba:feature/datafusion-catalog-config-v2 such as:

// table/mod.rs, to construct `inner`
pub(crate) async fn try_new(..)                -> pub

// physical_plan/scan.rs, to rebuild a scan on the executor
pub(crate) fn new(..)                          -> pub
pub fn with_predicates(..)                     // new. `new` derives the predicate from
                                               // Expr filters, but on decode we already
                                               // hold a Predicate and can't turn it back

// physical_plan/mod.rs, these are pub(crate) structs and so currently unnameable
pub use commit::IcebergCommitExec;
pub use write::IcebergWriteExec;
pub use metadata_scan::IcebergMetadataScan;
pub use project::PartitionExpr;

I still need to more deeply investigate the full lift here.

So the tradeoff is that the wrapper keeps new fields out of iceberg-datafusion, at the cost
of an extra abstraction layer + making execution-node constructors API public (which I needed to do anyway).

I'd also note the pieces in these PRs seem useful beyond Ballista.
IcebergStaticTableProvider supports time travel but has no catalog backing, so there's
currently no way to pin a snapshot on a catalog-registered table. My original instinct was
to put this in iceberg-datafusion so other engines could reuse it and to avoid the extra
abstraction layer, but I understand if you'd rather keep distributed concerns out of the
crate.

On the runtime-error concern specifically: an alternative is to have insert_into reset
snapshot_id to None and reload the schema, rather than erroring. That trades the error
for behavior some users might find surprising, so I don't have a strong preference. Happy
to go whichever way you prefer.

If we don't want to merge the snapshot change, I will persue the thread of keeping more distributed required features in the ballista-iceberg crate instead.

[xanderbailey]

I'd also note the pieces in these PRs seem useful beyond Ballista.
IcebergStaticTableProvider supports time travel but has no catalog backing, so there's
currently no way to pin a snapshot on a catalog-registered table.

I think this part could be achieved with a UDTF or better still an async UDTF when it’s supported in the future.

[corleyma]

Being able to timetravel from the catalog is just basic functionality that most other engines / iceberg clients have.

Not a rustacean but I can certify this would be useful even from a non-distributed execution context. We use this functionality heavily in pyiceberg.

[mbutrovich]

I had a five hour flight today, so I spent some time digging into this. Catalog-backed time travel is worth having on its own merits: pyiceberg and other Iceberg clients already treat it as basic functionality. The schema-per-snapshot fix and the write guard added across this thread both address real problems @xanderbailey raised, and the tests cover the resulting behavior well. The remaining open question is narrower: does the pinning capability belong on IcebergTableProvider specifically, the type whose other job is writing to HEAD?

Given #2613's discussion and the move of the Ballista-specific crate to apache/datafusion-ballista#2217, the useful framing here is finding the smallest, generally useful addition to this crate that unblocks that PR, independent of Ballista's own internal design choices. Three data points support that framing:

In apache/datafusion#7292 ("Table time travel support"), closed in 2023, the DataFusion community discussed this exact question directly: should time travel be a runtime argument threaded through TableProvider::scan, or a distinct provider instance per version, resolved ahead of planning? The discussion landed on the latter, explicitly because a version argument on scan() would also require versioning schema(), statistics(), and everything else that depends on them, multiplying the number of new parameters needed throughout the API. The resolution was to leave version resolution to each catalog integration, registering a separate TableProvider per version before a TableScan node is ever built, using the AS OF grammar sqlparser added for this purpose. That is the same schema-per-version problem this PR's own thread ran into (the snapshot_arrow_schema re-derivation fix), and it is already the shape IcebergStaticTableProvider has. The DataFusion community already moved away from the runtime-argument version of this design, for exactly the failure mode this thread hit. Adding snapshot_id: Option<i64> plus a runtime write guard to IcebergTableProvider goes against that precedent rather than following it.

In iceberg-java, SparkTable does guard writes to a pinned snapshot at runtime: Preconditions.checkArgument(snapshotId == null, "Cannot write to table at a specific snapshot: %s", snapshotId) in newWriteBuilder (spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java:285-287), so a runtime write guard on a pinned instance is not unprecedented by itself. But SparkTable.snapshotId is private final, set only via the constructor or copyWithSnapshotId (SparkTable.java:120, 180-182), and SparkCatalog.loadTable(Identifier, version) (SparkCatalog.java:176-207) asserts the table it starts from is unpinned before handing back a new copy. Pinning always produces a distinct instance; it never happens on a table a caller is already holding or has registered, matching the DataFusion resolution above.

Among DataFusion's other TableProviders, ViewTable, CteWorkTable, EmptyTable, StreamingTable, and GenerateSeriesTable get read-only-ness by simply never overriding insert_into, inheriting the trait's not_impl_err! default (datafusion/session/src/table.rs:341-348). None carries a field that flips an otherwise-writable impl into a rejecting one.

IcebergTableProvider::with_snapshot_id (crates/integrations/datafusion/src/table/mod.rs:139-144) doesn't mutate in place; it consumes self and returns a new owned value, so it's closer to copyWithSnapshotId than a true flag flip. But it's still the one type in this crate carrying the ~90-line insert path (project_with_partition, repartition, sort_by_partition, IcebergWriteExec, IcebergCommitExec), and once pinned that whole path exists only to error. IcebergStaticTableProvider already exists as the read-only, pinned-snapshot type, and per this thread, once the write guard landed, its read semantics became identical to a pinned IcebergTableProvider: both resolve schema through snapshot_arrow_schema and both refuse writes.

On the Ballista motivation specifically, I looked at whether a distributed engine genuinely forces this shape, and it doesn't, independent of any one integration's own design choices. The one confirmed, integration-independent constraint is that Ballista's LogicalExtensionCodec supports no custom TableProvider by default (it delegates straight to DataFusion's DefaultLogicalExtensionCodec, which unconditionally returns not_impl_err!), so any distributed integration needs its own codec regardless of how IcebergTableProvider is shaped.

Concretely, here's what would unblock apache/datafusion-ballista#2217 without the design tradeoff above: give IcebergStaticTableProvider a way to carry catalog identity (config + table ident) alongside the Table it already holds, so a codec can reconstruct it the same way it reconstructs IcebergTableProvider today. That's the actual gap @NoahKusaba identified upthread ("StaticTableProvider carries no catalog identity"). It's additive rather than a change to IcebergTableProvider's existing contract, and it keeps the read/write split exactly where DataFusion's own architecture (per apache/datafusion#7292) already says it should be. IcebergTableProvider would stay exactly as it is today: no snapshot_id, no write guard, no schema recomputed at pin time.

This is separate and doesn't block this PR. While digging into why a codec has to decide read-vs-write at decode time at all, I found that datafusion-proto encodes a DML write target as a synthetic TableScan (from_table_source/to_table_source, datafusion/proto/src/logical_plan/mod.rs:357-383) before it ever reaches LogicalExtensionCodec::try_decode_table_provider, so the codec gets no signal distinguishing a write target from a real scan. That's a generic DataFusion gap, not specific to Iceberg or Ballista, and worth raising upstream on its own: any custom TableProvider wanting to build something read-only at decode time hits the same wall. Fixing it there would remove the need for this kind of runtime write guard for every future integration, not just this one.

If the IcebergStaticTableProvider direction works for you, I can sketch what the catalog-identity change would look like.

[mbutrovich]

This reloads table metadata from the catalog purely to validate the snapshot and recompute the schema, and scan() (line 178-183 below) reloads again right after. That's two catalog round trips before a query that needs one, and it makes what reads like a builder setter into a fallible, async network call. Per the comment above it, this cost exists specifically to handle providers the catalog hands out and caches across calls. Is that caching concern this type's to own, or should the reload-for-freshness logic live wherever the caching actually happens (the Ballista scheduler side, by the description upthread)?

[NoahKusaba]

Thank you so much for taking the time to look at this both @mbutrovich and @xanderbailey. I understand much better now the conventions of the datafusion eco-system, and how I can address the problem properly.
I'll close the PR and look to pursue the suggestions made of:

• StaticTableProvider + catalog/table identity.
• datafusion-proto codec getting a distinguishing read/write target.

Don't have alot of time this weekend, so I will start putting up the PR's in a few days. Would appreciate any further advice you have to offer regarding shaping the StaticTableProvider PR!