You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by ma...@apache.org on 2026/10/10 04:55:08 UTC
(datafusion-iceberg) branch main updated (961bcee -> 7252680)
This is an automated email from the ASF dual-hosted git repository.
manuzhang pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-iceberg.git
from 961bcee chore(deps): bump rustls from 0.23.44 to 0.23.45 (#53)
new 79d68a5 Refactor folder structure
new 7252680 Refactor Iceberg reads to DataSource
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
AGENTS.md | 5 +-
Cargo.lock | 12 +-
.../src/lib.rs => datafusion/src/catalog/mod.rs} | 7 +-
.../src/{catalog.rs => catalog/provider.rs} | 2 +-
crates/datafusion/src/{ => catalog}/schema.rs | 12 +-
crates/datafusion/src/lib.rs | 14 +-
.../src/lib.rs => datafusion/src/metadata/mod.rs} | 8 +-
crates/datafusion/src/metadata/scan.rs | 104 +++
.../{table/metadata_table.rs => metadata/table.rs} | 7 +-
.../datafusion/src/physical_plan/metadata_scan.rs | 107 ---
crates/datafusion/src/physical_plan/scan.rs | 334 ----------
.../src/lib.rs => datafusion/src/read/mod.rs} | 8 +-
.../expr_to_predicate.rs => read/predicate.rs} | 2 +-
crates/datafusion/src/read/scan.rs | 257 ++++++++
.../catalog_table_provider.rs} | 725 ++++++---------------
.../factory.rs} | 0
.../src/{lib.rs => table_provider/mod.rs} | 21 +-
.../src/table_provider/static_table_provider.rs | 352 ++++++++++
.../src/{physical_plan => write}/commit.rs | 7 +-
.../src/{physical_plan/write.rs => write/exec.rs} | 5 +-
.../datafusion/src/{physical_plan => write}/mod.rs | 13 +-
.../src/{physical_plan => write}/project.rs | 2 +-
.../src/{physical_plan => write}/repartition.rs | 6 +-
.../src/{physical_plan => write}/sort.rs | 0
crates/datafusion/src/{ => write}/task_writer.rs | 0
.../tests/integration_datafusion_test.rs | 85 ++-
.../testdata/slts/df_test/basic_queries.slt | 7 +-
.../slts/df_test/binary_predicate_pushdown.slt | 6 +-
.../slts/df_test/boolean_predicate_pushdown.slt | 10 +-
.../slts/df_test/like_predicate_pushdown.slt | 6 +-
.../slts/df_test/timestamp_predicate_pushdown.slt | 8 +-
31 files changed, 1060 insertions(+), 1072 deletions(-)
copy crates/{playground/src/lib.rs => datafusion/src/catalog/mod.rs} (84%)
rename crates/datafusion/src/{catalog.rs => catalog/provider.rs} (98%)
rename crates/datafusion/src/{ => catalog}/schema.rs (97%)
copy crates/{playground/src/lib.rs => datafusion/src/metadata/mod.rs} (84%)
create mode 100644 crates/datafusion/src/metadata/scan.rs
rename crates/datafusion/src/{table/metadata_table.rs => metadata/table.rs} (94%)
delete mode 100644 crates/datafusion/src/physical_plan/metadata_scan.rs
delete mode 100644 crates/datafusion/src/physical_plan/scan.rs
copy crates/{playground/src/lib.rs => datafusion/src/read/mod.rs} (84%)
rename crates/datafusion/src/{physical_plan/expr_to_predicate.rs => read/predicate.rs} (99%)
create mode 100644 crates/datafusion/src/read/scan.rs
rename crates/datafusion/src/{table/mod.rs => table_provider/catalog_table_provider.rs} (62%)
rename crates/datafusion/src/{table/table_provider_factory.rs => table_provider/factory.rs} (100%)
copy crates/datafusion/src/{lib.rs => table_provider/mod.rs} (65%)
create mode 100644 crates/datafusion/src/table_provider/static_table_provider.rs
rename crates/datafusion/src/{physical_plan => write}/commit.rs (99%)
rename crates/datafusion/src/{physical_plan/write.rs => write/exec.rs} (99%)
rename crates/datafusion/src/{physical_plan => write}/mod.rs (74%)
rename crates/datafusion/src/{physical_plan => write}/project.rs (99%)
rename crates/datafusion/src/{physical_plan => write}/repartition.rs (99%)
rename crates/datafusion/src/{physical_plan => write}/sort.rs (100%)
rename crates/datafusion/src/{ => write}/task_writer.rs (100%)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-iceberg) 02/02: Refactor Iceberg reads to DataSource
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
manuzhang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-iceberg.git
commit 72526809d80787c375ebc765ecee9b1db418a232
Author: Gabriel <ga...@datadoghq.com>
AuthorDate: Tue Oct 6 08:39:27 2026 +0200
Refactor Iceberg reads to DataSource
---
crates/datafusion/src/metadata/mod.rs | 2 +-
crates/datafusion/src/metadata/scan.rs | 99 +++++----
crates/datafusion/src/metadata/table.rs | 7 +-
crates/datafusion/src/read/mod.rs | 2 +-
crates/datafusion/src/read/scan.rs | 223 +++++++--------------
.../src/table_provider/catalog_table_provider.rs | 18 +-
.../src/table_provider/static_table_provider.rs | 27 ++-
.../tests/integration_datafusion_test.rs | 73 +++++--
.../testdata/slts/df_test/basic_queries.slt | 7 +-
.../slts/df_test/binary_predicate_pushdown.slt | 6 +-
.../slts/df_test/boolean_predicate_pushdown.slt | 10 +-
.../slts/df_test/like_predicate_pushdown.slt | 6 +-
.../slts/df_test/timestamp_predicate_pushdown.slt | 8 +-
13 files changed, 226 insertions(+), 262 deletions(-)
diff --git a/crates/datafusion/src/metadata/mod.rs b/crates/datafusion/src/metadata/mod.rs
index 33893fe..7c4fcaa 100644
--- a/crates/datafusion/src/metadata/mod.rs
+++ b/crates/datafusion/src/metadata/mod.rs
@@ -18,5 +18,5 @@
pub(crate) mod scan;
pub(crate) mod table;
-pub use scan::IcebergMetadataScan;
+pub use scan::IcebergMetadataDataSource;
pub use table::IcebergMetadataTableProvider;
diff --git a/crates/datafusion/src/metadata/scan.rs b/crates/datafusion/src/metadata/scan.rs
index be8d36d..3d20cb8 100644
--- a/crates/datafusion/src/metadata/scan.rs
+++ b/crates/datafusion/src/metadata/scan.rs
@@ -17,91 +17,88 @@
use std::sync::Arc;
-use datafusion::catalog::TableProvider;
+use datafusion::common::Statistics;
use datafusion::common::tree_node::TreeNodeRecursion;
+use datafusion::datasource::TableProvider;
+use datafusion::datasource::source::DataSource;
+use datafusion::error::Result;
+use datafusion::execution::{SendableRecordBatchStream, TaskContext};
+use datafusion::physical_expr::projection::ProjectionExprs;
use datafusion::physical_expr::{EquivalenceProperties, PhysicalExpr};
-use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
-use datafusion::physical_plan::{DisplayAs, ExecutionPlan, Partitioning, PlanProperties};
+use datafusion::physical_plan::{DisplayFormatType, Partitioning};
use futures::TryStreamExt;
use super::table::IcebergMetadataTableProvider;
-/// Scans an Iceberg metadata table, such as `$snapshots`.
-#[derive(Debug)]
-pub struct IcebergMetadataScan {
+/// Reads rows from an Iceberg metadata table through DataFusion's `DataSource` API.
+#[derive(Debug, Clone)]
+pub struct IcebergMetadataDataSource {
provider: IcebergMetadataTableProvider,
- properties: Arc<PlanProperties>,
}
-impl IcebergMetadataScan {
- /// Creates a scan of the metadata table `provider` reads.
+impl IcebergMetadataDataSource {
pub fn new(provider: IcebergMetadataTableProvider) -> Self {
- let properties = Arc::new(PlanProperties::new(
- EquivalenceProperties::new(provider.schema()),
- Partitioning::UnknownPartitioning(1),
- EmissionType::Incremental,
- Boundedness::Bounded,
- ));
- Self {
- provider,
- properties,
- }
+ Self { provider }
}
- /// The provider this node scans.
+ /// The provider this source scans.
pub fn provider(&self) -> &IcebergMetadataTableProvider {
&self.provider
}
}
-impl DisplayAs for IcebergMetadataScan {
+impl DataSource for IcebergMetadataDataSource {
+ fn open(
+ &self,
+ _partition: usize,
+ _context: Arc<TaskContext>,
+ ) -> Result<SendableRecordBatchStream> {
+ let fut = self.provider.clone().scan();
+ let stream = futures::stream::once(fut).try_flatten();
+ let schema = self.provider.schema();
+ Ok(Box::pin(RecordBatchStreamAdapter::new(schema, stream)))
+ }
+
fn fmt_as(
&self,
- _t: datafusion::physical_plan::DisplayFormatType,
+ _t: DisplayFormatType,
f: &mut std::fmt::Formatter,
) -> std::fmt::Result {
- write!(f, "IcebergMetadataScan")
+ write!(f, "format=iceberg_metadata")
}
-}
-impl ExecutionPlan for IcebergMetadataScan {
- fn name(&self) -> &str {
- "IcebergMetadataScan"
+ fn output_partitioning(&self) -> Partitioning {
+ Partitioning::UnknownPartitioning(1)
}
- fn properties(&self) -> &Arc<PlanProperties> {
- &self.properties
+ fn eq_properties(&self) -> EquivalenceProperties {
+ EquivalenceProperties::new(self.provider.schema())
}
- fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
- vec![]
+ fn partition_statistics(&self, _partition: Option<usize>) -> Result<Arc<Statistics>> {
+ Ok(Arc::new(Statistics::new_unknown(&self.provider.schema())))
}
- fn apply_expressions(
- &self,
- _f: &mut dyn FnMut(
- &Arc<dyn PhysicalExpr>,
- ) -> datafusion::error::Result<TreeNodeRecursion>,
- ) -> datafusion::error::Result<TreeNodeRecursion> {
- Ok(TreeNodeRecursion::Continue)
+ fn with_fetch(&self, _fetch: Option<usize>) -> Option<Arc<dyn DataSource>> {
+ None
}
- fn with_new_children(
- self: Arc<Self>,
- _children: Vec<Arc<dyn ExecutionPlan>>,
- ) -> datafusion::error::Result<Arc<dyn ExecutionPlan>> {
- Ok(self)
+ fn fetch(&self) -> Option<usize> {
+ None
}
- fn execute(
+ fn try_swapping_with_projection(
&self,
- _partition: usize,
- _context: Arc<datafusion::execution::TaskContext>,
- ) -> datafusion::error::Result<datafusion::execution::SendableRecordBatchStream> {
- let fut = self.provider.clone().scan();
- let stream = futures::stream::once(fut).try_flatten();
- let schema = self.provider.schema();
- Ok(Box::pin(RecordBatchStreamAdapter::new(schema, stream)))
+ _projection: &ProjectionExprs,
+ ) -> Result<Option<Arc<dyn DataSource>>> {
+ Ok(None)
+ }
+
+ fn apply_expressions(
+ &self,
+ _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
+ ) -> Result<TreeNodeRecursion> {
+ Ok(TreeNodeRecursion::Continue)
}
}
diff --git a/crates/datafusion/src/metadata/table.rs b/crates/datafusion/src/metadata/table.rs
index f3c91f3..6bd9448 100644
--- a/crates/datafusion/src/metadata/table.rs
+++ b/crates/datafusion/src/metadata/table.rs
@@ -21,6 +21,7 @@ use async_trait::async_trait;
use datafusion::arrow::array::RecordBatch;
use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
use datafusion::catalog::Session;
+use datafusion::datasource::source::DataSourceExec;
use datafusion::datasource::{TableProvider, TableType};
use datafusion::error::Result;
use datafusion::logical_expr::Expr;
@@ -31,7 +32,7 @@ use iceberg::arrow::schema_to_arrow_schema;
use iceberg::inspect::MetadataTableType;
use iceberg::table::Table;
-use super::scan::IcebergMetadataScan;
+use super::scan::IcebergMetadataDataSource;
use crate::to_datafusion_error;
/// Represents a [`TableProvider`] for the Iceberg [`iceberg::Catalog`],
@@ -65,7 +66,9 @@ impl TableProvider for IcebergMetadataTableProvider {
_filters: &[Expr],
_limit: Option<usize>,
) -> Result<Arc<dyn ExecutionPlan>> {
- Ok(Arc::new(IcebergMetadataScan::new(self.clone())))
+ Ok(DataSourceExec::from_data_source(
+ IcebergMetadataDataSource::new(self.clone()),
+ ))
}
}
diff --git a/crates/datafusion/src/read/mod.rs b/crates/datafusion/src/read/mod.rs
index 7933c39..26e2fb4 100644
--- a/crates/datafusion/src/read/mod.rs
+++ b/crates/datafusion/src/read/mod.rs
@@ -19,4 +19,4 @@ pub(crate) mod predicate;
pub(crate) mod scan;
pub use predicate::convert_filters_to_predicate;
-pub use scan::IcebergTableScan;
+pub use scan::IcebergDataSource;
diff --git a/crates/datafusion/src/read/scan.rs b/crates/datafusion/src/read/scan.rs
index 71a1fb2..4432f0b 100644
--- a/crates/datafusion/src/read/scan.rs
+++ b/crates/datafusion/src/read/scan.rs
@@ -17,17 +17,18 @@
use std::pin::Pin;
use std::sync::Arc;
-use std::vec;
use datafusion::arrow::array::RecordBatch;
use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
+use datafusion::common::Statistics;
use datafusion::common::tree_node::TreeNodeRecursion;
+use datafusion::datasource::source::DataSource;
use datafusion::error::Result;
use datafusion::execution::{SendableRecordBatchStream, TaskContext};
+use datafusion::physical_expr::projection::ProjectionExprs;
use datafusion::physical_expr::{EquivalenceProperties, PhysicalExpr};
-use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
-use datafusion::physical_plan::{DisplayAs, ExecutionPlan, Partitioning, PlanProperties};
+use datafusion::physical_plan::{DisplayFormatType, Partitioning};
use datafusion::prelude::Expr;
use futures::{Stream, TryStreamExt};
use iceberg::expr::Predicate;
@@ -38,25 +39,24 @@ use crate::to_datafusion_error;
/// Manages the scanning process of an Iceberg [`Table`], encapsulating the
/// necessary details and computed properties required for execution planning.
-#[derive(Debug)]
-pub struct IcebergTableScan {
+#[derive(Debug, Clone)]
+pub struct IcebergDataSource {
/// A table in the catalog.
table: Table,
/// Snapshot of the table to scan.
snapshot_id: Option<i64>,
- /// Stores certain, often expensive to compute,
- /// plan properties used in query optimization.
- plan_properties: Arc<PlanProperties>,
- /// The columns to read, by name: the fields of the output schema
+ /// Schema returned by this data source after projection.
+ schema: ArrowSchemaRef,
+ /// Projection column names in the output schema.
projection: Vec<String>,
/// Filters to apply to the table scan
predicates: Option<Predicate>,
/// Optional limit on the number of rows to return
- limit: Option<usize>,
+ fetch: Option<usize>,
}
-impl IcebergTableScan {
- /// Creates a new [`IcebergTableScan`] object.
+impl IcebergDataSource {
+ /// Creates a new [`IcebergDataSource`] object.
pub(crate) fn new(
table: Table,
snapshot_id: Option<i64>,
@@ -78,104 +78,30 @@ impl IcebergTableScan {
))
}
- /// Creates a scan of `table` from an already-converted Iceberg
- /// [`Predicate`] rather than DataFusion filters, for rebuilding a scan from
- /// its parts, such as after sending them to another process. A predicate
- /// cannot be converted back to the filters it came from.
+ /// Rebuilds a source from the parts needed to scan an Iceberg table.
///
- /// Each argument takes what the matching accessor returns (`schema` what
- /// [`ExecutionPlan::schema`] does, and `predicates` what
- /// [`Self::predicates`] does):
- ///
- /// - `snapshot_id`: the snapshot to read, or `None` for the table's current
- /// snapshot.
- /// - `schema`: the Arrow schema the scan outputs. The scan reads the
- /// columns of the same names from the snapshot it scans, and no others.
- /// A name that snapshot's schema lacks fails the scan when it runs.
- /// - `predicates`: pushed down to Iceberg to skip data files and rows. The
- /// table providers report their filters as
- /// [`Inexact`](datafusion::logical_expr::TableProviderFilterPushDown::Inexact),
- /// so DataFusion still applies them above the scan.
- /// - `limit`: the most rows the scan returns, or `None` for all of them.
- ///
- /// # Example
- ///
- /// ```
- /// use std::collections::HashMap;
- ///
- /// use datafusion::catalog::TableProvider;
- /// use datafusion::physical_plan::ExecutionPlan;
- /// use datafusion::prelude::{SessionContext, col, lit};
- /// use datafusion_iceberg::IcebergStaticTableProvider;
- /// use datafusion_iceberg::read::IcebergTableScan;
- /// use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder};
- /// use iceberg::spec::{NestedField, PrimitiveType, Schema, Type};
- /// use iceberg::{Catalog, CatalogBuilder, NamespaceIdent, TableCreation};
- ///
- /// # tokio::runtime::Runtime::new()?.block_on(async {
- /// # let warehouse = tempfile::tempdir()?;
- /// # let props = HashMap::from([(
- /// # MEMORY_CATALOG_WAREHOUSE.to_string(),
- /// # warehouse.path().display().to_string(),
- /// # )]);
- /// # let catalog = MemoryCatalogBuilder::default().load("memory", props).await?;
- /// # let namespace = NamespaceIdent::new("ns".to_string());
- /// # catalog.create_namespace(&namespace, HashMap::new()).await?;
- /// # let schema = Schema::builder()
- /// # .with_fields(vec![
- /// # NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(),
- /// # NestedField::optional(2, "name", Type::Primitive(PrimitiveType::String))
- /// # .into(),
- /// # ])
- /// # .build()?;
- /// # let creation = TableCreation::builder().name("t".to_string()).schema(schema).build();
- /// # let table = catalog.create_table(&namespace, creation).await?;
- /// let provider = IcebergStaticTableProvider::try_new_from_table(table).await?;
- /// let ctx = SessionContext::new();
- /// let filters = [col("id").gt(lit(1))];
- /// let plan = provider
- /// .scan(&ctx.state(), Some(&vec![1]), &filters, None)
- /// .await?;
- /// let scan = plan.downcast_ref::<IcebergTableScan>().unwrap();
- ///
- /// // Rebuild an equivalent scan from the original's accessors alone.
- /// let rebuilt = IcebergTableScan::new_with_predicate(
- /// scan.table().clone(),
- /// scan.snapshot_id(),
- /// scan.schema(),
- /// scan.predicates().cloned(),
- /// scan.limit(),
- /// );
- /// assert_eq!(rebuilt.schema(), scan.schema());
- /// assert_eq!(rebuilt.projection(), scan.projection());
- /// assert_eq!(rebuilt.predicates(), scan.predicates());
- /// # Ok::<(), Box<dyn std::error::Error>>(())
- /// # })?;
- /// # Ok::<(), Box<dyn std::error::Error>>(())
- /// ```
+ /// The supplied schema determines both the source's output schema and the
+ /// columns read from Iceberg. Callers can recreate a source from its
+ /// accessors after transporting those values to another process.
pub fn new_with_predicate(
table: Table,
snapshot_id: Option<i64>,
schema: ArrowSchemaRef,
predicates: Option<Predicate>,
- limit: Option<usize>,
+ fetch: Option<usize>,
) -> Self {
- // Reading the columns by name, rather than all of them, keeps the
- // batches matching `schema` even when the table has columns it lacks.
let projection = schema
.fields()
.iter()
.map(|field| field.name().clone())
.collect();
- let plan_properties = Self::compute_properties(schema);
-
Self {
table,
snapshot_id,
- plan_properties,
+ schema,
projection,
predicates,
- limit,
+ fetch,
}
}
@@ -187,8 +113,10 @@ impl IcebergTableScan {
self.snapshot_id
}
- /// The names of the columns the scan reads, which are the fields of its
- /// schema.
+ pub fn schema(&self) -> ArrowSchemaRef {
+ self.schema.clone()
+ }
+
pub fn projection(&self) -> &[String] {
&self.projection
}
@@ -197,33 +125,12 @@ impl IcebergTableScan {
self.predicates.as_ref()
}
- pub fn limit(&self) -> Option<usize> {
- self.limit
- }
-
- /// Computes [`PlanProperties`] used in query optimization.
- fn compute_properties(schema: ArrowSchemaRef) -> Arc<PlanProperties> {
- // TODO:
- // This is more or less a placeholder, to be replaced
- // once we support output-partitioning
- Arc::new(PlanProperties::new(
- EquivalenceProperties::new(schema),
- Partitioning::UnknownPartitioning(1),
- EmissionType::Incremental,
- Boundedness::Bounded,
- ))
+ pub fn fetch(&self) -> Option<usize> {
+ self.fetch
}
}
-impl ExecutionPlan for IcebergTableScan {
- fn name(&self) -> &str {
- "IcebergTableScan"
- }
-
- fn children(&self) -> Vec<&Arc<dyn ExecutionPlan + 'static>> {
- vec![]
- }
-
+impl DataSource for IcebergDataSource {
fn apply_expressions(
&self,
_f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
@@ -231,18 +138,7 @@ impl ExecutionPlan for IcebergTableScan {
Ok(TreeNodeRecursion::Continue)
}
- fn with_new_children(
- self: Arc<Self>,
- _children: Vec<Arc<dyn ExecutionPlan>>,
- ) -> Result<Arc<dyn ExecutionPlan>> {
- Ok(self)
- }
-
- fn properties(&self) -> &Arc<PlanProperties> {
- &self.plan_properties
- }
-
- fn execute(
+ fn open(
&self,
_partition: usize,
_context: Arc<TaskContext>,
@@ -250,14 +146,14 @@ impl ExecutionPlan for IcebergTableScan {
let fut = get_batch_stream(
self.table.clone(),
self.snapshot_id,
- self.projection.clone(),
+ Some(self.projection.clone()),
self.predicates.clone(),
);
let stream = futures::stream::once(fut).try_flatten();
- // Apply limit if specified
+ // Apply the fetch if specified.
let limited_stream: Pin<Box<dyn Stream<Item = Result<RecordBatch>> + Send>> =
- if let Some(limit) = self.limit {
+ if let Some(limit) = self.fetch {
let mut remaining = limit;
Box::pin(stream.try_filter_map(move |batch| {
futures::future::ready(if remaining == 0 {
@@ -276,31 +172,55 @@ impl ExecutionPlan for IcebergTableScan {
};
Ok(Box::pin(RecordBatchStreamAdapter::new(
- self.schema(),
+ self.schema.clone(),
limited_stream,
)))
}
-}
-impl DisplayAs for IcebergTableScan {
fn fmt_as(
&self,
- _t: datafusion::physical_plan::DisplayFormatType,
+ _t: DisplayFormatType,
f: &mut std::fmt::Formatter,
) -> std::fmt::Result {
- write!(
- f,
- "IcebergTableScan projection:[{}] predicate:[{}]",
- self.projection.join(","),
- self.predicates
- .clone()
- .map_or(String::from(""), |p| format!("{p}"))
- )?;
- if let Some(limit) = self.limit {
- write!(f, " limit:[{limit}]")?;
+ write!(f, "format=iceberg",)?;
+ write!(f, ", projection=[{}]", self.projection.join(", "))?;
+ if let Some(predicate) = &self.predicates {
+ write!(f, ", predicate={predicate}")?;
+ }
+ if let Some(fetch) = self.fetch {
+ write!(f, ", fetch={fetch}")?;
}
Ok(())
}
+
+ fn output_partitioning(&self) -> Partitioning {
+ Partitioning::UnknownPartitioning(1)
+ }
+
+ fn eq_properties(&self) -> EquivalenceProperties {
+ EquivalenceProperties::new(self.schema.clone())
+ }
+
+ fn partition_statistics(&self, _partition: Option<usize>) -> Result<Arc<Statistics>> {
+ Ok(Arc::new(Statistics::new_unknown(&self.schema)))
+ }
+
+ fn with_fetch(&self, fetch: Option<usize>) -> Option<Arc<dyn DataSource>> {
+ let mut source = self.clone();
+ source.fetch = fetch;
+ Some(Arc::new(source))
+ }
+
+ fn fetch(&self) -> Option<usize> {
+ self.fetch
+ }
+
+ fn try_swapping_with_projection(
+ &self,
+ _projection: &ProjectionExprs,
+ ) -> Result<Option<Arc<dyn DataSource>>> {
+ Ok(None)
+ }
}
/// Asynchronously retrieves a stream of [`RecordBatch`] instances
@@ -311,7 +231,7 @@ impl DisplayAs for IcebergTableScan {
async fn get_batch_stream(
table: Table,
snapshot_id: Option<i64>,
- column_names: Vec<String>,
+ column_names: Option<Vec<String>>,
predicates: Option<Predicate>,
) -> Result<Pin<Box<dyn Stream<Item = Result<RecordBatch>> + Send>>> {
let scan_builder = match snapshot_id {
@@ -319,7 +239,10 @@ async fn get_batch_stream(
None => table.scan(),
};
- let mut scan_builder = scan_builder.select(column_names);
+ let mut scan_builder = match column_names {
+ Some(column_names) => scan_builder.select(column_names),
+ None => scan_builder.select_all(),
+ };
if let Some(pred) = predicates {
scan_builder = scan_builder.with_filter(pred);
}
diff --git a/crates/datafusion/src/table_provider/catalog_table_provider.rs b/crates/datafusion/src/table_provider/catalog_table_provider.rs
index a069359..5145b42 100644
--- a/crates/datafusion/src/table_provider/catalog_table_provider.rs
+++ b/crates/datafusion/src/table_provider/catalog_table_provider.rs
@@ -22,6 +22,7 @@ use async_trait::async_trait;
use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
use datafusion::catalog::Session;
use datafusion::common::{config_datafusion_err, not_impl_err};
+use datafusion::datasource::source::DataSourceExec;
use datafusion::datasource::{TableProvider, TableType};
use datafusion::error::Result;
use datafusion::logical_expr::dml::InsertOp;
@@ -35,7 +36,7 @@ use iceberg::{Catalog, NamespaceIdent, TableIdent};
use crate::error::to_datafusion_error;
use crate::metadata::table::IcebergMetadataTableProvider;
-use crate::read::scan::IcebergTableScan;
+use crate::read::IcebergDataSource;
use crate::write::commit::IcebergCommitExec;
use crate::write::exec::IcebergWriteExec;
use crate::write::project::project_with_partition;
@@ -138,7 +139,7 @@ impl TableProvider for IcebergCatalogTableProvider {
.map_err(to_datafusion_error)?;
// Create scan with fresh metadata (always use current snapshot)
- Ok(Arc::new(IcebergTableScan::new(
+ Ok(DataSourceExec::from_data_source(IcebergDataSource::new(
table,
None, // Always use current snapshot for catalog-backed provider
self.schema.clone(),
@@ -429,14 +430,17 @@ mod tests {
// Test scan with limit
let scan_plan = provider.scan(&state, None, &[], Some(5)).await.unwrap();
- // Verify that the scan plan is an IcebergTableScan
+ // Verify that the scan plan wraps an IcebergDataSource.
let iceberg_scan = scan_plan
- .downcast_ref::<IcebergTableScan>()
- .expect("Expected IcebergTableScan");
+ .downcast_ref::<DataSourceExec>()
+ .expect("Expected DataSourceExec")
+ .data_source()
+ .downcast_ref::<IcebergDataSource>()
+ .expect("Expected IcebergDataSource");
// Verify the limit is set
assert_eq!(
- iceberg_scan.limit(),
+ iceberg_scan.fetch(),
Some(5),
"Limit should be set to 5 in the scan plan"
);
@@ -452,7 +456,7 @@ mod tests {
Arc::new(schema_to_arrow_schema(table.metadata().current_schema()).unwrap());
let out_of_range = schema.fields().len();
- let err = IcebergTableScan::new(
+ let err = IcebergDataSource::new(
table,
None,
schema,
diff --git a/crates/datafusion/src/table_provider/static_table_provider.rs b/crates/datafusion/src/table_provider/static_table_provider.rs
index 4a2f48f..2a435cc 100644
--- a/crates/datafusion/src/table_provider/static_table_provider.rs
+++ b/crates/datafusion/src/table_provider/static_table_provider.rs
@@ -21,6 +21,7 @@ use async_trait::async_trait;
use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
use datafusion::catalog::Session;
use datafusion::common::{exec_datafusion_err, not_impl_err};
+use datafusion::datasource::source::DataSourceExec;
use datafusion::datasource::{TableProvider, TableType};
use datafusion::error::Result;
use datafusion::logical_expr::dml::InsertOp;
@@ -30,7 +31,7 @@ use iceberg::arrow::schema_to_arrow_schema;
use iceberg::table::Table;
use crate::error::to_datafusion_error;
-use crate::read::scan::IcebergTableScan;
+use crate::read::IcebergDataSource;
/// Static table provider for read-only snapshot access.
///
@@ -125,7 +126,7 @@ impl TableProvider for IcebergStaticTableProvider {
limit: Option<usize>,
) -> Result<Arc<dyn ExecutionPlan>> {
// Use cached table (no refresh)
- Ok(Arc::new(IcebergTableScan::new(
+ Ok(DataSourceExec::from_data_source(IcebergDataSource::new(
self.table.clone(),
self.snapshot_id,
self.schema.clone(),
@@ -281,14 +282,17 @@ mod tests {
.await
.unwrap();
- // Verify that the scan plan is an IcebergTableScan
+ // Verify that the scan plan wraps an IcebergDataSource.
let iceberg_scan = scan_plan
- .downcast_ref::<IcebergTableScan>()
- .expect("Expected IcebergTableScan");
+ .downcast_ref::<DataSourceExec>()
+ .expect("Expected DataSourceExec")
+ .data_source()
+ .downcast_ref::<IcebergDataSource>()
+ .expect("Expected IcebergDataSource");
// Verify the limit is set
assert_eq!(
- iceberg_scan.limit(),
+ iceberg_scan.fetch(),
Some(10),
"Limit should be set to 10 in the scan plan"
);
@@ -310,14 +314,17 @@ mod tests {
// Test scan without limit
let scan_plan = table_provider.scan(&state, None, &[], None).await.unwrap();
- // Verify that the scan plan is an IcebergTableScan
+ // Verify that the scan plan wraps an IcebergDataSource.
let iceberg_scan = scan_plan
- .downcast_ref::<IcebergTableScan>()
- .expect("Expected IcebergTableScan");
+ .downcast_ref::<DataSourceExec>()
+ .expect("Expected DataSourceExec")
+ .data_source()
+ .downcast_ref::<IcebergDataSource>()
+ .expect("Expected IcebergDataSource");
// Verify the limit is None
assert_eq!(
- iceberg_scan.limit(),
+ iceberg_scan.fetch(),
None,
"Limit should be None when not specified"
);
diff --git a/crates/datafusion/tests/integration_datafusion_test.rs b/crates/datafusion/tests/integration_datafusion_test.rs
index 0c222c7..6511a15 100644
--- a/crates/datafusion/tests/integration_datafusion_test.rs
+++ b/crates/datafusion/tests/integration_datafusion_test.rs
@@ -29,13 +29,16 @@ use datafusion::arrow::compute::{
use datafusion::arrow::datatypes::{DataType, Field, Int64Type, Schema as ArrowSchema};
use datafusion::arrow::util::pretty::pretty_format_batches;
use datafusion::datasource::TableProvider;
+use datafusion::datasource::source::DataSourceExec;
use datafusion::execution::context::SessionContext;
use datafusion::parquet::arrow::PARQUET_FIELD_ID_META_KEY;
use datafusion::physical_plan::ExecutionPlan;
use datafusion::physical_plan::coalesce_partitions::CoalescePartitionsExec;
use datafusion::physical_plan::common::collect;
-use datafusion_iceberg::metadata::{IcebergMetadataScan, IcebergMetadataTableProvider};
-use datafusion_iceberg::read::IcebergTableScan;
+use datafusion_iceberg::metadata::{
+ IcebergMetadataDataSource, IcebergMetadataTableProvider,
+};
+use datafusion_iceberg::read::IcebergDataSource;
use datafusion_iceberg::write::{IcebergCommitExec, IcebergWriteExec};
use datafusion_iceberg::{
IcebergCatalogProvider, IcebergCatalogTableProvider, IcebergStaticTableProvider,
@@ -269,7 +272,10 @@ async fn test_table_projection() -> Result<(), Box<dyn Error>> {
.unwrap();
assert_eq!(2, s.len());
// the first row is logical_plan, the second row is physical_plan
- assert!(s.value(1).contains("projection:[foo1,foo2,foo3]"));
+ assert!(
+ s.value(1)
+ .contains("DataSourceExec: format=iceberg, projection=[foo1, foo2, foo3]")
+ );
// datafusion doesn't support query foo3.s_foo1, use foo3 instead
let records = table_df
@@ -290,7 +296,7 @@ async fn test_table_projection() -> Result<(), Box<dyn Error>> {
assert_eq!(2, s.len());
assert!(
s.value(1)
- .contains("IcebergTableScan projection:[foo1,foo3]")
+ .contains("DataSourceExec: format=iceberg, projection=[foo1, foo3]")
);
Ok(())
@@ -337,7 +343,7 @@ async fn test_table_predict_pushdown() -> Result<(), Box<dyn Error>> {
.unwrap();
assert_eq!(2, s.len());
// the first row is logical_plan, the second row is physical_plan
- let expected = "predicate:[(foo > 1) OR (bar IS NULL)]";
+ let expected = "predicate=(foo > 1) OR (bar IS NULL)";
assert!(s.value(1).trim().contains(expected));
Ok(())
}
@@ -1012,16 +1018,24 @@ async fn run(
}
/// Rebuilds `scan` from its accessors alone, as a codec would.
-fn rebuild_scan(scan: &IcebergTableScan) -> IcebergTableScan {
- IcebergTableScan::new_with_predicate(
+fn rebuild_scan(scan: &IcebergDataSource) -> IcebergDataSource {
+ IcebergDataSource::new_with_predicate(
scan.table().clone(),
scan.snapshot_id(),
scan.schema(),
scan.predicates().cloned(),
- scan.limit(),
+ scan.fetch(),
)
}
+async fn run_source(
+ source: &IcebergDataSource,
+ ctx: &SessionContext,
+) -> Result<String, Box<dyn Error>> {
+ let exec = DataSourceExec::from_data_source(source.clone());
+ run(exec.as_ref(), ctx).await
+}
+
/// Returns the first node of type `T` in `plan`, depth first.
fn find_node<T: ExecutionPlan + 'static>(plan: &Arc<dyn ExecutionPlan>) -> Option<&T> {
plan.downcast_ref::<T>()
@@ -1121,7 +1135,11 @@ async fn test_plan_nodes_are_inspectable() -> Result<(), Box<dyn Error>> {
.await?
.create_physical_plan()
.await?;
- let scan = find_node::<IcebergTableScan>(&plan).expect("a scan");
+ let scan_exec = find_node::<DataSourceExec>(&plan).expect("a data source");
+ let scan = scan_exec
+ .data_source()
+ .downcast_ref::<IcebergDataSource>()
+ .expect("an Iceberg data source");
assert_eq!(
scan.predicates().map(ToString::to_string).as_deref(),
Some("foo1 = 1")
@@ -1129,7 +1147,7 @@ async fn test_plan_nodes_are_inspectable() -> Result<(), Box<dyn Error>> {
let rebuilt = rebuild_scan(scan);
assert_eq!(rebuilt.schema(), scan.schema());
assert_eq!(rebuilt.projection(), scan.projection());
- let expected = run(scan, &ctx).await?;
+ let expected = run_source(scan, &ctx).await?;
expect![[r#"
+------+------+
| foo1 | foo2 |
@@ -1137,16 +1155,22 @@ async fn test_plan_nodes_are_inspectable() -> Result<(), Box<dyn Error>> {
| 1 | alan |
+------+------+"#]]
.assert_eq(&expected);
- assert_eq!(run(&rebuilt, &ctx).await?, expected);
+ assert_eq!(run_source(&rebuilt, &ctx).await?, expected);
// Without a projection the scan reads every column by name, and its limit
// is kept.
let plan = pinned.scan(&ctx.state(), None, &[], Some(1)).await?;
- let scan = plan.downcast_ref::<IcebergTableScan>().expect("a scan");
+ let scan_exec = plan
+ .downcast_ref::<DataSourceExec>()
+ .expect("a data source");
+ let scan = scan_exec
+ .data_source()
+ .downcast_ref::<IcebergDataSource>()
+ .expect("an Iceberg data source");
assert_eq!(scan.projection(), ["foo1".to_string(), "foo2".to_string()]);
let rebuilt = rebuild_scan(scan);
- assert_eq!(rebuilt.limit(), Some(1));
- let expected = run(scan, &ctx).await?;
+ assert_eq!(rebuilt.fetch(), Some(1));
+ let expected = run_source(scan, &ctx).await?;
expect![[r#"
+------+------+
| foo1 | foo2 |
@@ -1154,12 +1178,12 @@ async fn test_plan_nodes_are_inspectable() -> Result<(), Box<dyn Error>> {
| 1 | alan |
+------+------+"#]]
.assert_eq(&expected);
- assert_eq!(run(&rebuilt, &ctx).await?, expected);
+ assert_eq!(run_source(&rebuilt, &ctx).await?, expected);
// A scan reads the columns of its schema and no others, so one built over
// part of the table returns only those columns, from the pinned snapshot.
let foo2_only = Arc::new(pinned.schema().project(&[1])?);
- let partial = IcebergTableScan::new_with_predicate(
+ let partial = IcebergDataSource::new_with_predicate(
pinned.table().clone(),
Some(snapshot_id),
foo2_only,
@@ -1173,7 +1197,7 @@ async fn test_plan_nodes_are_inspectable() -> Result<(), Box<dyn Error>> {
| alan |
| turing |
+--------+"#]]
- .assert_eq(&run(&partial, &ctx).await?);
+ .assert_eq(&run_source(&partial, &ctx).await?);
// Metadata tables: a scan rebuilt from a metadata scan's parts reads the
// same rows.
@@ -1182,14 +1206,19 @@ async fn test_plan_nodes_are_inspectable() -> Result<(), Box<dyn Error>> {
.await?
.create_physical_plan()
.await?;
- let metadata_scan = find_node::<IcebergMetadataScan>(&plan).expect("a metadata scan");
+ let metadata_exec =
+ find_node::<DataSourceExec>(&plan).expect("a metadata data source");
+ let metadata_scan = metadata_exec
+ .data_source()
+ .downcast_ref::<IcebergMetadataDataSource>()
+ .expect("an Iceberg metadata source");
let provider = metadata_scan.provider();
assert_eq!(provider.table().identifier(), &ident);
- let rebuilt = IcebergMetadataScan::new(IcebergMetadataTableProvider::new(
+ let rebuilt = IcebergMetadataDataSource::new(IcebergMetadataTableProvider::new(
provider.table().clone(),
provider.metadata_type().clone(),
));
- let batches = run_batches(metadata_scan, &ctx).await?;
+ let batches = run_batches(metadata_exec, &ctx).await?;
// Snapshots come back in no set order, so put the first, which has no
// parent, first. Their ids, times and paths differ on every run, so each
// row is checked against its snapshot's metadata.
@@ -1275,8 +1304,10 @@ async fn test_plan_nodes_are_inspectable() -> Result<(), Box<dyn Error>> {
);
// The rebuilt scan holds the same table, so lists its snapshots in the
// same order.
+ let rebuilt_exec = DataSourceExec::from_data_source(rebuilt);
assert_eq!(
- pretty_format_batches(&run_batches(&rebuilt, &ctx).await?)?.to_string(),
+ pretty_format_batches(&run_batches(rebuilt_exec.as_ref(), &ctx).await?)?
+ .to_string(),
pretty_format_batches(&batches)?.to_string()
);
diff --git a/crates/sqllogictest/testdata/slts/df_test/basic_queries.slt b/crates/sqllogictest/testdata/slts/df_test/basic_queries.slt
index a5ca4de..05b0300 100644
--- a/crates/sqllogictest/testdata/slts/df_test/basic_queries.slt
+++ b/crates/sqllogictest/testdata/slts/df_test/basic_queries.slt
@@ -43,7 +43,7 @@ INSERT INTO default.default.query_test_table VALUES
----
10
-# Verify EXPLAIN shows limit is pushed down to IcebergTableScan
+# Verify EXPLAIN shows limit is pushed down to DataSourceExec
query TT
EXPLAIN SELECT * FROM default.default.query_test_table LIMIT 3
----
@@ -51,9 +51,8 @@ logical_plan
01)Limit: skip=0, fetch=3
02)--TableScan: default.default.query_test_table projection=[id, name, score, category], fetch=3
physical_plan
-01)GlobalLimitExec: skip=0, fetch=3
-02)--CooperativeExec
-03)----IcebergTableScan projection:[id,name,score,category] predicate:[] limit:[3]
+01)CooperativeExec
+02)--DataSourceExec: format=iceberg, projection=[id, name, score, category], fetch=3
# Test SELECT * with ORDER BY and LIMIT
query ITRT
diff --git a/crates/sqllogictest/testdata/slts/df_test/binary_predicate_pushdown.slt b/crates/sqllogictest/testdata/slts/df_test/binary_predicate_pushdown.slt
index aa68ab2..7c725ee 100644
--- a/crates/sqllogictest/testdata/slts/df_test/binary_predicate_pushdown.slt
+++ b/crates/sqllogictest/testdata/slts/df_test/binary_predicate_pushdown.slt
@@ -15,10 +15,10 @@
# specific language governing permissions and limitations
# under the License.
-# Test that binary predicates are pushed down to IcebergTableScan
+# Test that binary predicates are pushed down to DataSourceExec
# This validates that ScalarValue::Binary is correctly converted to Datum::binary
-# Verify EXPLAIN shows binary predicate is pushed down to IcebergTableScan
+# Verify EXPLAIN shows binary predicate is pushed down to DataSourceExec
query TT
EXPLAIN SELECT * FROM default.default.test_binary_table WHERE data = X'0102'
----
@@ -28,7 +28,7 @@ logical_plan
physical_plan
01)FilterExec: data@1 = 0102
02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----IcebergTableScan projection:[id,data] predicate:[data = 0102]
+03)----DataSourceExec: format=iceberg, projection=[id, data], predicate=data = 0102
# Verify empty result from empty table
query I?
diff --git a/crates/sqllogictest/testdata/slts/df_test/boolean_predicate_pushdown.slt b/crates/sqllogictest/testdata/slts/df_test/boolean_predicate_pushdown.slt
index 496f719..9bd2542 100644
--- a/crates/sqllogictest/testdata/slts/df_test/boolean_predicate_pushdown.slt
+++ b/crates/sqllogictest/testdata/slts/df_test/boolean_predicate_pushdown.slt
@@ -29,7 +29,7 @@ INSERT INTO default.default.test_boolean_table VALUES
(5, true, 'Verified account'),
(6, NULL, 'Pending verification')
-# Verify boolean equality predicate is pushed down to IcebergTableScan
+# Verify boolean equality predicate is pushed down to DataSourceExec
query TT
EXPLAIN SELECT * FROM default.default.test_boolean_table WHERE is_active = true
----
@@ -39,7 +39,7 @@ logical_plan
physical_plan
01)FilterExec: is_active@1
02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----IcebergTableScan projection:[id,is_active,description] predicate:[is_active = true]
+03)----DataSourceExec: format=iceberg, projection=[id, is_active, description], predicate=is_active = true
# Query with is_active = true
query ITT rowsort
@@ -49,7 +49,7 @@ SELECT * FROM default.default.test_boolean_table WHERE is_active = true
3 true Premium member
5 true Verified account
-# Verify boolean false predicate is pushed down to IcebergTableScan
+# Verify boolean false predicate is pushed down to DataSourceExec
query TT
EXPLAIN SELECT * FROM default.default.test_boolean_table WHERE is_active = false
----
@@ -59,7 +59,7 @@ logical_plan
physical_plan
01)FilterExec: NOT is_active@1
02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----IcebergTableScan projection:[id,is_active,description] predicate:[is_active = false]
+03)----DataSourceExec: format=iceberg, projection=[id, is_active, description], predicate=is_active = false
# Query with is_active = false
query ITT rowsort
@@ -78,7 +78,7 @@ logical_plan
physical_plan
01)FilterExec: NOT is_active@1
02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----IcebergTableScan projection:[id,is_active,description] predicate:[is_active = false]
+03)----DataSourceExec: format=iceberg, projection=[id, is_active, description], predicate=is_active = false
# Query with is_active != true (includes false and NULL)
query ITT rowsort
diff --git a/crates/sqllogictest/testdata/slts/df_test/like_predicate_pushdown.slt b/crates/sqllogictest/testdata/slts/df_test/like_predicate_pushdown.slt
index 3d8b151..5331dd3 100644
--- a/crates/sqllogictest/testdata/slts/df_test/like_predicate_pushdown.slt
+++ b/crates/sqllogictest/testdata/slts/df_test/like_predicate_pushdown.slt
@@ -27,7 +27,7 @@ INSERT INTO default.default.test_unpartitioned_table VALUES (5, 'alice'), (6, 'A
----
4
-# Test basic LIKE prefix pattern pushdown - verify predicate is pushed to IcebergTableScan
+# Test basic LIKE prefix pattern pushdown - verify predicate is pushed to DataSourceExec
query TT
EXPLAIN SELECT * FROM default.default.test_unpartitioned_table WHERE name LIKE 'Al%'
----
@@ -37,7 +37,7 @@ logical_plan
physical_plan
01)FilterExec: name@1 LIKE Al%
02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----IcebergTableScan projection:[id,name] predicate:[name STARTS WITH "Al"]
+03)----DataSourceExec: format=iceberg, projection=[id, name], predicate=name STARTS WITH "Al"
# Test LIKE filtering with case-sensitive match
query IT rowsort
@@ -56,7 +56,7 @@ logical_plan
physical_plan
01)FilterExec: name@1 NOT LIKE Al%
02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----IcebergTableScan projection:[id,name] predicate:[name NOT STARTS WITH "Al"]
+03)----DataSourceExec: format=iceberg, projection=[id, name], predicate=name NOT STARTS WITH "Al"
# Test NOT LIKE filtering
query IT rowsort
diff --git a/crates/sqllogictest/testdata/slts/df_test/timestamp_predicate_pushdown.slt b/crates/sqllogictest/testdata/slts/df_test/timestamp_predicate_pushdown.slt
index 8b1f833..6f7b410 100644
--- a/crates/sqllogictest/testdata/slts/df_test/timestamp_predicate_pushdown.slt
+++ b/crates/sqllogictest/testdata/slts/df_test/timestamp_predicate_pushdown.slt
@@ -50,7 +50,7 @@ logical_plan
physical_plan
01)FilterExec: ts@1 = 1672921800000000000
02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----IcebergTableScan projection:[id,ts] predicate:[ts = 2023-01-05 12:30:00]
+03)----DataSourceExec: format=iceberg, projection=[id, ts], predicate=ts = 2023-01-05 12:30:00
# Verify timestamp equality filtering works
query I?
@@ -68,7 +68,7 @@ logical_plan
physical_plan
01)FilterExec: ts@1 > 1673308800000000000
02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----IcebergTableScan projection:[id,ts] predicate:[ts > 2023-01-10 00:00:00]
+03)----DataSourceExec: format=iceberg, projection=[id, ts], predicate=ts > 2023-01-10 00:00:00
# Verify timestamp greater than filtering
query I? rowsort
@@ -97,7 +97,7 @@ logical_plan
physical_plan
01)FilterExec: ts@1 >= 1672876800000000000 AND ts@1 <= 1673827199000000000
02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----IcebergTableScan projection:[id,ts] predicate:[(ts >= 2023-01-05 00:00:00) AND (ts <= 2023-01-15 23:59:59)]
+03)----DataSourceExec: format=iceberg, projection=[id, ts], predicate=(ts >= 2023-01-05 00:00:00) AND (ts <= 2023-01-15 23:59:59)
# Test timestamp range predicate filtering
query I? rowsort
@@ -162,7 +162,7 @@ logical_plan
physical_plan
01)FilterExec: ts@1 > 1672531200000000
02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----IcebergTableScan projection:[id,ts] predicate:[ts > 2023-01-01 00:00:00]
+03)----DataSourceExec: format=iceberg, projection=[id, ts], predicate=ts > 2023-01-01 00:00:00
query I?
SELECT * FROM default.default.test_timestamp_micros WHERE ts > CAST('2023-01-01 00:00:00' AS TIMESTAMP)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-iceberg) 01/02: Refactor folder structure
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
manuzhang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-iceberg.git
commit 79d68a5a93f293c78b1b29fa8927b7acf6c929da
Author: Gabriel <ga...@datadoghq.com>
AuthorDate: Mon Oct 5 15:20:26 2026 +0200
Refactor folder structure
---
AGENTS.md | 5 +-
Cargo.lock | 12 +-
crates/datafusion/src/{lib.rs => catalog/mod.rs} | 13 +-
.../src/{catalog.rs => catalog/provider.rs} | 2 +-
crates/datafusion/src/{ => catalog}/schema.rs | 12 +-
crates/datafusion/src/lib.rs | 14 +-
crates/datafusion/src/{lib.rs => metadata/mod.rs} | 16 +-
.../metadata_scan.rs => metadata/scan.rs} | 2 +-
.../{table/metadata_table.rs => metadata/table.rs} | 2 +-
crates/datafusion/src/{lib.rs => read/mod.rs} | 16 +-
.../expr_to_predicate.rs => read/predicate.rs} | 2 +-
.../datafusion/src/{physical_plan => read}/scan.rs | 4 +-
.../catalog_table_provider.rs} | 767 ++++++---------------
.../factory.rs} | 0
.../src/{lib.rs => table_provider/mod.rs} | 21 +-
.../src/table_provider/static_table_provider.rs | 345 +++++++++
.../src/{physical_plan => write}/commit.rs | 7 +-
.../src/{physical_plan/write.rs => write/exec.rs} | 5 +-
.../datafusion/src/{physical_plan => write}/mod.rs | 13 +-
.../src/{physical_plan => write}/project.rs | 2 +-
.../src/{physical_plan => write}/repartition.rs | 6 +-
.../src/{physical_plan => write}/sort.rs | 0
crates/datafusion/src/{ => write}/task_writer.rs | 0
.../tests/integration_datafusion_test.rs | 18 +-
24 files changed, 642 insertions(+), 642 deletions(-)
diff --git a/AGENTS.md b/AGENTS.md
index bc1fa61..c2ba657 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -58,14 +58,14 @@ Tests are self-contained (in-memory catalogs, temp dirs, checked-in metadata JSO
### Catalog → schema → table
-`IcebergCatalogProvider` wraps an `iceberg::Catalog` and builds one `IcebergSchemaProvider` per namespace, each holding an `IcebergTableProvider` per table. Namespaces and tables are listed once, at construction, and are never refreshed. Tables created by other clients after that point are invisible.
+`IcebergCatalogProvider` wraps an `iceberg::Catalog` and builds one `IcebergSchemaProvider` per namespace, each holding an `IcebergCatalogTableProvider` per table. Namespaces and tables are listed once, at construction, and are never refreshed. Tables created by other clients after that point are invisible.
- Names like `<table>$snapshots`, `$manifests`, and `$history` resolve to `IcebergMetadataTableProvider`.
- SQL `CREATE TABLE` and `DROP TABLE` reach the *synchronous* `SchemaProvider::register_table` and `deregister_table` methods, which call the async catalog through `spawn_blocking` + `block_on`. `register_table` rejects input tables that contain rows, so CTAS isn't supported. It auto-assigns field IDs and uses format V2 unless the schema needs V3.
### Table providers (`table/`)
-- **`IcebergTableProvider`**: catalog-backed. It caches the Arrow schema at construction but reloads table metadata from the catalog on every `scan` and `insert_into`. INSERT supports append only.
+- **`IcebergCatalogTableProvider`**: catalog-backed. It caches the Arrow schema at construction but reloads table metadata from the catalog on every `scan` and `insert_into`. INSERT supports append only.
- **`IcebergStaticTableProvider`**: wraps a fixed `Table`, optionally pinned to a snapshot id for time travel. Read-only.
- **`IcebergTableProviderFactory`**: handles `CREATE EXTERNAL TABLE t STORED AS ICEBERG LOCATION '<metadata.json>'` and produces a static provider. Bare names get the `default` namespace. Schema, partition, and order clauses are rejected.
@@ -114,5 +114,6 @@ Library code returns `datafusion::error::Result`. Convert iceberg errors with `t
## Conventions
+- Reserve `mod.rs` and `lib.rs` for module declarations and public re-exports. Put implementations, constants, and unit tests in the corresponding named source files.
- New files need the ASF license header; copy it from a neighbouring file.
- Any GitHub Action added to a workflow must be on the ASF allowlist; the `asf-allowlist-check` workflow enforces this.
diff --git a/Cargo.lock b/Cargo.lock
index 945361f..6a4d077 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1684,9 +1684,9 @@ dependencies = [
[[package]]
name = "datafusion-cli"
-version = "55.1.0"
+version = "55.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ffb9833a102f01a257994de966c73fbf7441d8347207d8676c821b7318c03ee6"
+checksum = "5d8b87ff78e243da48b836ced274799a1799b4a2fd321700e4b8f3426c01aa7e"
dependencies = [
"arrow",
"async-trait",
@@ -2297,9 +2297,9 @@ dependencies = [
[[package]]
name = "datafusion-spark"
-version = "55.1.0"
+version = "55.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f7809ecefb16fcb950c6b22ee3df99cb412dca6fae72c329d64c938ae53254f9"
+checksum = "992dd0b954f24cb576cbeee3555c3083b9cf7a5a5f2448ea25f7a437d444163f"
dependencies = [
"arrow",
"bigdecimal",
@@ -2347,9 +2347,9 @@ dependencies = [
[[package]]
name = "datafusion-sqllogictest"
-version = "55.1.0"
+version = "55.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d408561adfed7077e110911c9945de623551f4101367e1ef55dc5eda7729afb4"
+checksum = "2c921345d249b784a513ad2be71fc3df18fd8f019fe6c82233f085a1871154fd"
dependencies = [
"arrow",
"async-trait",
diff --git a/crates/datafusion/src/lib.rs b/crates/datafusion/src/catalog/mod.rs
similarity index 79%
copy from crates/datafusion/src/lib.rs
copy to crates/datafusion/src/catalog/mod.rs
index 4b0ea86..253aa98 100644
--- a/crates/datafusion/src/lib.rs
+++ b/crates/datafusion/src/catalog/mod.rs
@@ -15,16 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-mod catalog;
-pub use catalog::*;
-
-mod error;
-pub use error::*;
-
-pub mod physical_plan;
+mod provider;
mod schema;
-pub mod table;
-pub use table::table_provider_factory::IcebergTableProviderFactory;
-pub use table::*;
-pub(crate) mod task_writer;
+pub use provider::IcebergCatalogProvider;
diff --git a/crates/datafusion/src/catalog.rs b/crates/datafusion/src/catalog/provider.rs
similarity index 98%
rename from crates/datafusion/src/catalog.rs
rename to crates/datafusion/src/catalog/provider.rs
index a96f731..9784e45 100644
--- a/crates/datafusion/src/catalog.rs
+++ b/crates/datafusion/src/catalog/provider.rs
@@ -23,7 +23,7 @@ use datafusion::error::Result;
use futures::future::try_join_all;
use iceberg::{Catalog, NamespaceIdent};
-use crate::schema::IcebergSchemaProvider;
+use super::schema::IcebergSchemaProvider;
use crate::to_datafusion_error;
/// Provides an interface to manage and access multiple schemas
diff --git a/crates/datafusion/src/schema.rs b/crates/datafusion/src/catalog/schema.rs
similarity index 97%
rename from crates/datafusion/src/schema.rs
rename to crates/datafusion/src/catalog/schema.rs
index 6bc4b4d..900e874 100644
--- a/crates/datafusion/src/schema.rs
+++ b/crates/datafusion/src/catalog/schema.rs
@@ -32,7 +32,7 @@ use iceberg::inspect::MetadataTableType;
use iceberg::spec::FormatVersion;
use iceberg::{Catalog, NamespaceIdent, TableCreation, TableIdent};
-use crate::table::IcebergTableProvider;
+use crate::table_provider::IcebergCatalogTableProvider;
use crate::to_datafusion_error;
/// Represents a [`SchemaProvider`] for the Iceberg [`Catalog`], managing
@@ -47,7 +47,7 @@ pub(crate) struct IcebergSchemaProvider {
/// and values are dynamic references to objects implementing the
/// [`TableProvider`] trait.
/// Wrapped in Arc to allow sharing across async boundaries in register_table.
- tables: Arc<DashMap<String, Arc<IcebergTableProvider>>>,
+ tables: Arc<DashMap<String, Arc<IcebergCatalogTableProvider>>>,
}
impl IcebergSchemaProvider {
@@ -78,7 +78,11 @@ impl IcebergSchemaProvider {
table_names
.iter()
.map(|name| {
- IcebergTableProvider::try_new(client.clone(), namespace.clone(), name)
+ IcebergCatalogTableProvider::try_new(
+ client.clone(),
+ namespace.clone(),
+ name,
+ )
})
.collect::<Vec<_>>(),
)
@@ -187,7 +191,7 @@ impl SchemaProvider for IcebergSchemaProvider {
.map_err(to_datafusion_error)?;
// Create a new table provider using the catalog reference
- let table_provider = IcebergTableProvider::try_new(
+ let table_provider = IcebergCatalogTableProvider::try_new(
catalog.clone(),
namespace.clone(),
name_clone.clone(),
diff --git a/crates/datafusion/src/lib.rs b/crates/datafusion/src/lib.rs
index 4b0ea86..e714176 100644
--- a/crates/datafusion/src/lib.rs
+++ b/crates/datafusion/src/lib.rs
@@ -21,10 +21,10 @@ pub use catalog::*;
mod error;
pub use error::*;
-pub mod physical_plan;
-mod schema;
-pub mod table;
-pub use table::table_provider_factory::IcebergTableProviderFactory;
-pub use table::*;
-
-pub(crate) mod task_writer;
+pub mod metadata;
+pub mod read;
+pub mod table_provider;
+pub use table_provider::{
+ IcebergCatalogTableProvider, IcebergStaticTableProvider, IcebergTableProviderFactory,
+};
+pub mod write;
diff --git a/crates/datafusion/src/lib.rs b/crates/datafusion/src/metadata/mod.rs
similarity index 77%
copy from crates/datafusion/src/lib.rs
copy to crates/datafusion/src/metadata/mod.rs
index 4b0ea86..33893fe 100644
--- a/crates/datafusion/src/lib.rs
+++ b/crates/datafusion/src/metadata/mod.rs
@@ -15,16 +15,8 @@
// specific language governing permissions and limitations
// under the License.
-mod catalog;
-pub use catalog::*;
+pub(crate) mod scan;
+pub(crate) mod table;
-mod error;
-pub use error::*;
-
-pub mod physical_plan;
-mod schema;
-pub mod table;
-pub use table::table_provider_factory::IcebergTableProviderFactory;
-pub use table::*;
-
-pub(crate) mod task_writer;
+pub use scan::IcebergMetadataScan;
+pub use table::IcebergMetadataTableProvider;
diff --git a/crates/datafusion/src/physical_plan/metadata_scan.rs b/crates/datafusion/src/metadata/scan.rs
similarity index 98%
rename from crates/datafusion/src/physical_plan/metadata_scan.rs
rename to crates/datafusion/src/metadata/scan.rs
index 1394865..be8d36d 100644
--- a/crates/datafusion/src/physical_plan/metadata_scan.rs
+++ b/crates/datafusion/src/metadata/scan.rs
@@ -25,7 +25,7 @@ use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
use datafusion::physical_plan::{DisplayAs, ExecutionPlan, Partitioning, PlanProperties};
use futures::TryStreamExt;
-use crate::metadata_table::IcebergMetadataTableProvider;
+use super::table::IcebergMetadataTableProvider;
/// Scans an Iceberg metadata table, such as `$snapshots`.
#[derive(Debug)]
diff --git a/crates/datafusion/src/table/metadata_table.rs b/crates/datafusion/src/metadata/table.rs
similarity index 98%
rename from crates/datafusion/src/table/metadata_table.rs
rename to crates/datafusion/src/metadata/table.rs
index b2f3d19..f3c91f3 100644
--- a/crates/datafusion/src/table/metadata_table.rs
+++ b/crates/datafusion/src/metadata/table.rs
@@ -31,7 +31,7 @@ use iceberg::arrow::schema_to_arrow_schema;
use iceberg::inspect::MetadataTableType;
use iceberg::table::Table;
-use crate::physical_plan::metadata_scan::IcebergMetadataScan;
+use super::scan::IcebergMetadataScan;
use crate::to_datafusion_error;
/// Represents a [`TableProvider`] for the Iceberg [`iceberg::Catalog`],
diff --git a/crates/datafusion/src/lib.rs b/crates/datafusion/src/read/mod.rs
similarity index 77%
copy from crates/datafusion/src/lib.rs
copy to crates/datafusion/src/read/mod.rs
index 4b0ea86..7933c39 100644
--- a/crates/datafusion/src/lib.rs
+++ b/crates/datafusion/src/read/mod.rs
@@ -15,16 +15,8 @@
// specific language governing permissions and limitations
// under the License.
-mod catalog;
-pub use catalog::*;
+pub(crate) mod predicate;
+pub(crate) mod scan;
-mod error;
-pub use error::*;
-
-pub mod physical_plan;
-mod schema;
-pub mod table;
-pub use table::table_provider_factory::IcebergTableProviderFactory;
-pub use table::*;
-
-pub(crate) mod task_writer;
+pub use predicate::convert_filters_to_predicate;
+pub use scan::IcebergTableScan;
diff --git a/crates/datafusion/src/physical_plan/expr_to_predicate.rs b/crates/datafusion/src/read/predicate.rs
similarity index 99%
rename from crates/datafusion/src/physical_plan/expr_to_predicate.rs
rename to crates/datafusion/src/read/predicate.rs
index 58b3ee2..bb82720 100644
--- a/crates/datafusion/src/physical_plan/expr_to_predicate.rs
+++ b/crates/datafusion/src/read/predicate.rs
@@ -267,7 +267,7 @@ fn scalar_function_to_iceberg_predicate(
/// Multiplication/division by zero and `c / x` are intentionally rejected: e.g.
/// `x * 0` is NaN when `x` is `±inf`, so it does not imply `x` is NaN.
///
-/// [`IcebergTableProvider::supports_filters_pushdown`]: crate::table::IcebergTableProvider
+/// [`IcebergTableProvider::supports_filters_pushdown`]: crate::table_provider::IcebergCatalogTableProvider
fn resolve_nan_preserving_reference(expr: &Expr) -> Option<Reference> {
match expr {
Expr::Column(column) => Some(Reference::new(column.name())),
diff --git a/crates/datafusion/src/physical_plan/scan.rs b/crates/datafusion/src/read/scan.rs
similarity index 98%
rename from crates/datafusion/src/physical_plan/scan.rs
rename to crates/datafusion/src/read/scan.rs
index 6ede66d..71a1fb2 100644
--- a/crates/datafusion/src/physical_plan/scan.rs
+++ b/crates/datafusion/src/read/scan.rs
@@ -33,7 +33,7 @@ use futures::{Stream, TryStreamExt};
use iceberg::expr::Predicate;
use iceberg::table::Table;
-use super::expr_to_predicate::convert_filters_to_predicate;
+use super::predicate::convert_filters_to_predicate;
use crate::to_datafusion_error;
/// Manages the scanning process of an Iceberg [`Table`], encapsulating the
@@ -107,7 +107,7 @@ impl IcebergTableScan {
/// use datafusion::physical_plan::ExecutionPlan;
/// use datafusion::prelude::{SessionContext, col, lit};
/// use datafusion_iceberg::IcebergStaticTableProvider;
- /// use datafusion_iceberg::physical_plan::IcebergTableScan;
+ /// use datafusion_iceberg::read::IcebergTableScan;
/// use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder};
/// use iceberg::spec::{NestedField, PrimitiveType, Schema, Type};
/// use iceberg::{Catalog, CatalogBuilder, NamespaceIdent, TableCreation};
diff --git a/crates/datafusion/src/table/mod.rs b/crates/datafusion/src/table_provider/catalog_table_provider.rs
similarity index 63%
rename from crates/datafusion/src/table/mod.rs
rename to crates/datafusion/src/table_provider/catalog_table_provider.rs
index dc20a4a..a069359 100644
--- a/crates/datafusion/src/table/mod.rs
+++ b/crates/datafusion/src/table_provider/catalog_table_provider.rs
@@ -15,26 +15,13 @@
// specific language governing permissions and limitations
// under the License.
-//! Iceberg table providers for DataFusion.
-//!
-//! This module provides two table provider implementations:
-//!
-//! - [`IcebergTableProvider`]: Catalog-backed provider with automatic metadata refresh.
-//! Use for write operations and when you need to see the latest table state.
-//!
-//! - [`IcebergStaticTableProvider`]: Static provider for read-only access to a specific
-//! table snapshot. Use for consistent analytical queries or time-travel scenarios.
-
-pub mod metadata_table;
-pub mod table_provider_factory;
-
use std::num::NonZeroUsize;
use std::sync::Arc;
use async_trait::async_trait;
use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
use datafusion::catalog::Session;
-use datafusion::common::{config_datafusion_err, exec_datafusion_err, not_impl_err};
+use datafusion::common::{config_datafusion_err, not_impl_err};
use datafusion::datasource::{TableProvider, TableType};
use datafusion::error::Result;
use datafusion::logical_expr::dml::InsertOp;
@@ -44,17 +31,16 @@ use datafusion::physical_plan::coalesce_partitions::CoalescePartitionsExec;
use iceberg::arrow::schema_to_arrow_schema;
use iceberg::inspect::MetadataTableType;
use iceberg::spec::TableProperties;
-use iceberg::table::Table;
use iceberg::{Catalog, NamespaceIdent, TableIdent};
-pub use metadata_table::IcebergMetadataTableProvider;
use crate::error::to_datafusion_error;
-use crate::physical_plan::commit::IcebergCommitExec;
-use crate::physical_plan::project::project_with_partition;
-use crate::physical_plan::repartition::repartition;
-use crate::physical_plan::scan::IcebergTableScan;
-use crate::physical_plan::sort::sort_by_partition;
-use crate::physical_plan::write::IcebergWriteExec;
+use crate::metadata::table::IcebergMetadataTableProvider;
+use crate::read::scan::IcebergTableScan;
+use crate::write::commit::IcebergCommitExec;
+use crate::write::exec::IcebergWriteExec;
+use crate::write::project::project_with_partition;
+use crate::write::repartition::repartition;
+use crate::write::sort::sort_by_partition;
/// Catalog-backed table provider with automatic metadata refresh.
///
@@ -63,9 +49,9 @@ use crate::physical_plan::write::IcebergWriteExec;
/// write operations or want to see the most up-to-date data.
///
/// For read-only access to a specific snapshot without catalog overhead, use
-/// [`IcebergStaticTableProvider`] instead.
+/// [`IcebergStaticTableProvider`](super::IcebergStaticTableProvider) instead.
#[derive(Debug, Clone)]
-pub struct IcebergTableProvider {
+pub struct IcebergCatalogTableProvider {
/// The catalog that manages this table
catalog: Arc<dyn Catalog>,
/// The table identifier (namespace + name)
@@ -74,7 +60,7 @@ pub struct IcebergTableProvider {
schema: ArrowSchemaRef,
}
-impl IcebergTableProvider {
+impl IcebergCatalogTableProvider {
/// Creates a new catalog-backed table provider.
///
/// Loads the table once to get the initial schema, then stores the catalog
@@ -96,7 +82,7 @@ impl IcebergTableProvider {
.map_err(to_datafusion_error)?,
);
- Ok(IcebergTableProvider {
+ Ok(IcebergCatalogTableProvider {
catalog,
table_ident,
schema,
@@ -128,7 +114,7 @@ impl IcebergTableProvider {
}
#[async_trait]
-impl TableProvider for IcebergTableProvider {
+impl TableProvider for IcebergCatalogTableProvider {
fn schema(&self) -> ArrowSchemaRef {
self.schema.clone()
}
@@ -242,326 +228,29 @@ impl TableProvider for IcebergTableProvider {
}
}
-/// Static table provider for read-only snapshot access.
-///
-/// This provider holds a cached table instance and does not refresh metadata or support
-/// write operations. Use this for consistent analytical queries, time-travel scenarios,
-/// or when you want to avoid catalog overhead.
-///
-/// For catalog-backed tables with write support and automatic refresh, use
-/// [`IcebergTableProvider`] instead.
-#[derive(Debug, Clone)]
-pub struct IcebergStaticTableProvider {
- /// The static table instance (never refreshed)
- table: Table,
- /// Optional snapshot ID for this static view
- snapshot_id: Option<i64>,
- /// A reference-counted arrow `Schema`
- schema: ArrowSchemaRef,
-}
-
-impl IcebergStaticTableProvider {
- /// Creates a static provider from a table instance.
- ///
- /// Uses the table's current snapshot for all queries. Does not support write operations.
- pub async fn try_new_from_table(table: Table) -> Result<Self> {
- let schema = Arc::new(
- schema_to_arrow_schema(table.metadata().current_schema())
- .map_err(to_datafusion_error)?,
- );
- Ok(IcebergStaticTableProvider {
- table,
- snapshot_id: None,
- schema,
- })
- }
-
- /// Creates a static provider for a specific table snapshot.
- ///
- /// Queries the specified snapshot for all operations. Useful for time-travel queries.
- /// Does not support write operations.
- pub async fn try_new_from_table_snapshot(
- table: Table,
- snapshot_id: i64,
- ) -> Result<Self> {
- let snapshot = table
- .metadata()
- .snapshot_by_id(snapshot_id)
- .ok_or_else(|| {
- exec_datafusion_err!(
- "snapshot id {snapshot_id} not found in table {}",
- table.identifier().name()
- )
- })?;
- let table_schema = snapshot
- .schema(table.metadata())
- .map_err(to_datafusion_error)?;
- let schema =
- Arc::new(schema_to_arrow_schema(&table_schema).map_err(to_datafusion_error)?);
- Ok(IcebergStaticTableProvider {
- table,
- snapshot_id: Some(snapshot_id),
- schema,
- })
- }
-
- /// The table as loaded when this provider was built.
- pub fn table(&self) -> &Table {
- &self.table
- }
-
- /// The snapshot this provider reads, or `None` for the current snapshot of
- /// [`Self::table`].
- pub fn snapshot_id(&self) -> Option<i64> {
- self.snapshot_id
- }
-}
-
-#[async_trait]
-impl TableProvider for IcebergStaticTableProvider {
- fn schema(&self) -> ArrowSchemaRef {
- self.schema.clone()
- }
-
- fn table_type(&self) -> TableType {
- TableType::Base
- }
-
- async fn scan(
- &self,
- _state: &dyn Session,
- projection: Option<&Vec<usize>>,
- filters: &[Expr],
- limit: Option<usize>,
- ) -> Result<Arc<dyn ExecutionPlan>> {
- // Use cached table (no refresh)
- Ok(Arc::new(IcebergTableScan::new(
- self.table.clone(),
- self.snapshot_id,
- self.schema.clone(),
- projection,
- filters,
- limit,
- )?))
- }
-
- fn supports_filters_pushdown(
- &self,
- filters: &[&Expr],
- ) -> Result<Vec<TableProviderFilterPushDown>> {
- // Push down all filters, as a single source of truth, the scanner will drop the filters which couldn't be push down
- Ok(vec![TableProviderFilterPushDown::Inexact; filters.len()])
- }
-
- async fn insert_into(
- &self,
- _state: &dyn Session,
- _input: Arc<dyn ExecutionPlan>,
- _insert_op: InsertOp,
- ) -> Result<Arc<dyn ExecutionPlan>> {
- not_impl_err!(
- "Write operations are not supported on IcebergStaticTableProvider. \
- Use IcebergTableProvider with a catalog for write support."
- )
- }
-}
-
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use std::sync::Arc;
- use datafusion::common::Column;
+ use datafusion::datasource::TableProvider;
use datafusion::error::DataFusionError;
use datafusion::physical_plan::ExecutionPlan;
use datafusion::prelude::SessionContext;
- use iceberg::io::FileIO;
use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder};
use iceberg::spec::{NestedField, PrimitiveType, Schema, Type};
- use iceberg::table::{StaticTable, Table};
- use iceberg::{Catalog, CatalogBuilder, NamespaceIdent, TableCreation, TableIdent};
+ use iceberg::{CatalogBuilder, TableCreation};
use tempfile::TempDir;
use super::*;
- async fn get_test_table_from_metadata_file() -> Table {
- let metadata_file_name = "TableMetadataV2Valid.json";
- let metadata_file_path = format!(
- "{}/tests/test_data/{}",
- env!("CARGO_MANIFEST_DIR"),
- metadata_file_name
- );
- let file_io = FileIO::new_with_fs();
- let static_identifier =
- TableIdent::from_strs(["static_ns", "static_table"]).unwrap();
- let static_table = StaticTable::from_metadata_file(
- &metadata_file_path,
- static_identifier,
- file_io,
- )
- .await
- .unwrap();
- static_table.into_table()
- }
-
- async fn get_test_catalog_and_table()
- -> (Arc<dyn Catalog>, NamespaceIdent, String, TempDir) {
- let temp_dir = TempDir::new().unwrap();
- let warehouse_path = temp_dir.path().to_str().unwrap().to_string();
-
- let catalog = MemoryCatalogBuilder::default()
- .load(
- "memory",
- HashMap::from([(
- MEMORY_CATALOG_WAREHOUSE.to_string(),
- warehouse_path.clone(),
- )]),
- )
- .await
- .unwrap();
-
- let namespace = NamespaceIdent::new("test_ns".to_string());
- catalog
- .create_namespace(&namespace, HashMap::new())
- .await
- .unwrap();
-
- let schema = Schema::builder()
- .with_schema_id(0)
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(2, "name", Type::Primitive(PrimitiveType::String))
- .into(),
- ])
- .build()
- .unwrap();
-
- let table_creation = TableCreation::builder()
- .name("test_table".to_string())
- .location(format!("{warehouse_path}/test_table"))
- .schema(schema)
- .properties(HashMap::new())
- .build();
-
- catalog
- .create_table(&namespace, table_creation)
- .await
- .unwrap();
-
- (
- Arc::new(catalog),
- namespace,
- "test_table".to_string(),
- temp_dir,
- )
- }
-
- // Tests for IcebergStaticTableProvider
-
- #[tokio::test]
- async fn test_static_provider_from_table() {
- let table = get_test_table_from_metadata_file().await;
- let table_provider =
- IcebergStaticTableProvider::try_new_from_table(table.clone())
- .await
- .unwrap();
- let ctx = SessionContext::new();
- ctx.register_table("mytable", Arc::new(table_provider))
- .unwrap();
- let df = ctx.sql("SELECT * FROM mytable").await.unwrap();
- let df_schema = df.schema();
- let df_columns = df_schema.fields();
- assert_eq!(df_columns.len(), 3);
- let x_column = df_columns.first().unwrap();
- let column_data = format!(
- "{:?}:{:?}",
- x_column.name(),
- x_column.data_type().to_string()
- );
- assert_eq!(column_data, "\"x\":\"Int64\"");
- let has_column = df_schema.has_column(&Column::from_name("z"));
- assert!(has_column);
- }
-
- #[tokio::test]
- async fn test_static_provider_from_snapshot() {
- let table = get_test_table_from_metadata_file().await;
- let snapshot_id = table.metadata().snapshots().next().unwrap().snapshot_id();
- let table_provider = IcebergStaticTableProvider::try_new_from_table_snapshot(
- table.clone(),
- snapshot_id,
- )
- .await
- .unwrap();
- let ctx = SessionContext::new();
- ctx.register_table("mytable", Arc::new(table_provider))
- .unwrap();
- let df = ctx.sql("SELECT * FROM mytable").await.unwrap();
- let df_schema = df.schema();
- let df_columns = df_schema.fields();
- assert_eq!(df_columns.len(), 3);
- let x_column = df_columns.first().unwrap();
- let column_data = format!(
- "{:?}:{:?}",
- x_column.name(),
- x_column.data_type().to_string()
- );
- assert_eq!(column_data, "\"x\":\"Int64\"");
- let has_column = df_schema.has_column(&Column::from_name("z"));
- assert!(has_column);
- }
-
- #[tokio::test]
- async fn test_static_provider_rejects_writes() {
- let table = get_test_table_from_metadata_file().await;
- let table_provider =
- IcebergStaticTableProvider::try_new_from_table(table.clone())
- .await
- .unwrap();
- let ctx = SessionContext::new();
- ctx.register_table("mytable", Arc::new(table_provider))
- .unwrap();
-
- // Attempt to insert into the static provider should fail
- let result = ctx.sql("INSERT INTO mytable VALUES (1, 2, 3)").await;
-
- // The error should occur during planning or execution
- // We expect an error indicating write operations are not supported
- assert!(
- result.is_err() || {
- let df = result.unwrap();
- df.collect().await.is_err()
- }
- );
- }
-
- #[tokio::test]
- async fn test_static_provider_scan() {
- let table = get_test_table_from_metadata_file().await;
- let table_provider =
- IcebergStaticTableProvider::try_new_from_table(table.clone())
- .await
- .unwrap();
- let ctx = SessionContext::new();
- ctx.register_table("mytable", Arc::new(table_provider))
- .unwrap();
-
- // Test that scan operations work correctly
- let df = ctx.sql("SELECT count(*) FROM mytable").await.unwrap();
- let physical_plan = df.create_physical_plan().await;
- assert!(physical_plan.is_ok());
- }
-
- // Tests for IcebergTableProvider
-
#[tokio::test]
async fn test_catalog_backed_provider_creation() {
let (catalog, namespace, table_name, _temp_dir) =
get_test_catalog_and_table().await;
// Test creating a catalog-backed provider
- let provider = IcebergTableProvider::try_new(
+ let provider = IcebergCatalogTableProvider::try_new(
catalog.clone(),
namespace.clone(),
table_name.clone(),
@@ -581,7 +270,7 @@ mod tests {
let (catalog, namespace, table_name, _temp_dir) =
get_test_catalog_and_table().await;
- let provider = IcebergTableProvider::try_new(
+ let provider = IcebergCatalogTableProvider::try_new(
catalog.clone(),
namespace.clone(),
table_name.clone(),
@@ -611,7 +300,7 @@ mod tests {
let (catalog, namespace, table_name, _temp_dir) =
get_test_catalog_and_table().await;
- let provider = IcebergTableProvider::try_new(
+ let provider = IcebergCatalogTableProvider::try_new(
catalog.clone(),
namespace.clone(),
table_name.clone(),
@@ -642,7 +331,7 @@ mod tests {
let (catalog, namespace, table_name, _temp_dir) =
get_test_catalog_and_table().await;
- let provider = IcebergTableProvider::try_new(
+ let provider = IcebergCatalogTableProvider::try_new(
catalog.clone(),
namespace.clone(),
table_name.clone(),
@@ -680,66 +369,145 @@ mod tests {
}
}
- async fn get_partitioned_test_catalog_and_table(
- fanout_enabled: Option<bool>,
- ) -> (Arc<dyn Catalog>, NamespaceIdent, String, TempDir) {
- use iceberg::spec::{Transform, UnboundPartitionSpec};
-
- let temp_dir = TempDir::new().unwrap();
- let warehouse_path = temp_dir.path().to_str().unwrap().to_string();
+ #[tokio::test]
+ async fn test_catalog_backed_provider_rejects_non_append_op() {
+ use datafusion::physical_plan::empty::EmptyExec;
- let catalog = MemoryCatalogBuilder::default()
- .load(
- "memory",
- HashMap::from([(
- MEMORY_CATALOG_WAREHOUSE.to_string(),
- warehouse_path.clone(),
- )]),
- )
- .await
- .unwrap();
+ let (catalog, namespace, table_name, _temp_dir) =
+ get_test_catalog_and_table().await;
+ let provider =
+ IcebergCatalogTableProvider::try_new(catalog, namespace, table_name)
+ .await
+ .unwrap();
+ let ctx = SessionContext::new();
- let namespace = NamespaceIdent::new("test_ns".to_string());
- catalog
- .create_namespace(&namespace, HashMap::new())
- .await
- .unwrap();
+ for (insert_op, expected_message) in [
+ (
+ InsertOp::Overwrite,
+ "IcebergTableProvider supports only append inserts, got Insert Overwrite",
+ ),
+ (
+ InsertOp::Replace,
+ "IcebergTableProvider supports only append inserts, got Replace Into",
+ ),
+ ] {
+ let input =
+ Arc::new(EmptyExec::new(provider.schema())) as Arc<dyn ExecutionPlan>;
+ let error = provider
+ .insert_into(&ctx.state(), input, insert_op)
+ .await
+ .expect_err("non-append inserts should be rejected");
- let schema = Schema::builder()
- .with_schema_id(0)
- .with_fields(vec![
- NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
- .into(),
- NestedField::required(
- 2,
- "category",
- Type::Primitive(PrimitiveType::String),
- )
- .into(),
- ])
- .build()
- .unwrap();
+ assert!(
+ matches!(
+ error,
+ DataFusionError::NotImplemented(ref message) if message == expected_message
+ ),
+ "unexpected error: {error}"
+ );
+ }
+ }
- let partition_spec = UnboundPartitionSpec::builder()
- .with_spec_id(0)
- .add_partition_field(2, "category", Transform::Identity)
- .unwrap()
- .build();
+ #[tokio::test]
+ async fn test_limit_pushdown_catalog_backed_provider() {
+ use datafusion::datasource::TableProvider;
- let mut properties = HashMap::new();
- if let Some(enabled) = fanout_enabled {
- properties.insert(
- TableProperties::PROPERTY_DATAFUSION_WRITE_FANOUT_ENABLED.to_string(),
- enabled.to_string(),
- );
- }
+ let (catalog, namespace, table_name, _temp_dir) =
+ get_test_catalog_and_table().await;
+
+ let provider = IcebergCatalogTableProvider::try_new(
+ catalog.clone(),
+ namespace.clone(),
+ table_name.clone(),
+ )
+ .await
+ .unwrap();
+
+ let ctx = SessionContext::new();
+ let state = ctx.state();
+
+ // Test scan with limit
+ let scan_plan = provider.scan(&state, None, &[], Some(5)).await.unwrap();
+
+ // Verify that the scan plan is an IcebergTableScan
+ let iceberg_scan = scan_plan
+ .downcast_ref::<IcebergTableScan>()
+ .expect("Expected IcebergTableScan");
+
+ // Verify the limit is set
+ assert_eq!(
+ iceberg_scan.limit(),
+ Some(5),
+ "Limit should be set to 5 in the scan plan"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_scan_rejects_out_of_range_projection() {
+ let (catalog, namespace, table_name, _temp_dir) =
+ get_test_catalog_and_table().await;
+ let table_ident = TableIdent::new(namespace, table_name);
+ let table = catalog.load_table(&table_ident).await.unwrap();
+ let schema =
+ Arc::new(schema_to_arrow_schema(table.metadata().current_schema()).unwrap());
+ let out_of_range = schema.fields().len();
+
+ let err = IcebergTableScan::new(
+ table,
+ None,
+ schema,
+ Some(&vec![0, out_of_range]),
+ &[],
+ None,
+ )
+ .unwrap_err();
+ assert_eq!(
+ err.to_string(),
+ format!(
+ "Arrow error: Schema error: project index {out_of_range} out of \
+ bounds, max field {out_of_range}"
+ )
+ );
+ }
+
+ async fn get_test_catalog_and_table()
+ -> (Arc<dyn Catalog>, NamespaceIdent, String, TempDir) {
+ let temp_dir = TempDir::new().unwrap();
+ let warehouse_path = temp_dir.path().to_str().unwrap().to_string();
+
+ let catalog = MemoryCatalogBuilder::default()
+ .load(
+ "memory",
+ HashMap::from([(
+ MEMORY_CATALOG_WAREHOUSE.to_string(),
+ warehouse_path.clone(),
+ )]),
+ )
+ .await
+ .unwrap();
+
+ let namespace = NamespaceIdent::new("test_ns".to_string());
+ catalog
+ .create_namespace(&namespace, HashMap::new())
+ .await
+ .unwrap();
+
+ let schema = Schema::builder()
+ .with_schema_id(0)
+ .with_fields(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
+ .into(),
+ NestedField::required(2, "name", Type::Primitive(PrimitiveType::String))
+ .into(),
+ ])
+ .build()
+ .unwrap();
let table_creation = TableCreation::builder()
- .name("partitioned_table".to_string())
- .location(format!("{warehouse_path}/partitioned_table"))
+ .name("test_table".to_string())
+ .location(format!("{warehouse_path}/test_table"))
.schema(schema)
- .partition_spec(partition_spec)
- .properties(properties)
+ .properties(HashMap::new())
.build();
catalog
@@ -750,62 +518,11 @@ mod tests {
(
Arc::new(catalog),
namespace,
- "partitioned_table".to_string(),
+ "test_table".to_string(),
temp_dir,
)
}
- /// Helper to check if a plan contains a SortExec node
- fn plan_contains_sort(plan: &Arc<dyn ExecutionPlan>) -> bool {
- if plan.name() == "SortExec" {
- return true;
- }
- for child in plan.children() {
- if plan_contains_sort(child) {
- return true;
- }
- }
- false
- }
-
- #[tokio::test]
- async fn test_catalog_backed_provider_rejects_non_append_op() {
- use datafusion::physical_plan::empty::EmptyExec;
-
- let (catalog, namespace, table_name, _temp_dir) =
- get_test_catalog_and_table().await;
- let provider = IcebergTableProvider::try_new(catalog, namespace, table_name)
- .await
- .unwrap();
- let ctx = SessionContext::new();
-
- for (insert_op, expected_message) in [
- (
- InsertOp::Overwrite,
- "IcebergTableProvider supports only append inserts, got Insert Overwrite",
- ),
- (
- InsertOp::Replace,
- "IcebergTableProvider supports only append inserts, got Replace Into",
- ),
- ] {
- let input =
- Arc::new(EmptyExec::new(provider.schema())) as Arc<dyn ExecutionPlan>;
- let error = provider
- .insert_into(&ctx.state(), input, insert_op)
- .await
- .expect_err("non-append inserts should be rejected");
-
- assert!(
- matches!(
- error,
- DataFusionError::NotImplemented(ref message) if message == expected_message
- ),
- "unexpected error: {error}"
- );
- }
- }
-
#[tokio::test]
async fn test_insert_plan_fanout_enabled_no_sort() {
use datafusion::datasource::TableProvider;
@@ -816,7 +533,7 @@ mod tests {
let (catalog, namespace, table_name, _temp_dir) =
get_partitioned_test_catalog_and_table(Some(true)).await;
- let provider = IcebergTableProvider::try_new(
+ let provider = IcebergCatalogTableProvider::try_new(
catalog.clone(),
namespace.clone(),
table_name.clone(),
@@ -851,7 +568,7 @@ mod tests {
let (catalog, namespace, table_name, _temp_dir) =
get_partitioned_test_catalog_and_table(Some(false)).await;
- let provider = IcebergTableProvider::try_new(
+ let provider = IcebergCatalogTableProvider::try_new(
catalog.clone(),
namespace.clone(),
table_name.clone(),
@@ -876,125 +593,91 @@ mod tests {
);
}
- #[tokio::test]
- async fn test_limit_pushdown_static_provider() {
- use datafusion::datasource::TableProvider;
-
- let table = get_test_table_from_metadata_file().await;
- let table_provider =
- IcebergStaticTableProvider::try_new_from_table(table.clone())
- .await
- .unwrap();
+ async fn get_partitioned_test_catalog_and_table(
+ fanout_enabled: Option<bool>,
+ ) -> (Arc<dyn Catalog>, NamespaceIdent, String, TempDir) {
+ use iceberg::spec::{Transform, UnboundPartitionSpec};
- let ctx = SessionContext::new();
- let state = ctx.state();
+ let temp_dir = TempDir::new().unwrap();
+ let warehouse_path = temp_dir.path().to_str().unwrap().to_string();
- // Test scan with limit
- let scan_plan = table_provider
- .scan(&state, None, &[], Some(10))
+ let catalog = MemoryCatalogBuilder::default()
+ .load(
+ "memory",
+ HashMap::from([(
+ MEMORY_CATALOG_WAREHOUSE.to_string(),
+ warehouse_path.clone(),
+ )]),
+ )
.await
.unwrap();
- // Verify that the scan plan is an IcebergTableScan
- let iceberg_scan = scan_plan
- .downcast_ref::<IcebergTableScan>()
- .expect("Expected IcebergTableScan");
-
- // Verify the limit is set
- assert_eq!(
- iceberg_scan.limit(),
- Some(10),
- "Limit should be set to 10 in the scan plan"
- );
- }
-
- #[tokio::test]
- async fn test_limit_pushdown_catalog_backed_provider() {
- use datafusion::datasource::TableProvider;
-
- let (catalog, namespace, table_name, _temp_dir) =
- get_test_catalog_and_table().await;
-
- let provider = IcebergTableProvider::try_new(
- catalog.clone(),
- namespace.clone(),
- table_name.clone(),
- )
- .await
- .unwrap();
-
- let ctx = SessionContext::new();
- let state = ctx.state();
-
- // Test scan with limit
- let scan_plan = provider.scan(&state, None, &[], Some(5)).await.unwrap();
-
- // Verify that the scan plan is an IcebergTableScan
- let iceberg_scan = scan_plan
- .downcast_ref::<IcebergTableScan>()
- .expect("Expected IcebergTableScan");
-
- // Verify the limit is set
- assert_eq!(
- iceberg_scan.limit(),
- Some(5),
- "Limit should be set to 5 in the scan plan"
- );
- }
-
- #[tokio::test]
- async fn test_no_limit_pushdown() {
- use datafusion::datasource::TableProvider;
-
- let table = get_test_table_from_metadata_file().await;
- let table_provider =
- IcebergStaticTableProvider::try_new_from_table(table.clone())
- .await
- .unwrap();
+ let namespace = NamespaceIdent::new("test_ns".to_string());
+ catalog
+ .create_namespace(&namespace, HashMap::new())
+ .await
+ .unwrap();
- let ctx = SessionContext::new();
- let state = ctx.state();
+ let schema = Schema::builder()
+ .with_schema_id(0)
+ .with_fields(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int))
+ .into(),
+ NestedField::required(
+ 2,
+ "category",
+ Type::Primitive(PrimitiveType::String),
+ )
+ .into(),
+ ])
+ .build()
+ .unwrap();
- // Test scan without limit
- let scan_plan = table_provider.scan(&state, None, &[], None).await.unwrap();
+ let partition_spec = UnboundPartitionSpec::builder()
+ .with_spec_id(0)
+ .add_partition_field(2, "category", Transform::Identity)
+ .unwrap()
+ .build();
- // Verify that the scan plan is an IcebergTableScan
- let iceberg_scan = scan_plan
- .downcast_ref::<IcebergTableScan>()
- .expect("Expected IcebergTableScan");
+ let mut properties = HashMap::new();
+ if let Some(enabled) = fanout_enabled {
+ properties.insert(
+ TableProperties::PROPERTY_DATAFUSION_WRITE_FANOUT_ENABLED.to_string(),
+ enabled.to_string(),
+ );
+ }
- // Verify the limit is None
- assert_eq!(
- iceberg_scan.limit(),
- None,
- "Limit should be None when not specified"
- );
- }
+ let table_creation = TableCreation::builder()
+ .name("partitioned_table".to_string())
+ .location(format!("{warehouse_path}/partitioned_table"))
+ .schema(schema)
+ .partition_spec(partition_spec)
+ .properties(properties)
+ .build();
- #[tokio::test]
- async fn test_scan_rejects_out_of_range_projection() {
- let table = get_test_table_from_metadata_file().await;
- let provider = IcebergStaticTableProvider::try_new_from_table(table.clone())
+ catalog
+ .create_table(&namespace, table_creation)
.await
.unwrap();
- let schema = provider.schema();
- let out_of_range = schema.fields().len();
- let err = IcebergTableScan::new(
- table,
- None,
- schema,
- Some(&vec![0, out_of_range]),
- &[],
- None,
+ (
+ Arc::new(catalog),
+ namespace,
+ "partitioned_table".to_string(),
+ temp_dir,
)
- .unwrap_err();
- assert_eq!(
- err.to_string(),
- format!(
- "Arrow error: Schema error: project index {out_of_range} out of \
- bounds, max field {out_of_range}"
- )
- );
+ }
+
+ /// Helper to check if a plan contains a SortExec node.
+ fn plan_contains_sort(plan: &Arc<dyn ExecutionPlan>) -> bool {
+ if plan.name() == "SortExec" {
+ return true;
+ }
+ for child in plan.children() {
+ if plan_contains_sort(child) {
+ return true;
+ }
+ }
+ false
}
}
diff --git a/crates/datafusion/src/table/table_provider_factory.rs b/crates/datafusion/src/table_provider/factory.rs
similarity index 100%
rename from crates/datafusion/src/table/table_provider_factory.rs
rename to crates/datafusion/src/table_provider/factory.rs
diff --git a/crates/datafusion/src/lib.rs b/crates/datafusion/src/table_provider/mod.rs
similarity index 65%
copy from crates/datafusion/src/lib.rs
copy to crates/datafusion/src/table_provider/mod.rs
index 4b0ea86..d3ee9f7 100644
--- a/crates/datafusion/src/lib.rs
+++ b/crates/datafusion/src/table_provider/mod.rs
@@ -15,16 +15,15 @@
// specific language governing permissions and limitations
// under the License.
-mod catalog;
-pub use catalog::*;
+//! Iceberg table providers for DataFusion.
+//!
+//! The catalog-backed provider refreshes table metadata for reads and writes.
+//! The static provider reads a fixed table snapshot.
-mod error;
-pub use error::*;
+mod catalog_table_provider;
+mod factory;
+mod static_table_provider;
-pub mod physical_plan;
-mod schema;
-pub mod table;
-pub use table::table_provider_factory::IcebergTableProviderFactory;
-pub use table::*;
-
-pub(crate) mod task_writer;
+pub use catalog_table_provider::IcebergCatalogTableProvider;
+pub use factory::IcebergTableProviderFactory;
+pub use static_table_provider::IcebergStaticTableProvider;
diff --git a/crates/datafusion/src/table_provider/static_table_provider.rs b/crates/datafusion/src/table_provider/static_table_provider.rs
new file mode 100644
index 0000000..4a2f48f
--- /dev/null
+++ b/crates/datafusion/src/table_provider/static_table_provider.rs
@@ -0,0 +1,345 @@
+// 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 async_trait::async_trait;
+use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
+use datafusion::catalog::Session;
+use datafusion::common::{exec_datafusion_err, not_impl_err};
+use datafusion::datasource::{TableProvider, TableType};
+use datafusion::error::Result;
+use datafusion::logical_expr::dml::InsertOp;
+use datafusion::logical_expr::{Expr, TableProviderFilterPushDown};
+use datafusion::physical_plan::ExecutionPlan;
+use iceberg::arrow::schema_to_arrow_schema;
+use iceberg::table::Table;
+
+use crate::error::to_datafusion_error;
+use crate::read::scan::IcebergTableScan;
+
+/// Static table provider for read-only snapshot access.
+///
+/// This provider holds a cached table instance and does not refresh metadata or support
+/// write operations. Use this for consistent analytical queries, time-travel scenarios,
+/// or when you want to avoid catalog overhead.
+///
+/// For catalog-backed tables with write support and automatic refresh, use
+/// [`IcebergTableProvider`](super::IcebergCatalogTableProvider) instead.
+#[derive(Debug, Clone)]
+pub struct IcebergStaticTableProvider {
+ /// The static table instance (never refreshed)
+ table: Table,
+ /// Optional snapshot ID for this static view
+ snapshot_id: Option<i64>,
+ /// A reference-counted arrow `Schema`
+ schema: ArrowSchemaRef,
+}
+
+impl IcebergStaticTableProvider {
+ /// Creates a static provider from a table instance.
+ ///
+ /// Uses the table's current snapshot for all queries. Does not support write operations.
+ pub async fn try_new_from_table(table: Table) -> Result<Self> {
+ let schema = Arc::new(
+ schema_to_arrow_schema(table.metadata().current_schema())
+ .map_err(to_datafusion_error)?,
+ );
+ Ok(IcebergStaticTableProvider {
+ table,
+ snapshot_id: None,
+ schema,
+ })
+ }
+
+ /// Creates a static provider for a specific table snapshot.
+ ///
+ /// Queries the specified snapshot for all operations. Useful for time-travel queries.
+ /// Does not support write operations.
+ pub async fn try_new_from_table_snapshot(
+ table: Table,
+ snapshot_id: i64,
+ ) -> Result<Self> {
+ let snapshot = table
+ .metadata()
+ .snapshot_by_id(snapshot_id)
+ .ok_or_else(|| {
+ exec_datafusion_err!(
+ "snapshot id {snapshot_id} not found in table {}",
+ table.identifier().name()
+ )
+ })?;
+ let table_schema = snapshot
+ .schema(table.metadata())
+ .map_err(to_datafusion_error)?;
+ let schema =
+ Arc::new(schema_to_arrow_schema(&table_schema).map_err(to_datafusion_error)?);
+ Ok(IcebergStaticTableProvider {
+ table,
+ snapshot_id: Some(snapshot_id),
+ schema,
+ })
+ }
+
+ /// The table as loaded when this provider was built.
+ pub fn table(&self) -> &Table {
+ &self.table
+ }
+
+ /// The snapshot this provider reads, or `None` for the current snapshot of
+ /// [`Self::table`].
+ pub fn snapshot_id(&self) -> Option<i64> {
+ self.snapshot_id
+ }
+}
+
+#[async_trait]
+impl TableProvider for IcebergStaticTableProvider {
+ fn schema(&self) -> ArrowSchemaRef {
+ self.schema.clone()
+ }
+
+ fn table_type(&self) -> TableType {
+ TableType::Base
+ }
+
+ async fn scan(
+ &self,
+ _state: &dyn Session,
+ projection: Option<&Vec<usize>>,
+ filters: &[Expr],
+ limit: Option<usize>,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ // Use cached table (no refresh)
+ Ok(Arc::new(IcebergTableScan::new(
+ self.table.clone(),
+ self.snapshot_id,
+ self.schema.clone(),
+ projection,
+ filters,
+ limit,
+ )?))
+ }
+
+ fn supports_filters_pushdown(
+ &self,
+ filters: &[&Expr],
+ ) -> Result<Vec<TableProviderFilterPushDown>> {
+ // Push down all filters, as a single source of truth, the scanner will drop the filters which couldn't be push down
+ Ok(vec![TableProviderFilterPushDown::Inexact; filters.len()])
+ }
+
+ async fn insert_into(
+ &self,
+ _state: &dyn Session,
+ _input: Arc<dyn ExecutionPlan>,
+ _insert_op: InsertOp,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ not_impl_err!(
+ "Write operations are not supported on IcebergStaticTableProvider. \
+ Use IcebergTableProvider with a catalog for write support."
+ )
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::sync::Arc;
+
+ use datafusion::common::Column;
+ use datafusion::prelude::SessionContext;
+ use iceberg::TableIdent;
+ use iceberg::io::FileIO;
+ use iceberg::table::{StaticTable, Table};
+
+ use super::*;
+
+ #[tokio::test]
+ async fn test_static_provider_from_table() {
+ let table = get_test_table_from_metadata_file().await;
+ let table_provider =
+ IcebergStaticTableProvider::try_new_from_table(table.clone())
+ .await
+ .unwrap();
+ let ctx = SessionContext::new();
+ ctx.register_table("mytable", Arc::new(table_provider))
+ .unwrap();
+ let df = ctx.sql("SELECT * FROM mytable").await.unwrap();
+ let df_schema = df.schema();
+ let df_columns = df_schema.fields();
+ assert_eq!(df_columns.len(), 3);
+ let x_column = df_columns.first().unwrap();
+ let column_data = format!(
+ "{:?}:{:?}",
+ x_column.name(),
+ x_column.data_type().to_string()
+ );
+ assert_eq!(column_data, "\"x\":\"Int64\"");
+ let has_column = df_schema.has_column(&Column::from_name("z"));
+ assert!(has_column);
+ }
+
+ #[tokio::test]
+ async fn test_static_provider_from_snapshot() {
+ let table = get_test_table_from_metadata_file().await;
+ let snapshot_id = table.metadata().snapshots().next().unwrap().snapshot_id();
+ let table_provider = IcebergStaticTableProvider::try_new_from_table_snapshot(
+ table.clone(),
+ snapshot_id,
+ )
+ .await
+ .unwrap();
+ let ctx = SessionContext::new();
+ ctx.register_table("mytable", Arc::new(table_provider))
+ .unwrap();
+ let df = ctx.sql("SELECT * FROM mytable").await.unwrap();
+ let df_schema = df.schema();
+ let df_columns = df_schema.fields();
+ assert_eq!(df_columns.len(), 3);
+ let x_column = df_columns.first().unwrap();
+ let column_data = format!(
+ "{:?}:{:?}",
+ x_column.name(),
+ x_column.data_type().to_string()
+ );
+ assert_eq!(column_data, "\"x\":\"Int64\"");
+ let has_column = df_schema.has_column(&Column::from_name("z"));
+ assert!(has_column);
+ }
+
+ #[tokio::test]
+ async fn test_static_provider_rejects_writes() {
+ let table = get_test_table_from_metadata_file().await;
+ let table_provider =
+ IcebergStaticTableProvider::try_new_from_table(table.clone())
+ .await
+ .unwrap();
+ let ctx = SessionContext::new();
+ ctx.register_table("mytable", Arc::new(table_provider))
+ .unwrap();
+
+ // Attempt to insert into the static provider should fail
+ let result = ctx.sql("INSERT INTO mytable VALUES (1, 2, 3)").await;
+
+ // The error should occur during planning or execution
+ // We expect an error indicating write operations are not supported
+ assert!(
+ result.is_err() || {
+ let df = result.unwrap();
+ df.collect().await.is_err()
+ }
+ );
+ }
+
+ #[tokio::test]
+ async fn test_static_provider_scan() {
+ let table = get_test_table_from_metadata_file().await;
+ let table_provider =
+ IcebergStaticTableProvider::try_new_from_table(table.clone())
+ .await
+ .unwrap();
+ let ctx = SessionContext::new();
+ ctx.register_table("mytable", Arc::new(table_provider))
+ .unwrap();
+
+ // Test that scan operations work correctly
+ let df = ctx.sql("SELECT count(*) FROM mytable").await.unwrap();
+ let physical_plan = df.create_physical_plan().await;
+ assert!(physical_plan.is_ok());
+ }
+
+ #[tokio::test]
+ async fn test_limit_pushdown_static_provider() {
+ use datafusion::datasource::TableProvider;
+
+ let table = get_test_table_from_metadata_file().await;
+ let table_provider =
+ IcebergStaticTableProvider::try_new_from_table(table.clone())
+ .await
+ .unwrap();
+
+ let ctx = SessionContext::new();
+ let state = ctx.state();
+
+ // Test scan with limit
+ let scan_plan = table_provider
+ .scan(&state, None, &[], Some(10))
+ .await
+ .unwrap();
+
+ // Verify that the scan plan is an IcebergTableScan
+ let iceberg_scan = scan_plan
+ .downcast_ref::<IcebergTableScan>()
+ .expect("Expected IcebergTableScan");
+
+ // Verify the limit is set
+ assert_eq!(
+ iceberg_scan.limit(),
+ Some(10),
+ "Limit should be set to 10 in the scan plan"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_no_limit_pushdown() {
+ use datafusion::datasource::TableProvider;
+
+ let table = get_test_table_from_metadata_file().await;
+ let table_provider =
+ IcebergStaticTableProvider::try_new_from_table(table.clone())
+ .await
+ .unwrap();
+
+ let ctx = SessionContext::new();
+ let state = ctx.state();
+
+ // Test scan without limit
+ let scan_plan = table_provider.scan(&state, None, &[], None).await.unwrap();
+
+ // Verify that the scan plan is an IcebergTableScan
+ let iceberg_scan = scan_plan
+ .downcast_ref::<IcebergTableScan>()
+ .expect("Expected IcebergTableScan");
+
+ // Verify the limit is None
+ assert_eq!(
+ iceberg_scan.limit(),
+ None,
+ "Limit should be None when not specified"
+ );
+ }
+
+ async fn get_test_table_from_metadata_file() -> Table {
+ let metadata_file_name = "TableMetadataV2Valid.json";
+ let metadata_file_path = format!(
+ "{}/tests/test_data/{}",
+ env!("CARGO_MANIFEST_DIR"),
+ metadata_file_name
+ );
+ let file_io = FileIO::new_with_fs();
+ let static_identifier =
+ TableIdent::from_strs(["static_ns", "static_table"]).unwrap();
+ let static_table = StaticTable::from_metadata_file(
+ &metadata_file_path,
+ static_identifier,
+ file_io,
+ )
+ .await
+ .unwrap();
+ static_table.into_table()
+ }
+}
diff --git a/crates/datafusion/src/physical_plan/commit.rs b/crates/datafusion/src/write/commit.rs
similarity index 99%
rename from crates/datafusion/src/physical_plan/commit.rs
rename to crates/datafusion/src/write/commit.rs
index 468bf5a..a9f99e8 100644
--- a/crates/datafusion/src/physical_plan/commit.rs
+++ b/crates/datafusion/src/write/commit.rs
@@ -39,7 +39,7 @@ use iceberg::spec::{DataFile, deserialize_data_file_from_json};
use iceberg::table::Table;
use iceberg::transaction::{ApplyTransactionAction, Transaction};
-use crate::physical_plan::DATA_FILES_COL_NAME;
+use super::exec::DATA_FILES_COL_NAME;
use crate::to_datafusion_error;
/// IcebergCommitExec is responsible for collecting the files written and use
@@ -348,8 +348,7 @@ mod tests {
use iceberg::{Catalog, CatalogBuilder, NamespaceIdent, TableCreation, TableIdent};
use super::*;
- use crate::physical_plan::DATA_FILES_COL_NAME;
- use crate::table::IcebergTableProvider;
+ use crate::table_provider::IcebergCatalogTableProvider;
// A mock execution plan that returns record batches with serialized data files
#[derive(Debug)]
@@ -811,7 +810,7 @@ mod tests {
Arc::new(MemTable::try_new(Arc::clone(&arrow_schema), partitions)?);
ctx.register_table("source_table", source_table)?;
- let iceberg_table_provider = IcebergTableProvider::try_new(
+ let iceberg_table_provider = IcebergCatalogTableProvider::try_new(
catalog.clone(),
namespace.clone(),
table_name.to_string(),
diff --git a/crates/datafusion/src/physical_plan/write.rs b/crates/datafusion/src/write/exec.rs
similarity index 99%
rename from crates/datafusion/src/physical_plan/write.rs
rename to crates/datafusion/src/write/exec.rs
index 86f364c..82b7867 100644
--- a/crates/datafusion/src/physical_plan/write.rs
+++ b/crates/datafusion/src/write/exec.rs
@@ -46,10 +46,11 @@ use iceberg::writer::file_writer::location_generator::{
use iceberg::writer::file_writer::rolling_writer::RollingFileWriterBuilder;
use uuid::Uuid;
-use crate::physical_plan::DATA_FILES_COL_NAME;
-use crate::task_writer::TaskWriter;
+use super::task_writer::TaskWriter;
use crate::to_datafusion_error;
+pub(crate) const DATA_FILES_COL_NAME: &str = "data_files";
+
/// An execution plan node that writes data to an Iceberg table.
///
/// This execution plan takes input data from a child execution plan and writes it to an Iceberg table.
diff --git a/crates/datafusion/src/physical_plan/mod.rs b/crates/datafusion/src/write/mod.rs
similarity index 74%
rename from crates/datafusion/src/physical_plan/mod.rs
rename to crates/datafusion/src/write/mod.rs
index 024266a..7c57a16 100644
--- a/crates/datafusion/src/physical_plan/mod.rs
+++ b/crates/datafusion/src/write/mod.rs
@@ -16,19 +16,12 @@
// under the License.
pub(crate) mod commit;
-pub(crate) mod expr_to_predicate;
-pub(crate) mod metadata_scan;
+pub(crate) mod exec;
pub(crate) mod project;
pub(crate) mod repartition;
-pub(crate) mod scan;
pub(crate) mod sort;
-pub(crate) mod write;
-
-pub(crate) const DATA_FILES_COL_NAME: &str = "data_files";
+pub(crate) mod task_writer;
pub use commit::IcebergCommitExec;
-pub use expr_to_predicate::convert_filters_to_predicate;
-pub use metadata_scan::IcebergMetadataScan;
+pub use exec::IcebergWriteExec;
pub use project::{PartitionExpr, project_with_partition};
-pub use scan::IcebergTableScan;
-pub use write::IcebergWriteExec;
diff --git a/crates/datafusion/src/physical_plan/project.rs b/crates/datafusion/src/write/project.rs
similarity index 99%
rename from crates/datafusion/src/physical_plan/project.rs
rename to crates/datafusion/src/write/project.rs
index 9e335c8..f63553d 100644
--- a/crates/datafusion/src/physical_plan/project.rs
+++ b/crates/datafusion/src/write/project.rs
@@ -124,7 +124,7 @@ impl PartitionExpr {
/// ```
/// use std::sync::Arc;
///
- /// use datafusion_iceberg::physical_plan::PartitionExpr;
+ /// use datafusion_iceberg::write::PartitionExpr;
/// use iceberg::spec::{NestedField, PartitionSpec, PrimitiveType, Schema, Transform, Type};
///
/// let schema = Arc::new(
diff --git a/crates/datafusion/src/physical_plan/repartition.rs b/crates/datafusion/src/write/repartition.rs
similarity index 99%
rename from crates/datafusion/src/physical_plan/repartition.rs
rename to crates/datafusion/src/write/repartition.rs
index 0dbda0c..814fef7 100644
--- a/crates/datafusion/src/physical_plan/repartition.rs
+++ b/crates/datafusion/src/write/repartition.rs
@@ -46,7 +46,7 @@ use iceberg::spec::{TableMetadata, TableMetadataRef, Transform};
/// ## Requirements
///
/// - **For partitioned tables**: The input MUST include the `_partition` column.
-/// Add it by calling [`project_with_partition`](crate::physical_plan::project_with_partition) before [`repartition`].
+/// Add it by calling [`project_with_partition`](crate::write::project_with_partition) before [`repartition`].
/// - **For unpartitioned tables**: No special preparation needed.
/// - Returns an error if a partitioned table is missing the `_partition` column.
///
@@ -58,7 +58,7 @@ use iceberg::spec::{TableMetadata, TableMetadataRef, Transform};
/// # Arguments
///
/// * `input` - The input [`ExecutionPlan`]. For partitioned tables, must include the `_partition`
-/// column (added via [`project_with_partition`](crate::physical_plan::project_with_partition)).
+/// column (added via [`project_with_partition`](crate::write::project_with_partition)).
/// * `table_metadata` - Iceberg table metadata containing partition spec.
/// * `target_partitions` - Target number of partitions for parallel processing (must be > 0).
///
@@ -77,7 +77,7 @@ use iceberg::spec::{TableMetadata, TableMetadataRef, Transform};
///
/// ```ignore
/// use std::num::NonZeroUsize;
-/// use datafusion_iceberg::physical_plan::project_with_partition;
+/// use datafusion_iceberg::write::project_with_partition;
///
/// let plan_with_partition = project_with_partition(input_plan, &table)?;
///
diff --git a/crates/datafusion/src/physical_plan/sort.rs b/crates/datafusion/src/write/sort.rs
similarity index 100%
rename from crates/datafusion/src/physical_plan/sort.rs
rename to crates/datafusion/src/write/sort.rs
diff --git a/crates/datafusion/src/task_writer.rs b/crates/datafusion/src/write/task_writer.rs
similarity index 100%
rename from crates/datafusion/src/task_writer.rs
rename to crates/datafusion/src/write/task_writer.rs
diff --git a/crates/datafusion/tests/integration_datafusion_test.rs b/crates/datafusion/tests/integration_datafusion_test.rs
index 76e379c..0c222c7 100644
--- a/crates/datafusion/tests/integration_datafusion_test.rs
+++ b/crates/datafusion/tests/integration_datafusion_test.rs
@@ -34,12 +34,11 @@ use datafusion::parquet::arrow::PARQUET_FIELD_ID_META_KEY;
use datafusion::physical_plan::ExecutionPlan;
use datafusion::physical_plan::coalesce_partitions::CoalescePartitionsExec;
use datafusion::physical_plan::common::collect;
-use datafusion_iceberg::physical_plan::{
- IcebergCommitExec, IcebergMetadataScan, IcebergTableScan, IcebergWriteExec,
-};
+use datafusion_iceberg::metadata::{IcebergMetadataScan, IcebergMetadataTableProvider};
+use datafusion_iceberg::read::IcebergTableScan;
+use datafusion_iceberg::write::{IcebergCommitExec, IcebergWriteExec};
use datafusion_iceberg::{
- IcebergCatalogProvider, IcebergMetadataTableProvider, IcebergStaticTableProvider,
- IcebergTableProvider,
+ IcebergCatalogProvider, IcebergCatalogTableProvider, IcebergStaticTableProvider,
};
use expect_test::expect;
use iceberg::io::LocalFsStorageFactory;
@@ -1048,11 +1047,11 @@ async fn test_plan_nodes_are_inspectable() -> Result<(), Box<dyn Error>> {
.table_provider("catalog.test_plan_nodes.my_table")
.await?;
let provider = provider
- .downcast_ref::<IcebergTableProvider>()
+ .downcast_ref::<IcebergCatalogTableProvider>()
.expect("a catalog-backed provider");
assert_eq!(provider.table_ident(), &ident);
assert!(Arc::ptr_eq(provider.catalog(), &client));
- let rebuilt = IcebergTableProvider::try_new(
+ let rebuilt = IcebergCatalogTableProvider::try_new(
provider.catalog().clone(),
provider.table_ident().namespace().clone(),
provider.table_ident().name(),
@@ -1300,7 +1299,8 @@ async fn test_scan_after_schema_evolution_reads_provider_columns()
let client: Arc<dyn Catalog> = Arc::new(iceberg_catalog);
let provider = Arc::new(
- IcebergTableProvider::try_new(client.clone(), namespace, "my_table").await?,
+ IcebergCatalogTableProvider::try_new(client.clone(), namespace, "my_table")
+ .await?,
);
let ctx = SessionContext::new();
ctx.register_table("t", provider.clone())?;
@@ -1322,7 +1322,7 @@ async fn test_scan_after_schema_evolution_reads_provider_columns()
// A scan reads the schema of the snapshot it reads, so write a snapshot
// with the new column, through a provider that sees it.
- let evolved = IcebergTableProvider::try_new(
+ let evolved = IcebergCatalogTableProvider::try_new(
client.clone(),
ident.namespace().clone(),
ident.name(),
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org