NoahKusaba commented on PR #2862:
URL: https://github.com/apache/iceberg-rust/pull/2862#issuecomment-5112493757
> 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`:
```rust
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:
```rust
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:
```rust
// 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.
--
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]
