milenkovicm commented on code in PR #2638: URL: https://github.com/apache/datafusion-ballista/pull/2638#discussion_r4436435564
########## 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() Review Comment: Can we provide `SchedulerConfigExt::with_iceberg` which would hide all of this ceremony? ########## 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 + +The Iceberg codecs wrap the codecs a session already has, so one query can +read Iceberg tables alongside tables of another format, such as Delta or Hudi, +whose codecs are maintained elsewhere. Any codec that follows the same +convention can be stacked with these in any order: + +1. **Wrap an inner codec.** Take the session's existing codec in the + constructor rather than Ballista's default, and pass it everything that + isn't yours. +2. **Prefix your own bytes with a unique name.** Iceberg payloads start with + `iceberg`. Choose a name for your format, such as `delta`, that no other + codec uses and that doesn't start another codec's name. +3. **Pass everything else through untouched.** When encoding something that + isn't yours, write only what the inner codec writes, without adding a + marker. When decoding, if the bytes don't start with your prefix, hand all + of them to the inner codec. + +A logical codec for a hypothetical Delta integration would look like this: + +```rust +/// Rule 2: the name that starts every payload this codec writes. +const PREFIX: &[u8] = b"delta"; + +#[derive(Debug)] +pub struct DeltaLogicalCodec { + inner: Arc<dyn LogicalExtensionCodec>, +} + +impl DeltaLogicalCodec { + /// Rule 1: wrap whatever codec the session already has. + pub fn new(inner: Arc<dyn LogicalExtensionCodec>) -> Self { + Self { inner } + } +} + +impl LogicalExtensionCodec for DeltaLogicalCodec { + fn try_encode_table_provider( + &self, + table_ref: &TableReference, + node: Arc<dyn TableProvider>, + buf: &mut Vec<u8>, + ) -> Result<()> { + let Some(table) = node.downcast_ref::<DeltaTableProvider>() else { + // Rule 3: not ours, so write only what the inner codec writes. + return self.inner.try_encode_table_provider(table_ref, node, buf); + }; + buf.extend_from_slice(PREFIX); + encode_delta_table(table, buf) + } + + fn try_decode_table_provider( + &self, + buf: &[u8], + table_ref: &TableReference, + schema: SchemaRef, + ctx: &TaskContext, + ) -> Result<Arc<dyn TableProvider>> { + match buf.strip_prefix(PREFIX) { + Some(payload) => decode_delta_table(payload, schema), + // Rule 3: not ours, so hand over every byte unchanged. + None => self.inner.try_decode_table_provider(buf, table_ref, schema, ctx), + } + } + + // Every other method forwards to `self.inner`, for example: + fn try_decode_udf(&self, name: &str, buf: &[u8]) -> Result<Arc<ScalarUDF>> { + self.inner.try_decode_udf(name, buf) + } + // ... +} +``` + +A physical codec follows the same pattern in `try_encode` and `try_decode` +for its scan. Because each codec only claims bytes with its own prefix, the +codecs can be stacked in either order: + +```rust +let logical = IcebergLogicalCodec::new(Arc::new(DeltaLogicalCodec::new( Review Comment: What's the benefits of this approach over https://docs.rs/datafusion-proto/latest/datafusion_proto/physical_plan/struct.ComposedPhysicalExtensionCodec.html ? ########## 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: Would standard composable codec work with multiple codecs? ########## 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: I still see no reason to bring Hudi as dependency -- 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]
