wirybeaver commented on code in PR #2638: URL: https://github.com/apache/datafusion-ballista/pull/2638#discussion_r4433607099
########## docs/source/user-guide/iceberg.md: ########## @@ -0,0 +1,224 @@ +<!--- + Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, + software distributed under the License is distributed on an + "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + KIND, either express or implied. See the License for the + specific language governing permissions and limitations + under the License. +--> + +# Apache Iceberg + +The `ballista-iceberg` crate (`integrations/iceberg`) lets Ballista scan +[Apache Iceberg](https://iceberg.apache.org/) tables. On top of +[iceberg-datafusion](https://github.com/apache/datafusion-iceberg), it adds the +codecs that send Iceberg tables and scans between the client, the scheduler and +the executors. + +The crate is not published yet. It depends on unreleased revisions of +iceberg-datafusion and iceberg-rust, and needs Rust 1.95. + +## Usage + +Ready the session with `register_iceberg`, then mount an Iceberg catalog: + +```rust +use std::sync::Arc; + +use ballista::prelude::{SessionConfigExt, SessionContextExt}; +use datafusion::execution::SessionStateBuilder; +use datafusion::prelude::{SessionConfig, SessionContext}; +use ballista_iceberg::register_iceberg; +use iceberg_datafusion::IcebergCatalogProvider; + +let state = SessionStateBuilder::new() + .with_config(SessionConfig::new_with_ballista()) + .with_default_features() + .build(); +let ctx = SessionContext::standalone_with_state(register_iceberg(state)).await?; + +ctx.register_catalog("rest", Arc::new(IcebergCatalogProvider::try_new(catalog).await?)); +ctx.sql("SELECT * FROM rest.ns.t").await?.show().await?; +``` + +A single table registered as an `IcebergCatalogTableProvider` works the same +way. Either kind reads the table's latest version in each query. + +For time travel, register an `IcebergStaticTableProvider`, which keeps one +version: `IcebergStaticTableProvider::try_new_from_table_snapshot(table, snapshot_id)`. + +The example registers both on a table in a temporary local warehouse, and shows +a commit reaching the live table but not the static one: + +```shell +cargo run -p ballista-iceberg --example standalone-iceberg-scan +``` + +## SQL clients + +Flight SQL clients (JDBC, ADBC, BI tools) send SQL to the scheduler, which plans +it, so they hold no catalog. To let them query Iceberg tables, mount catalogs +on the scheduler with `iceberg_session_builder`, which wraps the scheduler's +session builder (`default_session_builder` here, or your own), and give the +scheduler and executors the Iceberg codecs: + +```rust +use ballista_core::utils::default_session_builder; +use ballista_iceberg::{IcebergLogicalCodec, IcebergPhysicalCodec, iceberg_session_builder}; + +let rest = IcebergCatalogProvider::try_new(catalog).await?; +let mut config = SchedulerConfig::default() + .with_flight_sql(true) + .with_override_session_builder(iceberg_session_builder( + Arc::new(default_session_builder), + [("rest".to_string(), Arc::new(rest) as _)], + )); +config.override_logical_codec = Some(Arc::new(IcebergLogicalCodec::default())); +config.override_physical_codec = Some(Arc::new(IcebergPhysicalCodec::default())); +``` + +Executors set the same two codecs on their `ExecutorProcessConfig`. A client can +then run `SELECT * FROM rest.ns.t`. Tables are pinned the same way as on the +client, while the scheduler plans the query. + +## How it works + +Ballista sends the client's plan to the scheduler, and each stage's plan to +the executors, as bytes. The codecs replace each live Iceberg table with one +version of it: the location of its metadata file and its serialized `FileIO`. +Metadata files never change, so every node rebuilds exactly that version. + +- **A catalog-backed table** is pinned to the version its catalog serves when + the client plans the query. DataFusion looks up each table a query names + once, so every scan of it, such as each side of a self-join, reads that + version, and each new query reads the latest one. +- **A static table** is sent as the version it holds. +- **A scan** is sent as the version the scheduler planned it against. + +Tables sent from the client also carry the metadata itself, so the scheduler +needn't read the file. Scans sent to executors carry only the location, which +keeps each task's plan small. + +Only the node that plans a query talks to a catalog: the client, or the +scheduler for SQL clients. Executors never need catalog configuration. + +## Combining with other table formats Review Comment: @NoahKusaba, the codec composition convention makes sense, including forwarding the UDF methods. I think we should establish the build and release boundary first, before landing individual format integrations. This was the motivation behind my [proposal in #1241](https://github.com/apache/datafusion-ballista/issues/1241#issuecomment-5648257640) and the split between #2388 and #2389: a minimal, SDK-free, opt-in build gate on the Ballista side, with the actual format integration owning its dependencies separately. The same convention should apply to Iceberg, Hudi, Lance, and Delta. Putting Iceberg in a separate crate is a useful step, but keeping it in Ballista’s root workspace and lockfile still couples dependency resolution. For example, Ballista may upgrade DataFusion before a format’s adapter supports that version. We should be able to release Ballista without waiting for every format integration to catch up. I’d suggest establishing the minimal gate first, with format SDKs and implementations outside Ballista’s main workspace, root lockfile, and release graph, then adding Iceberg against that boundary. Each integration could maintain its own compatibility requirements and release cadence. The gate would not make incompatible DataFusion versions interoperable; it would keep a format adapter’s compatibility constraints from becoming a prerequisite for Ballista’s build and release. Codec composition would then provide interoperability among integrations built against compatible versions. ########## docs/source/user-guide/iceberg.md: ########## @@ -0,0 +1,224 @@ +<!--- + Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, + software distributed under the License is distributed on an + "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + KIND, either express or implied. See the License for the + specific language governing permissions and limitations + under the License. +--> + +# Apache Iceberg + +The `ballista-iceberg` crate (`integrations/iceberg`) lets Ballista scan +[Apache Iceberg](https://iceberg.apache.org/) tables. On top of +[iceberg-datafusion](https://github.com/apache/datafusion-iceberg), it adds the +codecs that send Iceberg tables and scans between the client, the scheduler and +the executors. + +The crate is not published yet. It depends on unreleased revisions of +iceberg-datafusion and iceberg-rust, and needs Rust 1.95. + +## Usage + +Ready the session with `register_iceberg`, then mount an Iceberg catalog: + +```rust +use std::sync::Arc; + +use ballista::prelude::{SessionConfigExt, SessionContextExt}; +use datafusion::execution::SessionStateBuilder; +use datafusion::prelude::{SessionConfig, SessionContext}; +use ballista_iceberg::register_iceberg; +use iceberg_datafusion::IcebergCatalogProvider; + +let state = SessionStateBuilder::new() + .with_config(SessionConfig::new_with_ballista()) + .with_default_features() + .build(); +let ctx = SessionContext::standalone_with_state(register_iceberg(state)).await?; + +ctx.register_catalog("rest", Arc::new(IcebergCatalogProvider::try_new(catalog).await?)); +ctx.sql("SELECT * FROM rest.ns.t").await?.show().await?; +``` + +A single table registered as an `IcebergCatalogTableProvider` works the same +way. Either kind reads the table's latest version in each query. + +For time travel, register an `IcebergStaticTableProvider`, which keeps one +version: `IcebergStaticTableProvider::try_new_from_table_snapshot(table, snapshot_id)`. + +The example registers both on a table in a temporary local warehouse, and shows +a commit reaching the live table but not the static one: + +```shell +cargo run -p ballista-iceberg --example standalone-iceberg-scan +``` + +## SQL clients + +Flight SQL clients (JDBC, ADBC, BI tools) send SQL to the scheduler, which plans +it, so they hold no catalog. To let them query Iceberg tables, mount catalogs +on the scheduler with `iceberg_session_builder`, which wraps the scheduler's +session builder (`default_session_builder` here, or your own), and give the +scheduler and executors the Iceberg codecs: + +```rust +use ballista_core::utils::default_session_builder; +use ballista_iceberg::{IcebergLogicalCodec, IcebergPhysicalCodec, iceberg_session_builder}; + +let rest = IcebergCatalogProvider::try_new(catalog).await?; +let mut config = SchedulerConfig::default() + .with_flight_sql(true) + .with_override_session_builder(iceberg_session_builder( + Arc::new(default_session_builder), + [("rest".to_string(), Arc::new(rest) as _)], + )); +config.override_logical_codec = Some(Arc::new(IcebergLogicalCodec::default())); +config.override_physical_codec = Some(Arc::new(IcebergPhysicalCodec::default())); +``` + +Executors set the same two codecs on their `ExecutorProcessConfig`. A client can +then run `SELECT * FROM rest.ns.t`. Tables are pinned the same way as on the +client, while the scheduler plans the query. + +## How it works + +Ballista sends the client's plan to the scheduler, and each stage's plan to +the executors, as bytes. The codecs replace each live Iceberg table with one +version of it: the location of its metadata file and its serialized `FileIO`. +Metadata files never change, so every node rebuilds exactly that version. + +- **A catalog-backed table** is pinned to the version its catalog serves when + the client plans the query. DataFusion looks up each table a query names + once, so every scan of it, such as each side of a self-join, reads that + version, and each new query reads the latest one. +- **A static table** is sent as the version it holds. +- **A scan** is sent as the version the scheduler planned it against. + +Tables sent from the client also carry the metadata itself, so the scheduler +needn't read the file. Scans sent to executors carry only the location, which +keeps each task's plan small. + +Only the node that plans a query talks to a catalog: the client, or the +scheduler for SQL clients. Executors never need catalog configuration. + +## Combining with other table formats Review Comment: @andygrove, could you also take another look at whether it would make sense to reopen #2388 and #2389, with revisions as needed? Since the earlier discussion, hudi-rs has matured further, including the merged DataFusion 55 / Arrow 59 compatibility work in [apache/hudi-rs#698](https://github.com/apache/hudi-rs/pull/698). We also now have a [standalone Ballista–Hudi integration](https://github.com/wirybeaver/ballista-hudi) exercising the existing extension points. With Iceberg integration progressing too, I think this is a useful time to revisit the shared build and release convention. #2388 could serve as the starting point for the minimal SDK-free gate, and #2389 could demonstrate the separately maintained format implementation. Establishing that boundary first would give Iceberg, Hudi, Lance, and Delta a consistent integration model while keeping their dependency and release schedules separate from Ballista’s. I understand the earlier maintainer-bandwidth concern still applies. Would you be open to revisiting those PRs around this build and release boundary? -- 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]
