NoahKusaba opened a new pull request, #2217: URL: https://github.com/apache/datafusion-ballista/pull/2217
# Which issue does this PR close? Closes https://github.com/apache/datafusion-ballista/issues/1241 # What changes are included in this PR? (ONLY INCLUDING DISTRIBUTED READS/WRITES integration for now) (Note that READS are still not properly distributed, as output-partitioning is not supported in iceberg-datafusion yet https://github.com/apache/iceberg-rust/blob/0e7cc4f19ef3b8d29a5f81e3f666a76a6130a9d5/crates/integrations/datafusion/src/physical_plan/scan.rs#L106) Adds a new iceberg-ballista crate, which provides a distributed-query driver for Apache Iceberg for a distributed datafusion engine Apache Datafusion-Ballista + the targeted changes to iceberg-datafusion that make Iceberg's existing plan nodes serializable so they can cross node boundaries. The core problem it solves Iceberg's DataFusion integration already produces complete physical read and write plans, but every Iceberg plan node holds live, non-serializable state (Arc, an open Table/FileIO). Ballista ships logical and physical plans to remote schedulers/executors, so those nodes couldn't travel. This branch closes that gap with one consistent idea: serialize a minimal self-contained recipe (IcebergCatalogConfig + identifiers), rebuild the live objects on the receiving node. IcebergLogicalCodec: serializes the catalog-backed table provider (config + table ident, plus snapshot/metadata variants) so the scheduler can rebuild it and do physical planning, including INSERT. IcebergPhysicalCodec: serializes the four Iceberg execution nodes (IcebergTableScan, IcebergWriteExec, IcebergCommitExec, IcebergMetadataScan) and the PartitionExpr physical expression. Tagged-envelope wire framing (TAG_ICEBERG / TAG_DELEGATED):every blob carries a leading tag; non-Iceberg nodes are delegated to Ballista's own codec, so shuffles/sorts/etc. keep working and an unknown tag is a hard error--> Based off comments from https://github.com/milenkovicm/ballista_delta bridge.rs runtime bridge: Each executor node needs to build an HTTP client with the iceberg catalog which requires an async-call, but PhysicalExtensionCodec from datafusion_proto try_decode is synchronous. The block_on function is a workaround to make this async function call blocking. Each try_decode also performs a load_table catalog round-trip per plan node to resolve the table's current metadata pointer -Public API: register_iceberg_codecs(SessionConfig) and register_iceberg_table Snapshot pinning at encode time # Are there any user-facing changes? <!-- If there are user-facing changes then we may require documentation to be updated before approving the PR. --> - Distributed reads / writes tested against dockerized minio iceberg catalog ( Standalone + multi-executor cluster). Also tests partitioned files writes + iceberg table registration. - Codec serialization / deserialization changes. - Snapshot Read tests (will document better later) # Other notes There is a dependency on the changes to the iceberg-datafusion crate, which are not yet merged in. Currently referencing my branch found at https://github.com/apache/iceberg-rust/pull/2613 -- 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]
