You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "marvinlanhenke (via GitHub)" <gi...@apache.org> on 2024/04/06 04:28:44 UTC
[PR] [WIP] Integration with Datafusion [iceberg-rust]
marvinlanhenke opened a new pull request, #324:
URL: https://github.com/apache/iceberg-rust/pull/324
I had some fun, working with Datafusion and I want to share the (rough) design with you.
As outlined in #242 I went ahead and:
- created a new crate `integrations`
- a subfolder `datafusion` with its respective modules (catalog.rs, schema.rs, etc.)
- implemented some of the traits e.g. `CatalogProvider`
- basic test infra for integration test with HiveMetastore
---
The overall structure should be extensible, meaning - if we want to support an integration with e.g. `polars` we simply create a new subfolder and implement the traits.
The test infra is supposed to be used by every integration.
The IcebergCatalogProvider, for example, should be usable by any Catalog (since it Arc<dyn Catalog> for the client (dynamic dispatch))
I really would appreciate your thoughts on this approach - so we can discuss how to proceed, whats missing upstream (like #277), etc.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [WIP] Integration with Datafusion [iceberg-rust]
Posted by "tshauck (via GitHub)" <gi...@apache.org>.
tshauck commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1562769346
##########
crates/integrations/src/datafusion/schema.rs:
##########
@@ -0,0 +1,97 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use dashmap::DashMap;
+use datafusion::{
+ catalog::schema::SchemaProvider, datasource::TableProvider, error::DataFusionError,
+};
+use futures::FutureExt;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::datafusion::table::IcebergTableProvider;
+
+pub(crate) struct IcebergSchemaProvider {
+ tables: DashMap<String, Arc<dyn TableProvider>>,
+}
+
+impl IcebergSchemaProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: &NamespaceIdent,
+ ) -> Result<Self> {
+ let table_names: Vec<String> = client
+ .list_tables(namespace)
+ .await?
+ .iter()
+ .map(|t| t.name().to_owned())
+ .collect();
+
+ let mut tables = Vec::new();
+ for name in table_names {
+ let provider = IcebergTableProvider::try_new(client.clone(), namespace, &name).await?;
+ let provider = Arc::new(provider) as Arc<dyn TableProvider>;
+
+ tables.push((name, provider));
+ }
+
+ Ok(IcebergSchemaProvider {
+ tables: tables.into_iter().collect(),
+ })
+ }
+}
+
+impl SchemaProvider for IcebergSchemaProvider {
Review Comment:
Ok great, I'll get started on an `IcebergScan` implementation, or at least plan out some work in more detail and follow up if there's something ambiguous. Once I get something worth looking at, we can either merge it into this fork, or perhaps stack it.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1579940862
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,95 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use dashmap::DashMap;
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: DashMap<String, Arc<dyn SchemaProvider>>,
Review Comment:
> Will the map ever be mutated?
> It might be, I was just not seeing it in this PR.
No, I don't think it will be mutated after we've built the IcebergProvider.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1580505826
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -0,0 +1,79 @@
+// 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.
+
+use std::sync::Arc;
+
+use datafusion::datasource::TableProvider;
+use iceberg::{table::Table, Catalog, NamespaceIdent, Result, TableIdent};
+
+pub(crate) struct IcebergTableProvider {
+ _inner: Table,
+}
+
+impl IcebergTableProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ name: impl Into<String>,
+ ) -> Result<Self> {
+ let name = name.into();
+ let ident = TableIdent::new(namespace, name);
+ let table = client.load_table(&ident).await?;
+
+ Ok(IcebergTableProvider { _inner: table })
+ }
+}
+
+impl TableProvider for IcebergTableProvider {
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef {
+ todo!()
+ }
+
+ fn table_type(&self) -> datafusion::datasource::TableType {
+ todo!()
+ }
+
+ fn scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(
+ &'life0 self,
+ _state: &'life1 datafusion::execution::context::SessionState,
+ _projection: Option<&'life2 Vec<usize>>,
+ _filters: &'life3 [datafusion::prelude::Expr],
+ _limit: Option<usize>,
+ ) -> core::pin::Pin<
+ Box<
+ dyn core::future::Future<
+ Output = datafusion::error::Result<
+ Arc<dyn datafusion::physical_plan::ExecutionPlan>,
+ >,
+ > + core::marker::Send
+ + 'async_trait,
+ >,
+ >
+ where
+ 'life0: 'async_trait,
+ 'life1: 'async_trait,
+ 'life2: 'async_trait,
+ 'life3: 'async_trait,
+ Self: 'async_trait,
+ {
+ todo!()
Review Comment:
I took a look at the `ExecutionPlan` trait, and I think the main part if the `execute` method which could return an arrow record batch stream, which has been implemented by `TableScan::to_arrow`. Other parts, such as partitioning, ordering we can leave it as simple as possible.
I agree that this is not blocker and we can move it to next pr.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1580967711
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,94 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: HashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ /// Asynchronously tries to construct a new [`IcebergCatalogProvider`]
+ /// using the given client to fetch and initialize schema providers for
+ /// each namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves the list of namespace names
+ /// attempts to create a schema provider for each namespace, and
+ /// collects these providers into a concurrent `HashMap`.
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<_> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
+ .collect();
+
+ let providers = try_join_all(
+ schema_names
+ .iter()
+ .map(|name| {
+ IcebergSchemaProvider::try_new(
+ client.clone(),
+ NamespaceIdent::new(name.clone()),
+ )
+ })
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let schemas: Vec<_> = schema_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn SchemaProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergCatalogProvider {
+ schemas: schemas.into_iter().collect(),
+ })
+ }
+}
+
+impl CatalogProvider for IcebergCatalogProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn schema_names(&self) -> Vec<String> {
+ self.schemas.keys().cloned().collect()
Review Comment:
yes, you're right this is a simple but naive impl. based on the [docs ](https://docs.rs/datafusion/latest/datafusion/catalog/trait.CatalogProvider.html#implementing-remote-catalogs) and the ref from delta-rs.
I think having a caching wrapper would be a better solution.
However, since we have to provide a vec of all schema_names we cannot check if a single schema is in the cache and if not fetch again. So I'm guessing we can only implement it with a time based eviction policy i.e. cache all schemas for 1min in order to avoid multiple network calls in a short amount of time (same concept as debouncing)?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#issuecomment-2085319907
I've addressed those issues raised by @Xuanwo - thanks for the review.
I think we can merge this PR for now - and continue working on #357;
especially the caching issues; and proper integration testing with data.
@Fokko @liurenjie1024
I haven't checked now, but from memory I believe py-iceberg uses pyspark to setup tables with actual data for proper integration testing? Would this make sense to tackle such an issue in the next weeks in order to provide proper integration testing? Or should we wait once #349 lands and we have a dedicated crate for e2e testing?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [WIP] Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1561387375
##########
Cargo.toml:
##########
@@ -21,6 +21,7 @@ members = [
"crates/catalog/*",
"crates/examples",
"crates/iceberg",
+ "crates/integrations",
Review Comment:
yes - I think this makes a lot of sense. I'd definitely go for this design. thanks
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "tshauck (via GitHub)" <gi...@apache.org>.
tshauck commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1579818649
##########
crates/integrations/datafusion/Cargo.toml:
##########
@@ -0,0 +1,46 @@
+# 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.
+
+[package]
+name = "iceberg-integrations-datafusion"
+version = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+rust-version = { workspace = true }
+
+categories = ["database"]
+description = "Apache Iceberg Integrations Datafusion"
+repository = { workspace = true }
+license = { workspace = true }
+keywords = ["iceberg", "integrations", "datfusion"]
+
+[features]
+default = []
+datafusion = ["dep:datafusion"]
Review Comment:
Sorry for muddying the waters. I tried to suggest not going the feature route based [here](https://github.com/apache/iceberg-rust/pull/324#discussion_r1562767139) (near the end). I was a little confused by this project having all dependencies in the workspace's Cargo.toml even if only used by one crate.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1580506460
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,73 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use dashmap::DashMap;
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+pub struct IcebergCatalogProvider {
+ schemas: DashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<String> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
Review Comment:
Sounds reasonable to me.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [WIP] Integration with Datafusion [iceberg-rust]
Posted by "tshauck (via GitHub)" <gi...@apache.org>.
tshauck commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1562769346
##########
crates/integrations/src/datafusion/schema.rs:
##########
@@ -0,0 +1,97 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use dashmap::DashMap;
+use datafusion::{
+ catalog::schema::SchemaProvider, datasource::TableProvider, error::DataFusionError,
+};
+use futures::FutureExt;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::datafusion::table::IcebergTableProvider;
+
+pub(crate) struct IcebergSchemaProvider {
+ tables: DashMap<String, Arc<dyn TableProvider>>,
+}
+
+impl IcebergSchemaProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: &NamespaceIdent,
+ ) -> Result<Self> {
+ let table_names: Vec<String> = client
+ .list_tables(namespace)
+ .await?
+ .iter()
+ .map(|t| t.name().to_owned())
+ .collect();
+
+ let mut tables = Vec::new();
+ for name in table_names {
+ let provider = IcebergTableProvider::try_new(client.clone(), namespace, &name).await?;
+ let provider = Arc::new(provider) as Arc<dyn TableProvider>;
+
+ tables.push((name, provider));
+ }
+
+ Ok(IcebergSchemaProvider {
+ tables: tables.into_iter().collect(),
+ })
+ }
+}
+
+impl SchemaProvider for IcebergSchemaProvider {
Review Comment:
Ok great, I'll get started on an `IcebergScan` implementation, or at least plan out some work in more detail and follow up if there's something ambiguous.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1584912205
##########
crates/integrations/datafusion/README.md:
##########
@@ -0,0 +1,22 @@
+<!--
+ ~ 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 Datafusion Integration
+
+This crate contains the integration of Apache Datafusion and Apache Iceberg.
Review Comment:
```suggestion
# Apache Iceberg DataFusion Integration
This crate contains the integration of Apache DataFusion and Apache Iceberg.
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [WIP] Integration with Datafusion [iceberg-rust]
Posted by "tshauck (via GitHub)" <gi...@apache.org>.
tshauck commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1561484967
##########
crates/integrations/src/datafusion/schema.rs:
##########
@@ -0,0 +1,97 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use dashmap::DashMap;
+use datafusion::{
+ catalog::schema::SchemaProvider, datasource::TableProvider, error::DataFusionError,
+};
+use futures::FutureExt;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::datafusion::table::IcebergTableProvider;
+
+pub(crate) struct IcebergSchemaProvider {
+ tables: DashMap<String, Arc<dyn TableProvider>>,
+}
+
+impl IcebergSchemaProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: &NamespaceIdent,
+ ) -> Result<Self> {
+ let table_names: Vec<String> = client
+ .list_tables(namespace)
+ .await?
+ .iter()
+ .map(|t| t.name().to_owned())
+ .collect();
+
+ let mut tables = Vec::new();
+ for name in table_names {
+ let provider = IcebergTableProvider::try_new(client.clone(), namespace, &name).await?;
+ let provider = Arc::new(provider) as Arc<dyn TableProvider>;
+
+ tables.push((name, provider));
+ }
+
+ Ok(IcebergSchemaProvider {
+ tables: tables.into_iter().collect(),
+ })
+ }
+}
+
+impl SchemaProvider for IcebergSchemaProvider {
Review Comment:
Sweet, happy to help... for the catalog data [delta-rs](https://github.com/delta-io/delta-rs/blob/5eade5e1f07e80ecd053bd538e005894b7e88a8f/crates/core/src/data_catalog/unity/datafusion.rs#L140) is probably a good example to crib from. Also, using `async_trait` will improve the aesthetics of `IcebergSchemaProvider::table` and `IcebergTableProvider::scan`.
For execution then (`ExecutionPlan`), to your point it needs to be implemented. Similarly [`DeltaTable`](https://github.com/delta-io/delta-rs/blob/3094bd28cee94b41b502cff6a479965d1e81e990/crates/core/src/delta_datafusion/mod.rs#L654-L705) implements `TableProvider` which uses [`DeltaScan`](https://github.com/delta-io/delta-rs/blob/3094bd28cee94b41b502cff6a479965d1e81e990/crates/core/src/delta_datafusion/mod.rs#L789-L798) that implements `ExecutionPlan`.
What makes sense to you help-wise, so I don't step on your toes? Maybe I could branch off this and take a crack at `IcebergScan` or just commit to this branch or something else?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [WIP] Integration with Datafusion [iceberg-rust]
Posted by "tshauck (via GitHub)" <gi...@apache.org>.
tshauck commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1564807159
##########
crates/integrations/src/datafusion/schema.rs:
##########
@@ -0,0 +1,97 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use dashmap::DashMap;
+use datafusion::{
+ catalog::schema::SchemaProvider, datasource::TableProvider, error::DataFusionError,
+};
+use futures::FutureExt;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::datafusion::table::IcebergTableProvider;
+
+pub(crate) struct IcebergSchemaProvider {
+ tables: DashMap<String, Arc<dyn TableProvider>>,
+}
+
+impl IcebergSchemaProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: &NamespaceIdent,
+ ) -> Result<Self> {
+ let table_names: Vec<String> = client
+ .list_tables(namespace)
+ .await?
+ .iter()
+ .map(|t| t.name().to_owned())
+ .collect();
+
+ let mut tables = Vec::new();
+ for name in table_names {
+ let provider = IcebergTableProvider::try_new(client.clone(), namespace, &name).await?;
+ let provider = Arc::new(provider) as Arc<dyn TableProvider>;
+
+ tables.push((name, provider));
+ }
+
+ Ok(IcebergSchemaProvider {
+ tables: tables.into_iter().collect(),
+ })
+ }
+}
+
+impl SchemaProvider for IcebergSchemaProvider {
Review Comment:
I looked a little more at this, and it seems reasonable to get a first version up and running. Blocker right now is https://github.com/apache/iceberg-rust/pull/277 like you mentioned.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [WIP] Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1562442128
##########
Cargo.toml:
##########
@@ -21,6 +21,7 @@ members = [
"crates/catalog/*",
"crates/examples",
"crates/iceberg",
+ "crates/integrations",
Review Comment:
@tshauck
I updated the structure - to allow a specific crate per integration;
I'm not sure if I handled the feature-flags correctly - so PTAL and let me know what you think.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [WIP] Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1563872514
##########
Cargo.toml:
##########
@@ -21,6 +21,7 @@ members = [
"crates/catalog/*",
"crates/examples",
"crates/iceberg",
+ "crates/integrations",
Review Comment:
> I'm sorry for muddying the waters via the feature suggestion, but then I'm not sure the additional complexity of it would be needed.
Perhaps @liurenjie1024 @Xuanwo you can take another look - whats best practice here in terms of deps?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [WIP] Integration with Datafusion [iceberg-rust]
Posted by "simonvandel (via GitHub)" <gi...@apache.org>.
simonvandel commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1570757993
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,67 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use dashmap::DashMap;
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+pub struct IcebergCatalogProvider {
+ schemas: DashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<String> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
+ .collect();
+
+ let mut schemas = Vec::new();
+ for name in schema_names {
+ let provider =
+ IcebergSchemaProvider::try_new(client.clone(), &NamespaceIdent::new(name.clone()))
+ .await?;
Review Comment:
Drive-by review:
Since `IcebergSchemaProvider::try_new` is async, I'm guessing it might take some time to execute?
In that case, instead of awaiting in a for-loop, perhaps using https://docs.rs/futures/latest/futures/future/fn.join_all.html or similar would make sense?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1581362714
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -0,0 +1,79 @@
+// 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.
+
+use std::sync::Arc;
+
+use datafusion::datasource::TableProvider;
+use iceberg::{table::Table, Catalog, NamespaceIdent, Result, TableIdent};
+
+pub(crate) struct IcebergTableProvider {
+ _inner: Table,
+}
+
+impl IcebergTableProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ name: impl Into<String>,
+ ) -> Result<Self> {
+ let name = name.into();
+ let ident = TableIdent::new(namespace, name);
+ let table = client.load_table(&ident).await?;
+
+ Ok(IcebergTableProvider { _inner: table })
+ }
+}
+
+impl TableProvider for IcebergTableProvider {
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef {
+ todo!()
+ }
+
+ fn table_type(&self) -> datafusion::datasource::TableType {
+ todo!()
+ }
+
+ fn scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(
+ &'life0 self,
+ _state: &'life1 datafusion::execution::context::SessionState,
+ _projection: Option<&'life2 Vec<usize>>,
+ _filters: &'life3 [datafusion::prelude::Expr],
+ _limit: Option<usize>,
+ ) -> core::pin::Pin<
+ Box<
+ dyn core::future::Future<
+ Output = datafusion::error::Result<
+ Arc<dyn datafusion::physical_plan::ExecutionPlan>,
+ >,
+ > + core::marker::Send
+ + 'async_trait,
+ >,
+ >
+ where
+ 'life0: 'async_trait,
+ 'life1: 'async_trait,
+ 'life2: 'async_trait,
+ 'life3: 'async_trait,
+ Self: 'async_trait,
+ {
+ todo!()
Review Comment:
@simonvandel thanks a lot - this looks promising
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1579654225
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -0,0 +1,79 @@
+// 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.
+
+use std::sync::Arc;
+
+use datafusion::datasource::TableProvider;
+use iceberg::{table::Table, Catalog, NamespaceIdent, Result, TableIdent};
+
+pub(crate) struct IcebergTableProvider {
+ _inner: Table,
+}
+
+impl IcebergTableProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ name: impl Into<String>,
+ ) -> Result<Self> {
+ let name = name.into();
+ let ident = TableIdent::new(namespace, name);
+ let table = client.load_table(&ident).await?;
+
+ Ok(IcebergTableProvider { _inner: table })
+ }
+}
+
+impl TableProvider for IcebergTableProvider {
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef {
+ todo!()
+ }
+
+ fn table_type(&self) -> datafusion::datasource::TableType {
+ todo!()
+ }
+
+ fn scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(
Review Comment:
You need to add #[async_trait] to top of implementation
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#issuecomment-2085481329
Thanks for working on this @marvinlanhenke . Looks good to me. There are some more integrations with DataFusion I'm interested to work with like improving ExecutionPlan for IcebergTableScan. Let's move this forward.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1581067924
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -0,0 +1,79 @@
+// 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.
+
+use std::sync::Arc;
+
+use datafusion::datasource::TableProvider;
+use iceberg::{table::Table, Catalog, NamespaceIdent, Result, TableIdent};
+
+pub(crate) struct IcebergTableProvider {
+ _inner: Table,
+}
+
+impl IcebergTableProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ name: impl Into<String>,
+ ) -> Result<Self> {
+ let name = name.into();
+ let ident = TableIdent::new(namespace, name);
+ let table = client.load_table(&ident).await?;
+
+ Ok(IcebergTableProvider { _inner: table })
+ }
+}
+
+impl TableProvider for IcebergTableProvider {
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef {
+ todo!()
+ }
+
+ fn table_type(&self) -> datafusion::datasource::TableType {
+ todo!()
+ }
+
+ fn scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(
+ &'life0 self,
+ _state: &'life1 datafusion::execution::context::SessionState,
+ _projection: Option<&'life2 Vec<usize>>,
+ _filters: &'life3 [datafusion::prelude::Expr],
+ _limit: Option<usize>,
+ ) -> core::pin::Pin<
+ Box<
+ dyn core::future::Future<
+ Output = datafusion::error::Result<
+ Arc<dyn datafusion::physical_plan::ExecutionPlan>,
+ >,
+ > + core::marker::Send
+ + 'async_trait,
+ >,
+ >
+ where
+ 'life0: 'async_trait,
+ 'life1: 'async_trait,
+ 'life2: 'async_trait,
+ 'life3: 'async_trait,
+ Self: 'async_trait,
+ {
+ todo!()
Review Comment:
I don't think this will work?
```rust
fn execute(...) -> Result<SendableRecordBatchStream> {
...
// I can't await here since fn execute() needs to remain sync
// using tokio::Runtime block_on() seems weird?
table_scan.to_arrow().await.unwrap()
// also the return type is BoxStream<'static, Result<RecordBatch>>
// which does not satisfy SendableRecordBatchStream since we need to
// impl RecordBatchStream as well
}
```
...so, how to avoid those issues?
We could try to impl RecordBatchStream on TableScan itself, this however would require us to have datafusion as a dependency in the core and we'd still have the async issue - so I don't think this would be a good solution.
Or we try the same approach as [delta-rs](https://github.com/delta-io/delta-rs/blob/main/crates/core/src/delta_datafusion/mod.rs#L627-L642), and create an actual datafusion physical plan and wrap this:
```rust
struct IcebergTableScan {
plan: Arc<dyn ExecutionPlan>
...
}
impl ExecutionPlan for IcebergTableScan {
...
fn execute(...) -> Result<SendableRecordBatchStream> {
self.plan.execute()
}
}
```
... what do you think @liurenjie1024 ?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1581362714
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -0,0 +1,79 @@
+// 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.
+
+use std::sync::Arc;
+
+use datafusion::datasource::TableProvider;
+use iceberg::{table::Table, Catalog, NamespaceIdent, Result, TableIdent};
+
+pub(crate) struct IcebergTableProvider {
+ _inner: Table,
+}
+
+impl IcebergTableProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ name: impl Into<String>,
+ ) -> Result<Self> {
+ let name = name.into();
+ let ident = TableIdent::new(namespace, name);
+ let table = client.load_table(&ident).await?;
+
+ Ok(IcebergTableProvider { _inner: table })
+ }
+}
+
+impl TableProvider for IcebergTableProvider {
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef {
+ todo!()
+ }
+
+ fn table_type(&self) -> datafusion::datasource::TableType {
+ todo!()
+ }
+
+ fn scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(
+ &'life0 self,
+ _state: &'life1 datafusion::execution::context::SessionState,
+ _projection: Option<&'life2 Vec<usize>>,
+ _filters: &'life3 [datafusion::prelude::Expr],
+ _limit: Option<usize>,
+ ) -> core::pin::Pin<
+ Box<
+ dyn core::future::Future<
+ Output = datafusion::error::Result<
+ Arc<dyn datafusion::physical_plan::ExecutionPlan>,
+ >,
+ > + core::marker::Send
+ + 'async_trait,
+ >,
+ >
+ where
+ 'life0: 'async_trait,
+ 'life1: 'async_trait,
+ 'life2: 'async_trait,
+ 'life3: 'async_trait,
+ Self: 'async_trait,
+ {
+ todo!()
Review Comment:
@simonvandel thanks a lot - this looks promising.
EDIT:
...after some tinkering; especially with the correct return and error types;
I was able to execute the following query and retrieve an actual dataframe:
```rust
let ctx = SessionContext::new();
ctx.register_catalog("hive", catalog);
let df = ctx
.sql("SELECT * FROM hive.dev_hive.my_table")
.await
.unwrap();
//result:
+----+
| c1 |
+----+
| 3 |
| 2 |
| 1 |
+----+
```
I'll guess on monday - I'll have some time to cleanup the code and push it ready for review.
Thanks again for the hint at the correct spot in the docs.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1583019973
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -0,0 +1,123 @@
+// 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.
+
+use std::{any::Any, pin::Pin, sync::Arc};
+
+use datafusion::{
+ arrow::{array::RecordBatch, datatypes::SchemaRef as ArrowSchemaRef},
+ execution::{SendableRecordBatchStream, TaskContext},
+ physical_expr::EquivalenceProperties,
+ physical_plan::{
+ stream::RecordBatchStreamAdapter, DisplayAs, ExecutionMode, ExecutionPlan, Partitioning,
+ PlanProperties,
+ },
+};
+use futures::{Stream, TryStreamExt};
+use iceberg::table::Table;
+
+use crate::to_datafusion_error;
+
+#[derive(Debug)]
+pub(crate) struct IcebergTableScan {
+ table: Table,
+ schema: ArrowSchemaRef,
+ plan_properties: PlanProperties,
+}
+
+impl IcebergTableScan {
+ pub(crate) fn new(table: Table, schema: ArrowSchemaRef) -> Self {
+ let plan_properties = Self::compute_properties(schema.clone());
+
+ Self {
+ table,
+ schema,
+ plan_properties,
+ }
+ }
+
+ fn compute_properties(schema: ArrowSchemaRef) -> PlanProperties {
+ // TODO:
+ // This is more or less a placeholder, to be replaced
+ // once we support output-partitioning
+ PlanProperties::new(
+ EquivalenceProperties::new(schema),
+ Partitioning::UnknownPartitioning(1),
+ ExecutionMode::Bounded,
+ )
+ }
+}
+
+impl ExecutionPlan for IcebergTableScan {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn children(&self) -> Vec<Arc<dyn ExecutionPlan>> {
+ vec![]
+ }
+
+ fn with_new_children(
+ self: Arc<Self>,
+ _children: Vec<Arc<dyn ExecutionPlan>>,
+ ) -> datafusion::error::Result<Arc<dyn ExecutionPlan>> {
+ Ok(self)
Review Comment:
Cool, that sounds reasonable.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#issuecomment-2089449018
> I haven't checked now, but from memory I believe py-iceberg uses pyspark to setup tables with actual data for proper integration testing?
I think so. I think datafusion also python bindings so we can use python for all tests?
I've not reviewed #349 yet, but for datafusion, we can go on without a separate crate for integrate tests?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#issuecomment-2077663647
> I've left some comment to improve, but it looks great! I'll invite datafusion community to help review.
@liurenjie1024
Thanks for the review.
I fixed most of the basic issues regarding structure, naming, async-trait etc.
I'll work on the missing impls over the next couple of days - and see how far we can get with the current state of iceberg-rust.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "simonvandel (via GitHub)" <gi...@apache.org>.
simonvandel commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1579930294
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,95 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use dashmap::DashMap;
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: DashMap<String, Arc<dyn SchemaProvider>>,
Review Comment:
Will the map ever be mutated?
It might be, I was just not seeing it in this PR.
If not, then a std Hashmap seems sufficient to me.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1582080221
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,94 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: HashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ /// Asynchronously tries to construct a new [`IcebergCatalogProvider`]
+ /// using the given client to fetch and initialize schema providers for
+ /// each namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves the list of namespace names
+ /// attempts to create a schema provider for each namespace, and
+ /// collects these providers into a concurrent `HashMap`.
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<_> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
+ .collect();
+
+ let providers = try_join_all(
+ schema_names
+ .iter()
+ .map(|name| {
+ IcebergSchemaProvider::try_new(
+ client.clone(),
+ NamespaceIdent::new(name.clone()),
+ )
+ })
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let schemas: Vec<_> = schema_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn SchemaProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergCatalogProvider {
+ schemas: schemas.into_iter().collect(),
+ })
+ }
+}
+
+impl CatalogProvider for IcebergCatalogProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn schema_names(&self) -> Vec<String> {
+ self.schemas.keys().cloned().collect()
Review Comment:
@liurenjie1024
...I took another look; and I'm not sure we can call the Catalog API directly each time the funtion on the e.g. `trait CatalogProvider` is invoked. The trait is required to be synchronous; however our catalog api calls are async.
So perhaps; we leave it as is for this PR and work on the cache instead of wasting time and effort on another intermediate solution? But perhaps there is an obvious solution to this I'm missing?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#issuecomment-2081477394
> @liurenjie1024 @ZENOTME @viirya @simonvandel ...I just went ahead and pushed the recent updates; PTAL
>
> ## unresolved / todo:
> * [ ] proper integration tests for table scan (requires us to setup a table with actual snapshots / data )
> * [ ] a cache for `IcebergCatalogProvider` and `IcebergSchemaProvider` / so data does not become stale
> * [ ] improve impl ExecutionPlan for IcebergTableScan (once we support filter pushdown, etc.)
Hi, @marvinlanhenke How about opening a tracking issue to track the issues related to datafusion integration? I guess we will have more things to do, and we can edit on that tracking issue.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1582031879
##########
crates/integrations/datafusion/src/schema.rs:
##########
@@ -0,0 +1,95 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use async_trait::async_trait;
+use datafusion::{catalog::schema::SchemaProvider, datasource::TableProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::table::IcebergTableProvider;
+
+/// Represents a [`SchemaProvider`] for the Iceberg [`Catalog`], managing
+/// access to table providers within a specific namespace.
+pub(crate) struct IcebergSchemaProvider {
+ /// A concurrent `HashMap` where keys are table names
+ /// and values are dynamic references to objects implementing the
+ /// [`TableProvider`] trait.
+ tables: HashMap<String, Arc<dyn TableProvider>>,
+}
+
+impl IcebergSchemaProvider {
+ /// Asynchronously tries to construct a new [`IcebergSchemaProvider`]
+ /// using the given client to fetch and initialize table providers for
+ /// the provided namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves a list of table names
+ /// attempts to create a table provider for each table name, and
+ /// collects these providers into a concurrent `HashMap`.
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ ) -> Result<Self> {
+ let table_names: Vec<_> = client
+ .list_tables(&namespace)
+ .await?
+ .iter()
+ .map(|tbl| tbl.name().to_string())
+ .collect();
+
+ let providers = try_join_all(
+ table_names
+ .iter()
+ .map(|name| IcebergTableProvider::try_new(client.clone(), namespace.clone(), name))
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let tables: Vec<_> = table_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn TableProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergSchemaProvider {
+ tables: tables.into_iter().collect(),
+ })
+ }
+}
+
+#[async_trait]
+impl SchemaProvider for IcebergSchemaProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn table_names(&self) -> Vec<String> {
+ self.tables.keys().cloned().collect()
+ }
+
+ fn table_exist(&self, name: &str) -> bool {
+ self.tables.get(name).is_some()
+ }
Review Comment:
These implementations might have similar issue? The table names and table existence is not up to date when calling.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "simonvandel (via GitHub)" <gi...@apache.org>.
simonvandel commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1579897701
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,95 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use dashmap::DashMap;
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: DashMap<String, Arc<dyn SchemaProvider>>,
Review Comment:
Would a std Hashmap suffice? I don't see the concurrency properties of DashMap being used.
I might be missing something though.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1579919751
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,95 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use dashmap::DashMap;
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: DashMap<String, Arc<dyn SchemaProvider>>,
Review Comment:
the datafusion docs provide a link to a reference implementation by [delta-rs](https://github.com/delta-io/delta-rs/blob/951436ecec476ce65b5ed3b58b50fb0846ca7b91/crates/deltalake-core/src/data_catalog/unity/datafusion.rs#L21), that uses a DashMap as well.
My understanding based on the [docs](https://docs.rs/datafusion/latest/datafusion/catalog/trait.CatalogProvider.html#implementing-remote-catalogs) is that datafusion might share the `CatalogProvider` on multiple threads inside an Arc; thus concurrently accessing our "cached" schemas, etc.? Perhaps someone else can clarify this?
> [...] The pattern that DataFusion itself uses to plan SQL queries is to walk over the query to [find all schema / table references in an async function](https://docs.rs/datafusion/latest/datafusion/execution/context/struct.SessionState.html#method.resolve_table_references), performing required remote catalog in parallel, and then plans the query using that snapshot.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "simonvandel (via GitHub)" <gi...@apache.org>.
simonvandel commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1579945951
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,95 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use dashmap::DashMap;
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: DashMap<String, Arc<dyn SchemaProvider>>,
Review Comment:
Cool, well then I think a std Hashmap should be fine, since no actual concurrent operations need to happen
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [WIP] Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#issuecomment-2040961957
Thanks @marvinlanhenke This is amazing, I'll take a review later!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [WIP] Integration with Datafusion [iceberg-rust]
Posted by "tshauck (via GitHub)" <gi...@apache.org>.
tshauck commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1561209884
##########
Cargo.toml:
##########
@@ -21,6 +21,7 @@ members = [
"crates/catalog/*",
"crates/examples",
"crates/iceberg",
+ "crates/integrations",
Review Comment:
Perhaps as a V2 of this integration it'd make sense to use features and/or move the datafusion dep down into the integrations crate (or maybe have a crate per integration).
To your point about polars or other extensions, if this concept grew the dependency load would get challenging for folks who don't need it.
##########
crates/integrations/src/datafusion/schema.rs:
##########
@@ -0,0 +1,97 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use dashmap::DashMap;
+use datafusion::{
+ catalog::schema::SchemaProvider, datasource::TableProvider, error::DataFusionError,
+};
+use futures::FutureExt;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::datafusion::table::IcebergTableProvider;
+
+pub(crate) struct IcebergSchemaProvider {
+ tables: DashMap<String, Arc<dyn TableProvider>>,
+}
+
+impl IcebergSchemaProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: &NamespaceIdent,
+ ) -> Result<Self> {
+ let table_names: Vec<String> = client
+ .list_tables(namespace)
+ .await?
+ .iter()
+ .map(|t| t.name().to_owned())
+ .collect();
+
+ let mut tables = Vec::new();
+ for name in table_names {
+ let provider = IcebergTableProvider::try_new(client.clone(), namespace, &name).await?;
+ let provider = Arc::new(provider) as Arc<dyn TableProvider>;
+
+ tables.push((name, provider));
+ }
+
+ Ok(IcebergSchemaProvider {
+ tables: tables.into_iter().collect(),
+ })
+ }
+}
+
+impl SchemaProvider for IcebergSchemaProvider {
Review Comment:
I realize these are just stubs, I'd be happy to help fill in the guts here and for the `TableProvider` traits.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [WIP] Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#issuecomment-2040961635
@liurenjie1024 @ZENOTME @Fokko
PTAL and let me know what you think.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1579739285
##########
crates/integrations/datafusion/Cargo.toml:
##########
@@ -0,0 +1,46 @@
+# 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.
+
+[package]
+name = "iceberg-integrations-datafusion"
+version = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+rust-version = { workspace = true }
+
+categories = ["database"]
+description = "Apache Iceberg Integrations Datafusion"
+repository = { workspace = true }
+license = { workspace = true }
+keywords = ["iceberg", "integrations", "datfusion"]
+
+[features]
+default = []
+datafusion = ["dep:datafusion"]
Review Comment:
I believe we should have `integrations/datafusion` and `integrations/polars`. So this crate `iceberg-integrations-datafusion` should enable `datafusion` by default, and doesn't need switch.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "simonvandel (via GitHub)" <gi...@apache.org>.
simonvandel commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1581219550
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -0,0 +1,79 @@
+// 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.
+
+use std::sync::Arc;
+
+use datafusion::datasource::TableProvider;
+use iceberg::{table::Table, Catalog, NamespaceIdent, Result, TableIdent};
+
+pub(crate) struct IcebergTableProvider {
+ _inner: Table,
+}
+
+impl IcebergTableProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ name: impl Into<String>,
+ ) -> Result<Self> {
+ let name = name.into();
+ let ident = TableIdent::new(namespace, name);
+ let table = client.load_table(&ident).await?;
+
+ Ok(IcebergTableProvider { _inner: table })
+ }
+}
+
+impl TableProvider for IcebergTableProvider {
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef {
+ todo!()
+ }
+
+ fn table_type(&self) -> datafusion::datasource::TableType {
+ todo!()
+ }
+
+ fn scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(
+ &'life0 self,
+ _state: &'life1 datafusion::execution::context::SessionState,
+ _projection: Option<&'life2 Vec<usize>>,
+ _filters: &'life3 [datafusion::prelude::Expr],
+ _limit: Option<usize>,
+ ) -> core::pin::Pin<
+ Box<
+ dyn core::future::Future<
+ Output = datafusion::error::Result<
+ Arc<dyn datafusion::physical_plan::ExecutionPlan>,
+ >,
+ > + core::marker::Send
+ + 'async_trait,
+ >,
+ >
+ where
+ 'life0: 'async_trait,
+ 'life1: 'async_trait,
+ 'life2: 'async_trait,
+ 'life3: 'async_trait,
+ Self: 'async_trait,
+ {
+ todo!()
Review Comment:
I think you can do what is documented here: https://docs.rs/datafusion/latest/datafusion/physical_plan/trait.ExecutionPlan.html#lazily-async-create-a-stream
So inside `IcebergTableProvider::scan` you create `IcebergScanExec` where `IcebergScanExec` implements `ExecutionPlan`.
The `ExecutionPlan::execute` impl of `IcebergScanExec` would be something like this
```rust
impl ExecutionPlan for IcebergScanExec {
fn execute(
&self,
partition: usize,
context: Arc<TaskContext>
) -> Result<SendableRecordBatchStream> {
let fut = create_batch_stream(self.table_scan);
let stream = futures::stream::once(fut).try_flatten();
Ok(Box::pin(RecordBatchStreamAdapter::new(self.schema.clone(), stream)))
}
}
/// async function that returns a stream
async fn create_batch_stream(table: TableScan) -> Result<SendableRecordBatchStream> {
todo!()
}
```
(Not @liurenjie1024 but hopefully it is helpful - I didn't compile it :) )
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1579883806
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -0,0 +1,79 @@
+// 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.
+
+use std::sync::Arc;
+
+use datafusion::datasource::TableProvider;
+use iceberg::{table::Table, Catalog, NamespaceIdent, Result, TableIdent};
+
+pub(crate) struct IcebergTableProvider {
+ _inner: Table,
+}
+
+impl IcebergTableProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ name: impl Into<String>,
+ ) -> Result<Self> {
+ let name = name.into();
+ let ident = TableIdent::new(namespace, name);
+ let table = client.load_table(&ident).await?;
+
+ Ok(IcebergTableProvider { _inner: table })
+ }
+}
+
+impl TableProvider for IcebergTableProvider {
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef {
+ todo!()
Review Comment:
done
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [WIP] Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#issuecomment-2069378623
> Thanks! Sorry for replying late. I think this is a good start for the integration work. And I have completed #277. Maybe we can convert this PR to ready for review now. @marvinlanhenke
thanks for the feedback - i made some minor changes based on the suggestions; and converted this PR into ready for review. If we agree on the basic design here, we could merge and split the actual impl into multiple issues.
@liurenjie1024 @ZENOTME @Fokko @Xuanwo @sdd @tshauck PTAL - especially, the part about "feature-flags" I have no idea whats best practice in 🦀
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1584229759
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,94 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: HashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ /// Asynchronously tries to construct a new [`IcebergCatalogProvider`]
+ /// using the given client to fetch and initialize schema providers for
+ /// each namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves the list of namespace names
+ /// attempts to create a schema provider for each namespace, and
+ /// collects these providers into a concurrent `HashMap`.
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<_> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
+ .collect();
+
+ let providers = try_join_all(
+ schema_names
+ .iter()
+ .map(|name| {
+ IcebergSchemaProvider::try_new(
+ client.clone(),
+ NamespaceIdent::new(name.clone()),
+ )
+ })
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let schemas: Vec<_> = schema_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn SchemaProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergCatalogProvider {
+ schemas: schemas.into_iter().collect(),
+ })
+ }
+}
+
+impl CatalogProvider for IcebergCatalogProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn schema_names(&self) -> Vec<String> {
+ self.schemas.keys().cloned().collect()
Review Comment:
> is there any issue or discussion already related to that? Or would you mind to quickly outline whats the plan here?
There is a tracking issue to add runtime api: https://github.com/apache/iceberg-rust/issues/124
But I agree with @Xuanwo that we should avoid `rt.block_on` as much as possible.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1582083814
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -0,0 +1,123 @@
+// 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.
+
+use std::{any::Any, pin::Pin, sync::Arc};
+
+use datafusion::{
+ arrow::{array::RecordBatch, datatypes::SchemaRef as ArrowSchemaRef},
+ execution::{SendableRecordBatchStream, TaskContext},
+ physical_expr::EquivalenceProperties,
+ physical_plan::{
+ stream::RecordBatchStreamAdapter, DisplayAs, ExecutionMode, ExecutionPlan, Partitioning,
+ PlanProperties,
+ },
+};
+use futures::{Stream, TryStreamExt};
+use iceberg::table::Table;
+
+use crate::to_datafusion_error;
+
+#[derive(Debug)]
+pub(crate) struct IcebergTableScan {
+ table: Table,
+ schema: ArrowSchemaRef,
+ plan_properties: PlanProperties,
+}
+
+impl IcebergTableScan {
+ pub(crate) fn new(table: Table, schema: ArrowSchemaRef) -> Self {
+ let plan_properties = Self::compute_properties(schema.clone());
+
+ Self {
+ table,
+ schema,
+ plan_properties,
+ }
+ }
+
+ fn compute_properties(schema: ArrowSchemaRef) -> PlanProperties {
+ // TODO:
+ // This is more or less a placeholder, to be replaced
+ // once we support output-partitioning
+ PlanProperties::new(
+ EquivalenceProperties::new(schema),
+ Partitioning::UnknownPartitioning(1),
+ ExecutionMode::Bounded,
+ )
+ }
+}
+
+impl ExecutionPlan for IcebergTableScan {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn children(&self) -> Vec<Arc<dyn ExecutionPlan>> {
+ vec![]
+ }
+
+ fn with_new_children(
+ self: Arc<Self>,
+ _children: Vec<Arc<dyn ExecutionPlan>>,
+ ) -> datafusion::error::Result<Arc<dyn ExecutionPlan>> {
+ Ok(self)
+ }
+
+ fn properties(&self) -> &PlanProperties {
+ &self.plan_properties
+ }
+
+ fn execute(
+ &self,
+ _partition: usize,
+ _context: Arc<TaskContext>,
+ ) -> datafusion::error::Result<SendableRecordBatchStream> {
+ let fut = get_batch_stream(self.table.clone());
+ let stream = futures::stream::once(fut).try_flatten();
+
+ Ok(Box::pin(RecordBatchStreamAdapter::new(
+ self.schema.clone(),
+ stream,
+ )))
+ }
+}
+
+impl DisplayAs for IcebergTableScan {
+ fn fmt_as(
+ &self,
+ _t: datafusion::physical_plan::DisplayFormatType,
+ f: &mut std::fmt::Formatter,
+ ) -> std::fmt::Result {
+ write!(f, "IcebergTableScan")
+ }
+}
+
+async fn get_batch_stream(
+ table: Table,
+) -> datafusion::error::Result<
+ Pin<Box<dyn Stream<Item = datafusion::error::Result<RecordBatch>> + Send>>,
+> {
+ let table_scan = table.scan().build().map_err(to_datafusion_error)?;
+
+ let stream = table_scan
+ .to_arrow()
+ .await
+ .map_err(to_datafusion_error)?
+ .map_err(to_datafusion_error);
Review Comment:
@liurenjie1024 @ZENOTME @viirya
I had no idea how to solve this better; but it still looks weird and redundant.
Why do I need to do this? First `map_err` unwraps the result from `to_arrow()` and the second `map_err` converts the outer result (wrapping the boxed stream) into a `datafusion::error:Result` which is required by `futures::stream::once(fut).try_flatten()`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1584472188
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -0,0 +1,136 @@
+// 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.
+
+use std::{any::Any, pin::Pin, sync::Arc};
+
+use datafusion::{
+ arrow::{array::RecordBatch, datatypes::SchemaRef as ArrowSchemaRef},
+ execution::{SendableRecordBatchStream, TaskContext},
+ physical_expr::EquivalenceProperties,
+ physical_plan::{
+ stream::RecordBatchStreamAdapter, DisplayAs, ExecutionMode, ExecutionPlan, Partitioning,
+ PlanProperties,
+ },
+};
+use futures::{Stream, TryStreamExt};
+use iceberg::table::Table;
+
+use crate::to_datafusion_error;
+
+#[derive(Debug)]
Review Comment:
Seem this feature is still under discussion: https://github.com/rust-lang/rustfmt/issues/3744
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1584513718
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,97 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: HashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ /// Asynchronously tries to construct a new [`IcebergCatalogProvider`]
+ /// using the given client to fetch and initialize schema providers for
+ /// each namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves the list of namespace names
+ /// attempts to create a schema provider for each namespace, and
+ /// collects these providers into a `HashMap`.
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ // TODO:
+ // Schemas and providers should be cached and evicted based on time
+ // As of right now; schemas might become stale.
Review Comment:
> _There are 2 hard problems in computer science: cache invalidation, naming things, and off-by-1 errors._ - Leon Bambrick
😄the classic.
I think leaving it up to the user, leads us to the issue about blocking an async call in a sync trait function? I think if we have an idea how to handle this, we can better reason about if, when, and where to cache?
from the docs:
> To implement [CatalogProvider](https://docs.rs/datafusion/latest/datafusion/catalog/trait.CatalogProvider.html) and [SchemaProvider](https://docs.rs/datafusion/latest/datafusion/catalog/schema/trait.SchemaProvider.html) for remote catalogs, you need to provide an in memory snapshot of the required metadata. Most systems typically either already have this information cached locally or can batch access to the remote catalog to retrieve multiple schemas and tables in a single network call.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1582159023
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -0,0 +1,123 @@
+// 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.
+
+use std::{any::Any, pin::Pin, sync::Arc};
+
+use datafusion::{
+ arrow::{array::RecordBatch, datatypes::SchemaRef as ArrowSchemaRef},
+ execution::{SendableRecordBatchStream, TaskContext},
+ physical_expr::EquivalenceProperties,
+ physical_plan::{
+ stream::RecordBatchStreamAdapter, DisplayAs, ExecutionMode, ExecutionPlan, Partitioning,
+ PlanProperties,
+ },
+};
+use futures::{Stream, TryStreamExt};
+use iceberg::table::Table;
+
+use crate::to_datafusion_error;
+
+#[derive(Debug)]
+pub(crate) struct IcebergTableScan {
+ table: Table,
+ schema: ArrowSchemaRef,
+ plan_properties: PlanProperties,
+}
+
+impl IcebergTableScan {
+ pub(crate) fn new(table: Table, schema: ArrowSchemaRef) -> Self {
+ let plan_properties = Self::compute_properties(schema.clone());
+
+ Self {
+ table,
+ schema,
+ plan_properties,
+ }
+ }
+
+ fn compute_properties(schema: ArrowSchemaRef) -> PlanProperties {
+ // TODO:
+ // This is more or less a placeholder, to be replaced
+ // once we support output-partitioning
+ PlanProperties::new(
+ EquivalenceProperties::new(schema),
+ Partitioning::UnknownPartitioning(1),
+ ExecutionMode::Bounded,
+ )
+ }
+}
+
+impl ExecutionPlan for IcebergTableScan {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn children(&self) -> Vec<Arc<dyn ExecutionPlan>> {
+ vec![]
+ }
+
+ fn with_new_children(
+ self: Arc<Self>,
+ _children: Vec<Arc<dyn ExecutionPlan>>,
+ ) -> datafusion::error::Result<Arc<dyn ExecutionPlan>> {
+ Ok(self)
Review Comment:
I thought the same; then I checked `CsvExec` in Datafusion since it's also a leaf-node that should also not have any children. turns out they do the same and return `Ok(self)` as well. [link](https://github.com/apache/datafusion/blob/main/datafusion/core/src/datasource/physical_plan/csv.rs#L181-L186)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1582055651
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,94 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: HashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ /// Asynchronously tries to construct a new [`IcebergCatalogProvider`]
+ /// using the given client to fetch and initialize schema providers for
+ /// each namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves the list of namespace names
+ /// attempts to create a schema provider for each namespace, and
+ /// collects these providers into a concurrent `HashMap`.
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<_> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
+ .collect();
+
+ let providers = try_join_all(
+ schema_names
+ .iter()
+ .map(|name| {
+ IcebergSchemaProvider::try_new(
+ client.clone(),
+ NamespaceIdent::new(name.clone()),
+ )
+ })
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let schemas: Vec<_> = schema_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn SchemaProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergCatalogProvider {
+ schemas: schemas.into_iter().collect(),
+ })
+ }
+}
+
+impl CatalogProvider for IcebergCatalogProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn schema_names(&self) -> Vec<String> {
+ self.schemas.keys().cloned().collect()
Review Comment:
sure, we can do it like that for this PR and impl. the more sophisticated solution later.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1579827548
##########
crates/integrations/datafusion/Cargo.toml:
##########
@@ -0,0 +1,46 @@
+# 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.
+
+[package]
+name = "iceberg-integrations-datafusion"
+version = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+rust-version = { workspace = true }
+
+categories = ["database"]
+description = "Apache Iceberg Integrations Datafusion"
+repository = { workspace = true }
+license = { workspace = true }
+keywords = ["iceberg", "integrations", "datfusion"]
+
+[features]
+default = []
+datafusion = ["dep:datafusion"]
Review Comment:
> I believe we should have `integrations/datafusion` and `integrations/polars`. So this crate `iceberg-integrations-datafusion` should enable `datafusion` by default, and doesn't need switch.
@Xuanwo
thank you - then I guess my last commit fixed it.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1584505866
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,94 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: HashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ /// Asynchronously tries to construct a new [`IcebergCatalogProvider`]
+ /// using the given client to fetch and initialize schema providers for
+ /// each namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves the list of namespace names
+ /// attempts to create a schema provider for each namespace, and
+ /// collects these providers into a concurrent `HashMap`.
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<_> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
+ .collect();
+
+ let providers = try_join_all(
+ schema_names
+ .iter()
+ .map(|name| {
+ IcebergSchemaProvider::try_new(
+ client.clone(),
+ NamespaceIdent::new(name.clone()),
+ )
+ })
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let schemas: Vec<_> = schema_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn SchemaProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergCatalogProvider {
+ schemas: schemas.into_iter().collect(),
+ })
+ }
+}
+
+impl CatalogProvider for IcebergCatalogProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn schema_names(&self) -> Vec<String> {
+ self.schemas.keys().cloned().collect()
Review Comment:
> This makes sense, but I see the issue with blocking on async calls. At first, I would take the price of waiting for the blocking calls. Even it is still a remote call, the ones to the REST catalog should be lightning-fast (that's where the caching happens).
I think blocking + timeout would be a reasonable solution before we implement sophiscated caching. For this pr, we can leave it as now since it's even not caching, it's a snapshot. WDYT?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#issuecomment-2089469920
Thanks @marvinlanhenke for this pr, and @Fokko @Xuanwo @viirya @simonvandel @tshauck for review!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1582152303
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -0,0 +1,123 @@
+// 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.
+
+use std::{any::Any, pin::Pin, sync::Arc};
+
+use datafusion::{
+ arrow::{array::RecordBatch, datatypes::SchemaRef as ArrowSchemaRef},
+ execution::{SendableRecordBatchStream, TaskContext},
+ physical_expr::EquivalenceProperties,
+ physical_plan::{
+ stream::RecordBatchStreamAdapter, DisplayAs, ExecutionMode, ExecutionPlan, Partitioning,
+ PlanProperties,
+ },
+};
+use futures::{Stream, TryStreamExt};
+use iceberg::table::Table;
+
+use crate::to_datafusion_error;
+
+#[derive(Debug)]
+pub(crate) struct IcebergTableScan {
+ table: Table,
+ schema: ArrowSchemaRef,
+ plan_properties: PlanProperties,
+}
+
+impl IcebergTableScan {
+ pub(crate) fn new(table: Table, schema: ArrowSchemaRef) -> Self {
+ let plan_properties = Self::compute_properties(schema.clone());
+
+ Self {
+ table,
+ schema,
+ plan_properties,
+ }
+ }
+
+ fn compute_properties(schema: ArrowSchemaRef) -> PlanProperties {
+ // TODO:
+ // This is more or less a placeholder, to be replaced
+ // once we support output-partitioning
+ PlanProperties::new(
+ EquivalenceProperties::new(schema),
+ Partitioning::UnknownPartitioning(1),
+ ExecutionMode::Bounded,
+ )
+ }
+}
+
+impl ExecutionPlan for IcebergTableScan {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn children(&self) -> Vec<Arc<dyn ExecutionPlan>> {
+ vec![]
+ }
+
+ fn with_new_children(
+ self: Arc<Self>,
+ _children: Vec<Arc<dyn ExecutionPlan>>,
+ ) -> datafusion::error::Result<Arc<dyn ExecutionPlan>> {
+ Ok(self)
+ }
+
+ fn properties(&self) -> &PlanProperties {
+ &self.plan_properties
+ }
+
+ fn execute(
+ &self,
+ _partition: usize,
+ _context: Arc<TaskContext>,
+ ) -> datafusion::error::Result<SendableRecordBatchStream> {
+ let fut = get_batch_stream(self.table.clone());
+ let stream = futures::stream::once(fut).try_flatten();
+
+ Ok(Box::pin(RecordBatchStreamAdapter::new(
+ self.schema.clone(),
+ stream,
+ )))
+ }
+}
+
+impl DisplayAs for IcebergTableScan {
+ fn fmt_as(
+ &self,
+ _t: datafusion::physical_plan::DisplayFormatType,
+ f: &mut std::fmt::Formatter,
+ ) -> std::fmt::Result {
+ write!(f, "IcebergTableScan")
+ }
+}
+
+async fn get_batch_stream(
+ table: Table,
+) -> datafusion::error::Result<
+ Pin<Box<dyn Stream<Item = datafusion::error::Result<RecordBatch>> + Send>>,
+> {
+ let table_scan = table.scan().build().map_err(to_datafusion_error)?;
+
+ let stream = table_scan
+ .to_arrow()
+ .await
+ .map_err(to_datafusion_error)?
+ .map_err(to_datafusion_error);
Review Comment:
I think it's ok for now. There is [flatten](https://doc.rust-lang.org/std/result/enum.Result.html#method.flatten) api to simplify it, but it's not stable. We can refine it in future.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1582152848
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -0,0 +1,123 @@
+// 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.
+
+use std::{any::Any, pin::Pin, sync::Arc};
+
+use datafusion::{
+ arrow::{array::RecordBatch, datatypes::SchemaRef as ArrowSchemaRef},
+ execution::{SendableRecordBatchStream, TaskContext},
+ physical_expr::EquivalenceProperties,
+ physical_plan::{
+ stream::RecordBatchStreamAdapter, DisplayAs, ExecutionMode, ExecutionPlan, Partitioning,
+ PlanProperties,
+ },
+};
+use futures::{Stream, TryStreamExt};
+use iceberg::table::Table;
+
+use crate::to_datafusion_error;
+
+#[derive(Debug)]
+pub(crate) struct IcebergTableScan {
+ table: Table,
+ schema: ArrowSchemaRef,
+ plan_properties: PlanProperties,
+}
+
+impl IcebergTableScan {
+ pub(crate) fn new(table: Table, schema: ArrowSchemaRef) -> Self {
+ let plan_properties = Self::compute_properties(schema.clone());
+
+ Self {
+ table,
+ schema,
+ plan_properties,
+ }
+ }
+
+ fn compute_properties(schema: ArrowSchemaRef) -> PlanProperties {
+ // TODO:
+ // This is more or less a placeholder, to be replaced
+ // once we support output-partitioning
+ PlanProperties::new(
+ EquivalenceProperties::new(schema),
+ Partitioning::UnknownPartitioning(1),
+ ExecutionMode::Bounded,
+ )
+ }
+}
+
+impl ExecutionPlan for IcebergTableScan {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn children(&self) -> Vec<Arc<dyn ExecutionPlan>> {
+ vec![]
+ }
+
+ fn with_new_children(
+ self: Arc<Self>,
+ _children: Vec<Arc<dyn ExecutionPlan>>,
+ ) -> datafusion::error::Result<Arc<dyn ExecutionPlan>> {
+ Ok(self)
Review Comment:
I think this should never be called, how about throwing an error here?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#issuecomment-2081482903
Let's wait a moment to see what others think about the catalog snapshot problem. cc @Xuanwo @Fokko @viirya @sdd PTAL
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1584789483
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -0,0 +1,136 @@
+// 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.
+
+use std::{any::Any, pin::Pin, sync::Arc};
+
+use datafusion::{
+ arrow::{array::RecordBatch, datatypes::SchemaRef as ArrowSchemaRef},
+ execution::{SendableRecordBatchStream, TaskContext},
+ physical_expr::EquivalenceProperties,
+ physical_plan::{
+ stream::RecordBatchStreamAdapter, DisplayAs, ExecutionMode, ExecutionPlan, Partitioning,
+ PlanProperties,
+ },
+};
+use futures::{Stream, TryStreamExt};
+use iceberg::table::Table;
+
+use crate::to_datafusion_error;
+
+#[derive(Debug)]
Review Comment:
> Seem this feature is still under discussion: [rust-lang/rustfmt#3744](https://github.com/rust-lang/rustfmt/issues/3744)
uff with no update since 2020?
I rearranged the stmt to go with the suggested approach.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1584781574
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,94 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: HashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ /// Asynchronously tries to construct a new [`IcebergCatalogProvider`]
+ /// using the given client to fetch and initialize schema providers for
+ /// each namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves the list of namespace names
+ /// attempts to create a schema provider for each namespace, and
+ /// collects these providers into a concurrent `HashMap`.
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<_> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
+ .collect();
+
+ let providers = try_join_all(
+ schema_names
+ .iter()
+ .map(|name| {
+ IcebergSchemaProvider::try_new(
+ client.clone(),
+ NamespaceIdent::new(name.clone()),
+ )
+ })
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let schemas: Vec<_> = schema_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn SchemaProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergCatalogProvider {
+ schemas: schemas.into_iter().collect(),
+ })
+ }
+}
+
+impl CatalogProvider for IcebergCatalogProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn schema_names(&self) -> Vec<String> {
+ self.schemas.keys().cloned().collect()
Review Comment:
> For this pr, we can leave it as now since it's even not caching, it's a snapshot. WDYT?
+1 for "moving on" and create or track those issues in #357 before this PR gets too big.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1579715245
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -0,0 +1,79 @@
+// 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.
+
+use std::sync::Arc;
+
+use datafusion::datasource::TableProvider;
+use iceberg::{table::Table, Catalog, NamespaceIdent, Result, TableIdent};
+
+pub(crate) struct IcebergTableProvider {
+ _inner: Table,
+}
+
+impl IcebergTableProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ name: impl Into<String>,
+ ) -> Result<Self> {
+ let name = name.into();
+ let ident = TableIdent::new(namespace, name);
+ let table = client.load_table(&ident).await?;
+
+ Ok(IcebergTableProvider { _inner: table })
+ }
+}
+
+impl TableProvider for IcebergTableProvider {
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef {
+ todo!()
Review Comment:
yes; i believe we merged this 4 days ago; so i can take a look at this again now
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1579713512
##########
crates/integrations/datafusion/Cargo.toml:
##########
@@ -0,0 +1,46 @@
+# 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.
+
+[package]
+name = "iceberg-integrations-datafusion"
+version = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+rust-version = { workspace = true }
+
+categories = ["database"]
+description = "Apache Iceberg Integrations Datafusion"
+repository = { workspace = true }
+license = { workspace = true }
+keywords = ["iceberg", "integrations", "datfusion"]
+
+[features]
+default = []
+datafusion = ["dep:datafusion"]
Review Comment:
Here I'm probably even more confused; I tried this based on the review by @tshauck - but I have no experience with this and I'm not sure how to structure the crate, in order to allow a possible end-user to download only the integration dependency he needs e.g. iceberg-datafusion or iceberg-polars. etc.
So here I'm hoping for feedback and help.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1584791992
##########
crates/integrations/datafusion/src/schema.rs:
##########
@@ -0,0 +1,99 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use async_trait::async_trait;
+use datafusion::{catalog::schema::SchemaProvider, datasource::TableProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::table::IcebergTableProvider;
+
+/// Represents a [`SchemaProvider`] for the Iceberg [`Catalog`], managing
+/// access to table providers within a specific namespace.
+pub(crate) struct IcebergSchemaProvider {
+ /// A `HashMap` where keys are table names
+ /// and values are dynamic references to objects implementing the
+ /// [`TableProvider`] trait.
+ tables: HashMap<String, Arc<dyn TableProvider>>,
+}
+
+impl IcebergSchemaProvider {
+ /// Asynchronously tries to construct a new [`IcebergSchemaProvider`]
+ /// using the given client to fetch and initialize table providers for
+ /// the provided namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves a list of table names
+ /// attempts to create a table provider for each table name, and
+ /// collects these providers into a `HashMap`.
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ ) -> Result<Self> {
+ // TODO:
+ // Tables and providers should be cached based on table_name
+ // if we have a cache miss; we update our internal cache & check again
+ // As of right now; tables might become stale.
+ let table_names: Vec<_> = client
+ .list_tables(&namespace)
+ .await?
+ .iter()
+ .map(|tbl| tbl.name().to_string())
+ .collect();
+
+ let providers = try_join_all(
+ table_names
+ .iter()
+ .map(|name| IcebergTableProvider::try_new(client.clone(), namespace.clone(), name))
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let tables: Vec<_> = table_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn TableProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergSchemaProvider {
+ tables: tables.into_iter().collect(),
Review Comment:
yes, this was a left-over from before where I used DashMap. thanks for spotting that.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 merged PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1584230121
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -0,0 +1,136 @@
+// 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.
+
+use std::{any::Any, pin::Pin, sync::Arc};
+
+use datafusion::{
+ arrow::{array::RecordBatch, datatypes::SchemaRef as ArrowSchemaRef},
+ execution::{SendableRecordBatchStream, TaskContext},
+ physical_expr::EquivalenceProperties,
+ physical_plan::{
+ stream::RecordBatchStreamAdapter, DisplayAs, ExecutionMode, ExecutionPlan, Partitioning,
+ PlanProperties,
+ },
+};
+use futures::{Stream, TryStreamExt};
+use iceberg::table::Table;
+
+use crate::to_datafusion_error;
+
+#[derive(Debug)]
Review Comment:
nit: I personaly prefer put `#[derive]` under the doc comment for better reading.
Feel free to ignore this if you think it's fine.
##########
crates/integrations/datafusion/src/schema.rs:
##########
@@ -0,0 +1,99 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use async_trait::async_trait;
+use datafusion::{catalog::schema::SchemaProvider, datasource::TableProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::table::IcebergTableProvider;
+
+/// Represents a [`SchemaProvider`] for the Iceberg [`Catalog`], managing
+/// access to table providers within a specific namespace.
+pub(crate) struct IcebergSchemaProvider {
+ /// A `HashMap` where keys are table names
+ /// and values are dynamic references to objects implementing the
+ /// [`TableProvider`] trait.
+ tables: HashMap<String, Arc<dyn TableProvider>>,
+}
+
+impl IcebergSchemaProvider {
+ /// Asynchronously tries to construct a new [`IcebergSchemaProvider`]
+ /// using the given client to fetch and initialize table providers for
+ /// the provided namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves a list of table names
+ /// attempts to create a table provider for each table name, and
+ /// collects these providers into a `HashMap`.
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ ) -> Result<Self> {
+ // TODO:
+ // Tables and providers should be cached based on table_name
+ // if we have a cache miss; we update our internal cache & check again
+ // As of right now; tables might become stale.
+ let table_names: Vec<_> = client
+ .list_tables(&namespace)
+ .await?
+ .iter()
+ .map(|tbl| tbl.name().to_string())
+ .collect();
+
+ let providers = try_join_all(
+ table_names
+ .iter()
+ .map(|name| IcebergTableProvider::try_new(client.clone(), namespace.clone(), name))
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let tables: Vec<_> = table_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn TableProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergSchemaProvider {
+ tables: tables.into_iter().collect(),
+ })
+ }
+}
+
+#[async_trait]
+impl SchemaProvider for IcebergSchemaProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn table_names(&self) -> Vec<String> {
+ self.tables.keys().cloned().collect()
+ }
+
+ fn table_exist(&self, name: &str) -> bool {
+ self.tables.get(name).is_some()
+ }
+
+ async fn table(&self, name: &str) -> datafusion::error::Result<Option<Arc<dyn TableProvider>>> {
Review Comment:
We can `use datafusion::error::Result as DfResult`, what do you think?
##########
crates/integrations/datafusion/src/schema.rs:
##########
@@ -0,0 +1,99 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use async_trait::async_trait;
+use datafusion::{catalog::schema::SchemaProvider, datasource::TableProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::table::IcebergTableProvider;
+
+/// Represents a [`SchemaProvider`] for the Iceberg [`Catalog`], managing
+/// access to table providers within a specific namespace.
+pub(crate) struct IcebergSchemaProvider {
+ /// A `HashMap` where keys are table names
+ /// and values are dynamic references to objects implementing the
+ /// [`TableProvider`] trait.
+ tables: HashMap<String, Arc<dyn TableProvider>>,
+}
+
+impl IcebergSchemaProvider {
+ /// Asynchronously tries to construct a new [`IcebergSchemaProvider`]
+ /// using the given client to fetch and initialize table providers for
+ /// the provided namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves a list of table names
+ /// attempts to create a table provider for each table name, and
+ /// collects these providers into a `HashMap`.
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ ) -> Result<Self> {
+ // TODO:
+ // Tables and providers should be cached based on table_name
+ // if we have a cache miss; we update our internal cache & check again
+ // As of right now; tables might become stale.
+ let table_names: Vec<_> = client
+ .list_tables(&namespace)
+ .await?
+ .iter()
+ .map(|tbl| tbl.name().to_string())
+ .collect();
+
+ let providers = try_join_all(
+ table_names
+ .iter()
+ .map(|name| IcebergTableProvider::try_new(client.clone(), namespace.clone(), name))
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let tables: Vec<_> = table_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn TableProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergSchemaProvider {
+ tables: tables.into_iter().collect(),
Review Comment:
Seems we can collect into `HashMap` directly without into `Vec<_>` first?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1584803007
##########
crates/integrations/datafusion/src/schema.rs:
##########
@@ -0,0 +1,99 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use async_trait::async_trait;
+use datafusion::{catalog::schema::SchemaProvider, datasource::TableProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::table::IcebergTableProvider;
+
+/// Represents a [`SchemaProvider`] for the Iceberg [`Catalog`], managing
+/// access to table providers within a specific namespace.
+pub(crate) struct IcebergSchemaProvider {
+ /// A `HashMap` where keys are table names
+ /// and values are dynamic references to objects implementing the
+ /// [`TableProvider`] trait.
+ tables: HashMap<String, Arc<dyn TableProvider>>,
+}
+
+impl IcebergSchemaProvider {
+ /// Asynchronously tries to construct a new [`IcebergSchemaProvider`]
+ /// using the given client to fetch and initialize table providers for
+ /// the provided namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves a list of table names
+ /// attempts to create a table provider for each table name, and
+ /// collects these providers into a `HashMap`.
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ ) -> Result<Self> {
+ // TODO:
+ // Tables and providers should be cached based on table_name
+ // if we have a cache miss; we update our internal cache & check again
+ // As of right now; tables might become stale.
+ let table_names: Vec<_> = client
+ .list_tables(&namespace)
+ .await?
+ .iter()
+ .map(|tbl| tbl.name().to_string())
+ .collect();
+
+ let providers = try_join_all(
+ table_names
+ .iter()
+ .map(|name| IcebergTableProvider::try_new(client.clone(), namespace.clone(), name))
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let tables: Vec<_> = table_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn TableProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergSchemaProvider {
+ tables: tables.into_iter().collect(),
+ })
+ }
+}
+
+#[async_trait]
+impl SchemaProvider for IcebergSchemaProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn table_names(&self) -> Vec<String> {
+ self.tables.keys().cloned().collect()
+ }
+
+ fn table_exist(&self, name: &str) -> bool {
+ self.tables.get(name).is_some()
+ }
+
+ async fn table(&self, name: &str) -> datafusion::error::Result<Option<Arc<dyn TableProvider>>> {
Review Comment:
yes - i like that. fixed.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1584368969
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -0,0 +1,136 @@
+// 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.
+
+use std::{any::Any, pin::Pin, sync::Arc};
+
+use datafusion::{
+ arrow::{array::RecordBatch, datatypes::SchemaRef as ArrowSchemaRef},
+ execution::{SendableRecordBatchStream, TaskContext},
+ physical_expr::EquivalenceProperties,
+ physical_plan::{
+ stream::RecordBatchStreamAdapter, DisplayAs, ExecutionMode, ExecutionPlan, Partitioning,
+ PlanProperties,
+ },
+};
+use futures::{Stream, TryStreamExt};
+use iceberg::table::Table;
+
+use crate::to_datafusion_error;
+
+#[derive(Debug)]
Review Comment:
> nit: I personaly prefer put `#[derive]` under the doc comment for better reading.
>
> Feel free to ignore this if you think it's fine.
I often asked myself whats best-practice here - I'm fine with both approaches.
+1 for enforcing this with a linter (if possible)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1584213613
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,94 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: HashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ /// Asynchronously tries to construct a new [`IcebergCatalogProvider`]
+ /// using the given client to fetch and initialize schema providers for
+ /// each namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves the list of namespace names
+ /// attempts to create a schema provider for each namespace, and
+ /// collects these providers into a concurrent `HashMap`.
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<_> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
+ .collect();
+
+ let providers = try_join_all(
+ schema_names
+ .iter()
+ .map(|name| {
+ IcebergSchemaProvider::try_new(
+ client.clone(),
+ NamespaceIdent::new(name.clone()),
+ )
+ })
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let schemas: Vec<_> = schema_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn SchemaProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergCatalogProvider {
+ schemas: schemas.into_iter().collect(),
+ })
+ }
+}
+
+impl CatalogProvider for IcebergCatalogProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn schema_names(&self) -> Vec<String> {
+ self.schemas.keys().cloned().collect()
Review Comment:
> It's possible to block on an async call, but it requires runtime api.
It does possible but we should avoid using `rt.block_on()` in our lib.
> My suggestion is to leave it as it now, and fix it when we have runtime api landed.
I'm second with this suggestion. We can move on and figure how to improve this part in the future.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1582042879
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -0,0 +1,79 @@
+// 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.
+
+use std::sync::Arc;
+
+use datafusion::datasource::TableProvider;
+use iceberg::{table::Table, Catalog, NamespaceIdent, Result, TableIdent};
+
+pub(crate) struct IcebergTableProvider {
+ _inner: Table,
+}
+
+impl IcebergTableProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ name: impl Into<String>,
+ ) -> Result<Self> {
+ let name = name.into();
+ let ident = TableIdent::new(namespace, name);
+ let table = client.load_table(&ident).await?;
+
+ Ok(IcebergTableProvider { _inner: table })
+ }
+}
+
+impl TableProvider for IcebergTableProvider {
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef {
+ todo!()
+ }
+
+ fn table_type(&self) -> datafusion::datasource::TableType {
+ todo!()
+ }
+
+ fn scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(
+ &'life0 self,
+ _state: &'life1 datafusion::execution::context::SessionState,
+ _projection: Option<&'life2 Vec<usize>>,
+ _filters: &'life3 [datafusion::prelude::Expr],
+ _limit: Option<usize>,
+ ) -> core::pin::Pin<
+ Box<
+ dyn core::future::Future<
+ Output = datafusion::error::Result<
+ Arc<dyn datafusion::physical_plan::ExecutionPlan>,
+ >,
+ > + core::marker::Send
+ + 'async_trait,
+ >,
+ >
+ where
+ 'life0: 'async_trait,
+ 'life1: 'async_trait,
+ 'life2: 'async_trait,
+ 'life3: 'async_trait,
+ Self: 'async_trait,
+ {
+ todo!()
Review Comment:
Thanks @simonvandel for the comments! Yeah, that looks great to me.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1582080221
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,94 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: HashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ /// Asynchronously tries to construct a new [`IcebergCatalogProvider`]
+ /// using the given client to fetch and initialize schema providers for
+ /// each namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves the list of namespace names
+ /// attempts to create a schema provider for each namespace, and
+ /// collects these providers into a concurrent `HashMap`.
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<_> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
+ .collect();
+
+ let providers = try_join_all(
+ schema_names
+ .iter()
+ .map(|name| {
+ IcebergSchemaProvider::try_new(
+ client.clone(),
+ NamespaceIdent::new(name.clone()),
+ )
+ })
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let schemas: Vec<_> = schema_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn SchemaProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergCatalogProvider {
+ schemas: schemas.into_iter().collect(),
+ })
+ }
+}
+
+impl CatalogProvider for IcebergCatalogProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn schema_names(&self) -> Vec<String> {
+ self.schemas.keys().cloned().collect()
Review Comment:
@liurenjie1024
...I took another look; and I'm not sure we can call the Catalog API directly each time the funtion on the e.g. `trait CatalogProvider` is invoked. The trait is required to be synchronous; however our catalog api calls are async.
So perhaps; we leave it as is for this PR and work on the cache instead? However, I'll guess we're facing the same problem here as well? We could use `rt.block_on(...)` to solve this, but maybe, I'm just missing something and we can do this without needing to block?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [WIP] Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1574731993
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,67 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use dashmap::DashMap;
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+pub struct IcebergCatalogProvider {
+ schemas: DashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<String> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
+ .collect();
+
+ let mut schemas = Vec::new();
+ for name in schema_names {
+ let provider =
+ IcebergSchemaProvider::try_new(client.clone(), &NamespaceIdent::new(name.clone()))
+ .await?;
Review Comment:
thanks for the drive-by 😄 - i'll guess i fixed it, but since this is only a draft there might be some refactoring neccessary
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1579601294
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -0,0 +1,79 @@
+// 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.
+
+use std::sync::Arc;
+
+use datafusion::datasource::TableProvider;
+use iceberg::{table::Table, Catalog, NamespaceIdent, Result, TableIdent};
+
+pub(crate) struct IcebergTableProvider {
+ _inner: Table,
+}
+
+impl IcebergTableProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ name: impl Into<String>,
+ ) -> Result<Self> {
+ let name = name.into();
+ let ident = TableIdent::new(namespace, name);
+ let table = client.load_table(&ident).await?;
+
+ Ok(IcebergTableProvider { _inner: table })
+ }
+}
+
+impl TableProvider for IcebergTableProvider {
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef {
+ todo!()
+ }
+
+ fn table_type(&self) -> datafusion::datasource::TableType {
+ todo!()
+ }
+
+ fn scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(
Review Comment:
We should remove these auto generate lifetimes
##########
crates/integrations/datafusion/README.md:
##########
@@ -0,0 +1,22 @@
+<!--
+ ~ 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 Integrations
+
+This crate contains the official Native Rust implementation of Apache Iceberg Integrations.
Review Comment:
```suggestion
This crate contains integration of apache datafusion and apache iceberg.
```
##########
crates/integrations/datafusion/Cargo.toml:
##########
@@ -0,0 +1,46 @@
+# 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.
+
+[package]
+name = "iceberg-integrations-datafusion"
+version = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+rust-version = { workspace = true }
+
+categories = ["database"]
+description = "Apache Iceberg Integrations Datafusion"
+repository = { workspace = true }
+license = { workspace = true }
+keywords = ["iceberg", "integrations", "datfusion"]
+
+[features]
+default = []
+datafusion = ["dep:datafusion"]
Review Comment:
Sorry I'm a little confusing here, as a crate for integration with datafusion, why this is a feature rather enabled by default?
##########
crates/integrations/datafusion/Cargo.toml:
##########
@@ -0,0 +1,46 @@
+# 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.
+
+[package]
+name = "iceberg-integrations-datafusion"
Review Comment:
```suggestion
name = "iceberg-datafusion"
```
How about just this?
##########
crates/integrations/datafusion/src/schema.rs:
##########
@@ -0,0 +1,102 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use dashmap::DashMap;
+use datafusion::{
+ catalog::schema::SchemaProvider, datasource::TableProvider, error::DataFusionError,
+};
+use futures::{future::try_join_all, FutureExt};
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::table::IcebergTableProvider;
+
+pub(crate) struct IcebergSchemaProvider {
+ tables: DashMap<String, Arc<dyn TableProvider>>,
+}
+
+impl IcebergSchemaProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ ) -> Result<Self> {
+ let table_names: Vec<String> = client
+ .list_tables(&namespace)
+ .await?
+ .iter()
+ .map(|t| t.name().to_owned())
+ .collect();
+
+ let futures: Vec<_> = table_names
+ .iter()
+ .map(|name| IcebergTableProvider::try_new(client.clone(), namespace.clone(), name))
+ .collect();
+
+ let providers = try_join_all(futures).await?;
+
+ let mut tables = Vec::new();
+ for (name, provider) in table_names.into_iter().zip(providers.into_iter()) {
+ let provider = Arc::new(provider) as Arc<dyn TableProvider>;
+ tables.push((name, provider));
+ }
+
+ Ok(IcebergSchemaProvider {
+ tables: tables.into_iter().collect(),
+ })
+ }
+}
+
+impl SchemaProvider for IcebergSchemaProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn table_names(&self) -> Vec<String> {
+ self.tables.iter().map(|c| c.key().clone()).collect()
+ }
+
+ fn table_exist(&self, name: &str) -> bool {
+ self.tables.get(name).is_some()
+ }
+
+ fn table<'life0, 'life1, 'async_trait>(
Review Comment:
We should remove these auto generated lifetimes to make it more readable.
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,73 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use dashmap::DashMap;
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+pub struct IcebergCatalogProvider {
+ schemas: DashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<String> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
Review Comment:
I'm not sure if datafusion allows nested namespace here, if not, maybe we should check namespace name.
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -0,0 +1,79 @@
+// 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.
+
+use std::sync::Arc;
+
+use datafusion::datasource::TableProvider;
+use iceberg::{table::Table, Catalog, NamespaceIdent, Result, TableIdent};
+
+pub(crate) struct IcebergTableProvider {
+ _inner: Table,
+}
+
+impl IcebergTableProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ name: impl Into<String>,
+ ) -> Result<Self> {
+ let name = name.into();
+ let ident = TableIdent::new(namespace, name);
+ let table = client.load_table(&ident).await?;
+
+ Ok(IcebergTableProvider { _inner: table })
+ }
+}
+
+impl TableProvider for IcebergTableProvider {
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef {
+ todo!()
Review Comment:
I think the convertion between iceberg schema and arrow schema has been implemented?
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -0,0 +1,79 @@
+// 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.
+
+use std::sync::Arc;
+
+use datafusion::datasource::TableProvider;
+use iceberg::{table::Table, Catalog, NamespaceIdent, Result, TableIdent};
+
+pub(crate) struct IcebergTableProvider {
+ _inner: Table,
+}
+
+impl IcebergTableProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ name: impl Into<String>,
+ ) -> Result<Self> {
+ let name = name.into();
+ let ident = TableIdent::new(namespace, name);
+ let table = client.load_table(&ident).await?;
+
+ Ok(IcebergTableProvider { _inner: table })
+ }
+}
+
+impl TableProvider for IcebergTableProvider {
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef {
+ todo!()
+ }
+
+ fn table_type(&self) -> datafusion::datasource::TableType {
+ todo!()
Review Comment:
I think `TableType::Base` is the answer.
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -0,0 +1,79 @@
+// 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.
+
+use std::sync::Arc;
+
+use datafusion::datasource::TableProvider;
+use iceberg::{table::Table, Catalog, NamespaceIdent, Result, TableIdent};
+
+pub(crate) struct IcebergTableProvider {
+ _inner: Table,
+}
+
+impl IcebergTableProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ name: impl Into<String>,
+ ) -> Result<Self> {
+ let name = name.into();
+ let ident = TableIdent::new(namespace, name);
+ let table = client.load_table(&ident).await?;
+
+ Ok(IcebergTableProvider { _inner: table })
+ }
+}
+
+impl TableProvider for IcebergTableProvider {
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef {
+ todo!()
+ }
+
+ fn table_type(&self) -> datafusion::datasource::TableType {
+ todo!()
+ }
+
+ fn scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(
+ &'life0 self,
+ _state: &'life1 datafusion::execution::context::SessionState,
+ _projection: Option<&'life2 Vec<usize>>,
+ _filters: &'life3 [datafusion::prelude::Expr],
+ _limit: Option<usize>,
+ ) -> core::pin::Pin<
+ Box<
+ dyn core::future::Future<
+ Output = datafusion::error::Result<
+ Arc<dyn datafusion::physical_plan::ExecutionPlan>,
+ >,
+ > + core::marker::Send
+ + 'async_trait,
+ >,
+ >
+ where
+ 'life0: 'async_trait,
+ 'life1: 'async_trait,
+ 'life2: 'async_trait,
+ 'life3: 'async_trait,
+ Self: 'async_trait,
+ {
+ todo!()
Review Comment:
I think this could be implemented with current scan api without filter pushdown?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1582042661
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,94 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: HashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ /// Asynchronously tries to construct a new [`IcebergCatalogProvider`]
+ /// using the given client to fetch and initialize schema providers for
+ /// each namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves the list of namespace names
+ /// attempts to create a schema provider for each namespace, and
+ /// collects these providers into a concurrent `HashMap`.
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<_> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
+ .collect();
+
+ let providers = try_join_all(
+ schema_names
+ .iter()
+ .map(|name| {
+ IcebergSchemaProvider::try_new(
+ client.clone(),
+ NamespaceIdent::new(name.clone()),
+ )
+ })
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let schemas: Vec<_> = schema_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn SchemaProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergCatalogProvider {
+ schemas: schemas.into_iter().collect(),
+ })
+ }
+}
+
+impl CatalogProvider for IcebergCatalogProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn schema_names(&self) -> Vec<String> {
+ self.schemas.keys().cloned().collect()
Review Comment:
Yes, when adding a cache, we need to a design for the trade of between performance and consistency. But I think we could leave it to actualy cache implentation? For this pr, we should not cache them in the providers, e.g. `CatalogProvider`, `SchemaProvider`, etc, but calling `Catalog` api directly.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1582158406
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,94 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: HashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ /// Asynchronously tries to construct a new [`IcebergCatalogProvider`]
+ /// using the given client to fetch and initialize schema providers for
+ /// each namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves the list of namespace names
+ /// attempts to create a schema provider for each namespace, and
+ /// collects these providers into a concurrent `HashMap`.
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<_> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
+ .collect();
+
+ let providers = try_join_all(
+ schema_names
+ .iter()
+ .map(|name| {
+ IcebergSchemaProvider::try_new(
+ client.clone(),
+ NamespaceIdent::new(name.clone()),
+ )
+ })
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let schemas: Vec<_> = schema_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn SchemaProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergCatalogProvider {
+ schemas: schemas.into_iter().collect(),
+ })
+ }
+}
+
+impl CatalogProvider for IcebergCatalogProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn schema_names(&self) -> Vec<String> {
+ self.schemas.keys().cloned().collect()
Review Comment:
> when we have runtime api landed
is there any issue or discussion already related to that? Or would you mind to quickly outline whats the plan here?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#issuecomment-2081488204
> Let's wait a moment to see what others think about the catalog snapshot problem. cc @Xuanwo @Fokko @viirya @sdd PTAL
If it turns out, that the PR is okay (for now) - let me update the missing docs; before we merge - should make the follow up work a lot easier in a couple of days/weeks.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1579728551
##########
crates/integrations/datafusion/README.md:
##########
@@ -0,0 +1,22 @@
+<!--
+ ~ 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 Integrations
+
+This crate contains the official Native Rust implementation of Apache Iceberg Integrations.
Review Comment:
Apache DataFusion and Apache Iceberg
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1580510837
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,94 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: HashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ /// Asynchronously tries to construct a new [`IcebergCatalogProvider`]
+ /// using the given client to fetch and initialize schema providers for
+ /// each namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves the list of namespace names
+ /// attempts to create a schema provider for each namespace, and
+ /// collects these providers into a concurrent `HashMap`.
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<_> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
+ .collect();
+
+ let providers = try_join_all(
+ schema_names
+ .iter()
+ .map(|name| {
+ IcebergSchemaProvider::try_new(
+ client.clone(),
+ NamespaceIdent::new(name.clone()),
+ )
+ })
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let schemas: Vec<_> = schema_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn SchemaProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergCatalogProvider {
+ schemas: schemas.into_iter().collect(),
+ })
+ }
+}
+
+impl CatalogProvider for IcebergCatalogProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn schema_names(&self) -> Vec<String> {
+ self.schemas.keys().cloned().collect()
Review Comment:
I'm thinking that this maybe incorrect since others processes may create new namespaces after creating the `try_new` method? We should the result of `list_namespaces` each time. For performance issue, we may create sth like [`CachingCatalog` in java](https://github.com/apache/iceberg/blob/c07f2aabc0a1d02f068ecf1514d2479c0fbdd3b0/core/src/main/java/org/apache/iceberg/CachingCatalog.java#L46) to implement `Catalog` trait, what do you think?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [WIP] Integration with Datafusion [iceberg-rust]
Posted by "tshauck (via GitHub)" <gi...@apache.org>.
tshauck commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1562767139
##########
Cargo.toml:
##########
@@ -21,6 +21,7 @@ members = [
"crates/catalog/*",
"crates/examples",
"crates/iceberg",
+ "crates/integrations",
Review Comment:
Updating the integrations to be datafusion specific looks good to me. For the actual datafusion dependency, you could move that into the Cargo.toml of the new datafusion subcrate itself. So e.g...
```toml
[dependencies]
dashmap = { workspace = true }
datafusion = "37.0.0"
futures = { workspace = true }
iceberg = { workspace = true }
log = { workspace = true }
tokio = { workspace = true }
```
I think it's generally considered better to scope subcrate specific dependencies to the crate that needs it. I'm sorry for muddying the waters via the feature suggestion, but then I'm not sure the additional complexity of it would be needed.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [WIP] Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1561386024
##########
crates/integrations/src/datafusion/schema.rs:
##########
@@ -0,0 +1,97 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use dashmap::DashMap;
+use datafusion::{
+ catalog::schema::SchemaProvider, datasource::TableProvider, error::DataFusionError,
+};
+use futures::FutureExt;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::datafusion::table::IcebergTableProvider;
+
+pub(crate) struct IcebergSchemaProvider {
+ tables: DashMap<String, Arc<dyn TableProvider>>,
+}
+
+impl IcebergSchemaProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: &NamespaceIdent,
+ ) -> Result<Self> {
+ let table_names: Vec<String> = client
+ .list_tables(namespace)
+ .await?
+ .iter()
+ .map(|t| t.name().to_owned())
+ .collect();
+
+ let mut tables = Vec::new();
+ for name in table_names {
+ let provider = IcebergTableProvider::try_new(client.clone(), namespace, &name).await?;
+ let provider = Arc::new(provider) as Arc<dyn TableProvider>;
+
+ tables.push((name, provider));
+ }
+
+ Ok(IcebergSchemaProvider {
+ tables: tables.into_iter().collect(),
+ })
+ }
+}
+
+impl SchemaProvider for IcebergSchemaProvider {
Review Comment:
yes - I just wanted to share a draft as early as possible - in order to get the design ideas aligned. And sure - any help is appreciated since I'm not so familiar with the Datafusion traits and I think we also need to implement `ExecutionPlan` in order to provide the correct return type for the `TableProvider`...?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [WIP] Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1562444674
##########
crates/integrations/src/datafusion/schema.rs:
##########
@@ -0,0 +1,97 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use dashmap::DashMap;
+use datafusion::{
+ catalog::schema::SchemaProvider, datasource::TableProvider, error::DataFusionError,
+};
+use futures::FutureExt;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::datafusion::table::IcebergTableProvider;
+
+pub(crate) struct IcebergSchemaProvider {
+ tables: DashMap<String, Arc<dyn TableProvider>>,
+}
+
+impl IcebergSchemaProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: &NamespaceIdent,
+ ) -> Result<Self> {
+ let table_names: Vec<String> = client
+ .list_tables(namespace)
+ .await?
+ .iter()
+ .map(|t| t.name().to_owned())
+ .collect();
+
+ let mut tables = Vec::new();
+ for name in table_names {
+ let provider = IcebergTableProvider::try_new(client.clone(), namespace, &name).await?;
+ let provider = Arc::new(provider) as Arc<dyn TableProvider>;
+
+ tables.push((name, provider));
+ }
+
+ Ok(IcebergSchemaProvider {
+ tables: tables.into_iter().collect(),
+ })
+ }
+}
+
+impl SchemaProvider for IcebergSchemaProvider {
Review Comment:
you can simply pick what you like - and let me know, how you'd want to contribute - if you want to commit to this branch, I guess I need to invite you to my fork?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1579715245
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -0,0 +1,79 @@
+// 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.
+
+use std::sync::Arc;
+
+use datafusion::datasource::TableProvider;
+use iceberg::{table::Table, Catalog, NamespaceIdent, Result, TableIdent};
+
+pub(crate) struct IcebergTableProvider {
+ _inner: Table,
+}
+
+impl IcebergTableProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ name: impl Into<String>,
+ ) -> Result<Self> {
+ let name = name.into();
+ let ident = TableIdent::new(namespace, name);
+ let table = client.load_table(&ident).await?;
+
+ Ok(IcebergTableProvider { _inner: table })
+ }
+}
+
+impl TableProvider for IcebergTableProvider {
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef {
+ todo!()
Review Comment:
yes; i believe we merged this 2 days ago; so i can take a look at this again now
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1579713512
##########
crates/integrations/datafusion/Cargo.toml:
##########
@@ -0,0 +1,46 @@
+# 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.
+
+[package]
+name = "iceberg-integrations-datafusion"
+version = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+rust-version = { workspace = true }
+
+categories = ["database"]
+description = "Apache Iceberg Integrations Datafusion"
+repository = { workspace = true }
+license = { workspace = true }
+keywords = ["iceberg", "integrations", "datfusion"]
+
+[features]
+default = []
+datafusion = ["dep:datafusion"]
Review Comment:
Here I'm probably even more confused; I tried this based on the review by @tshauck - but I have no experience with this and I'm not sure how to structure the crate, in order to allow a possible end-user to download only the integration dependency he needs e.g. iceberg-datafusion or iceberg-polars. etc.
So here I'm hoping for feedback and help.
EDIT:
I removed the feature flag for now. So the crate simply resides in /crates/integrations/datafusion
We can figure out later how to deal with features - or whats best practice here.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1579728551
##########
crates/integrations/datafusion/README.md:
##########
@@ -0,0 +1,22 @@
+<!--
+ ~ 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 Integrations
+
+This crate contains the official Native Rust implementation of Apache Iceberg Integrations.
Review Comment:
For above refined one, maybe Apache DataFusion and Apache Iceberg
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1579755826
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,73 @@
+// 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.
+
+use std::{any::Any, sync::Arc};
+
+use dashmap::DashMap;
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+pub struct IcebergCatalogProvider {
+ schemas: DashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<String> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
Review Comment:
In this case; we provide the catalog (e.g. hive metastore) and we don't support nested namespaces.
Even if datafusion allows nested namespaces; once we hit our catalog provider we'd only return an empty list.
So I don't think we need to perform any checks here?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1580024955
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -0,0 +1,79 @@
+// 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.
+
+use std::sync::Arc;
+
+use datafusion::datasource::TableProvider;
+use iceberg::{table::Table, Catalog, NamespaceIdent, Result, TableIdent};
+
+pub(crate) struct IcebergTableProvider {
+ _inner: Table,
+}
+
+impl IcebergTableProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ name: impl Into<String>,
+ ) -> Result<Self> {
+ let name = name.into();
+ let ident = TableIdent::new(namespace, name);
+ let table = client.load_table(&ident).await?;
+
+ Ok(IcebergTableProvider { _inner: table })
+ }
+}
+
+impl TableProvider for IcebergTableProvider {
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef {
+ todo!()
+ }
+
+ fn table_type(&self) -> datafusion::datasource::TableType {
+ todo!()
+ }
+
+ fn scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(
+ &'life0 self,
+ _state: &'life1 datafusion::execution::context::SessionState,
+ _projection: Option<&'life2 Vec<usize>>,
+ _filters: &'life3 [datafusion::prelude::Expr],
+ _limit: Option<usize>,
+ ) -> core::pin::Pin<
+ Box<
+ dyn core::future::Future<
+ Output = datafusion::error::Result<
+ Arc<dyn datafusion::physical_plan::ExecutionPlan>,
+ >,
+ > + core::marker::Send
+ + 'async_trait,
+ >,
+ >
+ where
+ 'life0: 'async_trait,
+ 'life1: 'async_trait,
+ 'life2: 'async_trait,
+ 'life3: 'async_trait,
+ Self: 'async_trait,
+ {
+ todo!()
Review Comment:
...I think this has to be done (PTAL @liurenjie1024):
- impl asyn fn scan() on IcebergTableProvider which has to return an `Arc<dyn Execution Plan>`
- therefore we need to: impl the trait ExecutionPlan for IcebergTableScan which is basically a wrapper around an actual PhysicalPlan e.g. ParquetExec
- therefore we need to: create an PhysicalPlan when building our IcebergTableScan
I think creating the PhysicalPlan and providing all the input like FileScanConfig will be the majority of the work and I'm not sure we can provide all of this yet...but well see...
reference from delta-rs:
- [DeltaScan](https://github.com/delta-io/delta-rs/blob/main/crates/core/src/delta_datafusion/mod.rs#L790-L798)
- [ExecutionPlan for DeltaScan](https://github.com/delta-io/delta-rs/blob/main/crates/core/src/delta_datafusion/mod.rs#L814-L881)
- [building DeltaScan](https://github.com/delta-io/delta-rs/blob/main/crates/core/src/delta_datafusion/mod.rs#L507-L651) and the actual PhysicalPlan // [ParquetExec](https://docs.rs/datafusion/latest/datafusion/datasource/physical_plan/parquet/struct.ParquetExec.html)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [WIP] Integration with Datafusion [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#issuecomment-2068383273
Thanks! Sorry for replying late. I think this is a good start for the integration work. And I have completed https://github.com/apache/iceberg-rust/pull/277. Maybe we can convert this PR to ready for review now. @marvinlanhenke
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1584261856
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,97 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: HashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ /// Asynchronously tries to construct a new [`IcebergCatalogProvider`]
+ /// using the given client to fetch and initialize schema providers for
+ /// each namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves the list of namespace names
+ /// attempts to create a schema provider for each namespace, and
+ /// collects these providers into a `HashMap`.
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ // TODO:
+ // Schemas and providers should be cached and evicted based on time
+ // As of right now; schemas might become stale.
Review Comment:
> _There are 2 hard problems in computer science: cache invalidation, naming things, and off-by-1 errors._ - Leon Bambrick
Is there no way to leave this up to the user?
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -0,0 +1,136 @@
+// 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.
+
+use std::{any::Any, pin::Pin, sync::Arc};
+
+use datafusion::{
+ arrow::{array::RecordBatch, datatypes::SchemaRef as ArrowSchemaRef},
+ execution::{SendableRecordBatchStream, TaskContext},
+ physical_expr::EquivalenceProperties,
+ physical_plan::{
+ stream::RecordBatchStreamAdapter, DisplayAs, ExecutionMode, ExecutionPlan, Partitioning,
+ PlanProperties,
+ },
+};
+use futures::{Stream, TryStreamExt};
+use iceberg::table::Table;
+
+use crate::to_datafusion_error;
+
+#[derive(Debug)]
Review Comment:
Can't we enforce this kind of thing in a linter? :)
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,94 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: HashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ /// Asynchronously tries to construct a new [`IcebergCatalogProvider`]
+ /// using the given client to fetch and initialize schema providers for
+ /// each namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves the list of namespace names
+ /// attempts to create a schema provider for each namespace, and
+ /// collects these providers into a concurrent `HashMap`.
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<_> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
+ .collect();
+
+ let providers = try_join_all(
+ schema_names
+ .iter()
+ .map(|name| {
+ IcebergSchemaProvider::try_new(
+ client.clone(),
+ NamespaceIdent::new(name.clone()),
+ )
+ })
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let schemas: Vec<_> = schema_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn SchemaProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergCatalogProvider {
+ schemas: schemas.into_iter().collect(),
+ })
+ }
+}
+
+impl CatalogProvider for IcebergCatalogProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn schema_names(&self) -> Vec<String> {
+ self.schemas.keys().cloned().collect()
Review Comment:
We should be hesitant with caching, especially avoiding upfront optimizations. For Iceberg in general, consistency is king.
> For this pr, we should not cache them in the providers, e.g. CatalogProvider, SchemaProvider, etc, but calling Catalog api directly.
This makes sense, but I see the issue with blocking on async calls. At first, I would take the price of waiting for the blocking calls. Even it is still a remote call, the ones to the REST catalog should be lightning-fast (that's where the caching happens).
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "simonvandel (via GitHub)" <gi...@apache.org>.
simonvandel commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1581219550
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -0,0 +1,79 @@
+// 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.
+
+use std::sync::Arc;
+
+use datafusion::datasource::TableProvider;
+use iceberg::{table::Table, Catalog, NamespaceIdent, Result, TableIdent};
+
+pub(crate) struct IcebergTableProvider {
+ _inner: Table,
+}
+
+impl IcebergTableProvider {
+ pub(crate) async fn try_new(
+ client: Arc<dyn Catalog>,
+ namespace: NamespaceIdent,
+ name: impl Into<String>,
+ ) -> Result<Self> {
+ let name = name.into();
+ let ident = TableIdent::new(namespace, name);
+ let table = client.load_table(&ident).await?;
+
+ Ok(IcebergTableProvider { _inner: table })
+ }
+}
+
+impl TableProvider for IcebergTableProvider {
+ fn as_any(&self) -> &dyn std::any::Any {
+ self
+ }
+
+ fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef {
+ todo!()
+ }
+
+ fn table_type(&self) -> datafusion::datasource::TableType {
+ todo!()
+ }
+
+ fn scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(
+ &'life0 self,
+ _state: &'life1 datafusion::execution::context::SessionState,
+ _projection: Option<&'life2 Vec<usize>>,
+ _filters: &'life3 [datafusion::prelude::Expr],
+ _limit: Option<usize>,
+ ) -> core::pin::Pin<
+ Box<
+ dyn core::future::Future<
+ Output = datafusion::error::Result<
+ Arc<dyn datafusion::physical_plan::ExecutionPlan>,
+ >,
+ > + core::marker::Send
+ + 'async_trait,
+ >,
+ >
+ where
+ 'life0: 'async_trait,
+ 'life1: 'async_trait,
+ 'life2: 'async_trait,
+ 'life3: 'async_trait,
+ Self: 'async_trait,
+ {
+ todo!()
Review Comment:
I think you can do what is documented here: https://docs.rs/datafusion/latest/datafusion/physical_plan/trait.ExecutionPlan.html#lazily-async-create-a-stream
So inside `IcebergTableProvider::scan` you create `IcebergScanExec` where `IcebergScanExec` implements `ExecutionPlan`.
The `ExecutionPlan::execute` impl of `IcebergScanExec` would be something like this
```rust
impl ExecutionPlan for IcebergScanExec {
fn execute(
&self,
partition: usize,
context: Arc<TaskContext>
) -> Result<SendableRecordBatchStream> {
let fut = create_batch_stream(self.table_scan);
let stream = futures::stream::once(fut).try_flatten();
Ok(Box::pin(RecordBatchStreamAdapter::new(self.schema.clone(), stream)))
}
}
```
/// async function that returns a stream
async fn create_batch_stream(table: TableScan) -> Result<SendableRecordBatchStream> {
todo!()
}
(Not @liurenjie1024 but hopefully it is helpful - I didn't compile it :) )
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#issuecomment-2081431552
@liurenjie1024 @ZENOTME @viirya @simonvandel
...I just went ahead and pushed the recent updates; PTAL
## unresolved / todo:
- [ ] proper integration tests for table scan (requires us to setup a table with actual snapshots / data )
- [ ] a cache for `IcebergCatalogProvider` and `IcebergSchemaProvider` / so data does not become stale
- [ ] improve impl ExecutionPlan for IcebergTableScan (once we support filter pushdown, etc.)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "marvinlanhenke (via GitHub)" <gi...@apache.org>.
marvinlanhenke commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1582080221
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,94 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: HashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ /// Asynchronously tries to construct a new [`IcebergCatalogProvider`]
+ /// using the given client to fetch and initialize schema providers for
+ /// each namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves the list of namespace names
+ /// attempts to create a schema provider for each namespace, and
+ /// collects these providers into a concurrent `HashMap`.
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<_> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
+ .collect();
+
+ let providers = try_join_all(
+ schema_names
+ .iter()
+ .map(|name| {
+ IcebergSchemaProvider::try_new(
+ client.clone(),
+ NamespaceIdent::new(name.clone()),
+ )
+ })
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let schemas: Vec<_> = schema_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn SchemaProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergCatalogProvider {
+ schemas: schemas.into_iter().collect(),
+ })
+ }
+}
+
+impl CatalogProvider for IcebergCatalogProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn schema_names(&self) -> Vec<String> {
+ self.schemas.keys().cloned().collect()
Review Comment:
@liurenjie1024
...I took another look; and I'm not sure we can call the Catalog API directly each time the funtion on the e.g. `trait CatalogProvider` is invoked. The trait is required to be synchronous; however our catalog api calls are async.
So perhaps; we leave it as is for this PR and work on the cache instead? However, I'll guess we're facing the same problem here as well? Perhaps, I'm just missing something?
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,94 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: HashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ /// Asynchronously tries to construct a new [`IcebergCatalogProvider`]
+ /// using the given client to fetch and initialize schema providers for
+ /// each namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves the list of namespace names
+ /// attempts to create a schema provider for each namespace, and
+ /// collects these providers into a concurrent `HashMap`.
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<_> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
+ .collect();
+
+ let providers = try_join_all(
+ schema_names
+ .iter()
+ .map(|name| {
+ IcebergSchemaProvider::try_new(
+ client.clone(),
+ NamespaceIdent::new(name.clone()),
+ )
+ })
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let schemas: Vec<_> = schema_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn SchemaProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergCatalogProvider {
+ schemas: schemas.into_iter().collect(),
+ })
+ }
+}
+
+impl CatalogProvider for IcebergCatalogProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn schema_names(&self) -> Vec<String> {
+ self.schemas.keys().cloned().collect()
Review Comment:
@liurenjie1024
...I took another look; and I'm not sure we can call the Catalog API directly each time the funtion on the e.g. `trait CatalogProvider` is invoked. The trait is required to be synchronous; however our catalog api calls are async.
So perhaps; we leave it as is for this PR and work on the cache instead? However, I'll guess we're facing the same problem here as well? But maybe, I'm just missing something?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Basic Integration with Datafusion [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #324:
URL: https://github.com/apache/iceberg-rust/pull/324#discussion_r1582151715
##########
crates/integrations/datafusion/src/catalog.rs:
##########
@@ -0,0 +1,94 @@
+// 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.
+
+use std::{any::Any, collections::HashMap, sync::Arc};
+
+use datafusion::catalog::{schema::SchemaProvider, CatalogProvider};
+use futures::future::try_join_all;
+use iceberg::{Catalog, NamespaceIdent, Result};
+
+use crate::schema::IcebergSchemaProvider;
+
+/// Provides an interface to manage and access multiple schemas
+/// within an Iceberg [`Catalog`].
+///
+/// Acts as a centralized catalog provider that aggregates
+/// multiple [`SchemaProvider`], each associated with distinct namespaces.
+pub struct IcebergCatalogProvider {
+ /// A concurrent `HashMap` where keys are namespace names
+ /// and values are dynamic references to objects implementing the
+ /// [`SchemaProvider`] trait.
+ schemas: HashMap<String, Arc<dyn SchemaProvider>>,
+}
+
+impl IcebergCatalogProvider {
+ /// Asynchronously tries to construct a new [`IcebergCatalogProvider`]
+ /// using the given client to fetch and initialize schema providers for
+ /// each namespace in the Iceberg [`Catalog`].
+ ///
+ /// This method retrieves the list of namespace names
+ /// attempts to create a schema provider for each namespace, and
+ /// collects these providers into a concurrent `HashMap`.
+ pub async fn try_new(client: Arc<dyn Catalog>) -> Result<Self> {
+ let schema_names: Vec<_> = client
+ .list_namespaces(None)
+ .await?
+ .iter()
+ .flat_map(|ns| ns.as_ref().clone())
+ .collect();
+
+ let providers = try_join_all(
+ schema_names
+ .iter()
+ .map(|name| {
+ IcebergSchemaProvider::try_new(
+ client.clone(),
+ NamespaceIdent::new(name.clone()),
+ )
+ })
+ .collect::<Vec<_>>(),
+ )
+ .await?;
+
+ let schemas: Vec<_> = schema_names
+ .into_iter()
+ .zip(providers.into_iter())
+ .map(|(name, provider)| {
+ let provider = Arc::new(provider) as Arc<dyn SchemaProvider>;
+ (name, provider)
+ })
+ .collect();
+
+ Ok(IcebergCatalogProvider {
+ schemas: schemas.into_iter().collect(),
+ })
+ }
+}
+
+impl CatalogProvider for IcebergCatalogProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn schema_names(&self) -> Vec<String> {
+ self.schemas.keys().cloned().collect()
Review Comment:
Yes, you are right. As of now, we can only provide a snapshot version. It's possible to block on an async call, but it requires runtime api. cc @Xuanwo @Fokko @viirya What dou you think? My suggestion is to leave it as it now, and fix it when we have runtime api landed.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org