gabotechs commented on code in PR #20:
URL: https://github.com/apache/datafusion-iceberg/pull/20#discussion_r4203497861
##########
crates/datafusion/tests/integration_datafusion_test.rs:
##########
@@ -977,3 +992,395 @@ async fn test_insert_into_partitioned() -> Result<(),
Box<dyn Error>> {
Ok(())
}
+
+/// Executes `plan`, which must have a single partition, and returns its rows.
+async fn run_batches(
+ plan: &dyn ExecutionPlan,
+ ctx: &SessionContext,
+) -> Result<Vec<RecordBatch>, Box<dyn Error>> {
+ assert_eq!(plan.properties().partitioning.partition_count(), 1);
+ let stream = plan.execute(0, ctx.task_ctx())?;
+ Ok(collect(stream).await?)
+}
+
+/// Executes `plan`, which must have a single partition, and renders its rows
+/// as a table.
+async fn run(
+ plan: &dyn ExecutionPlan,
+ ctx: &SessionContext,
+) -> Result<String, Box<dyn Error>> {
+ Ok(pretty_format_batches(&run_batches(plan, ctx).await?)?.to_string())
+}
+
+/// Rebuilds `scan` from its accessors alone, as a codec would.
+fn rebuild_scan(scan: &IcebergTableScan) -> IcebergTableScan {
+ IcebergTableScan::new_with_predicate(
+ scan.table().clone(),
+ scan.snapshot_id(),
+ scan.schema(),
+ scan.predicates().cloned(),
+ scan.limit(),
+ )
+}
+
+/// 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>()
+ .or_else(|| plan.children().into_iter().find_map(find_node::<T>))
+}
+
+/// The plan nodes and providers can be named and inspected from outside this
+/// crate, and rebuilt from their parts, as a codec that serializes them does.
+#[tokio::test]
+async fn test_plan_nodes_are_inspectable() -> Result<(), Box<dyn Error>> {
Review Comment:
This test seems unreasonably long and verbose, it can easily become a
maintainability hazard. Some improvement points that I see:
1. It claims to be an integration test, but it does not interact with the
code as a user would, it deals with internals by building execution plans
manually rather than just issuing sql statements.
2. It's assertions will be hard to maintain, like asserting pointer
equality, raw `assert!`s over ascii graphics, etc...
3. It's hard to read, as it's very long and interacts with the public API of
the project in a convoluted and unrealistic way.
It has the traits of a purely AI generated test, I'd recommend to revisit
this one in a follow up PR to make it a bit more manageable for the people that
will come after. I'm trying to make this contributions a bit more manageable
and straightforward in https://github.com/apache/datafusion-iceberg/pull/18,
let me know if using the test framework introduced there could help.
##########
crates/datafusion/src/physical_plan/scan.rs:
##########
@@ -64,14 +64,110 @@ impl IcebergTableScan {
projection: Option<&Vec<usize>>,
filters: &[Expr],
limit: Option<usize>,
- ) -> Self {
+ ) -> Result<Self> {
Review Comment:
If this method is fallible now, it should be called `try_new()` instead,
following Rust conventions.
##########
crates/datafusion/tests/integration_datafusion_test.rs:
##########
@@ -977,3 +992,395 @@ async fn test_insert_into_partitioned() -> Result<(),
Box<dyn Error>> {
Ok(())
}
+
+/// Executes `plan`, which must have a single partition, and returns its rows.
+async fn run_batches(
+ plan: &dyn ExecutionPlan,
+ ctx: &SessionContext,
+) -> Result<Vec<RecordBatch>, Box<dyn Error>> {
+ assert_eq!(plan.properties().partitioning.partition_count(), 1);
+ let stream = plan.execute(0, ctx.task_ctx())?;
+ Ok(collect(stream).await?)
+}
+
+/// Executes `plan`, which must have a single partition, and renders its rows
+/// as a table.
+async fn run(
+ plan: &dyn ExecutionPlan,
+ ctx: &SessionContext,
+) -> Result<String, Box<dyn Error>> {
+ Ok(pretty_format_batches(&run_batches(plan, ctx).await?)?.to_string())
+}
+
+/// Rebuilds `scan` from its accessors alone, as a codec would.
+fn rebuild_scan(scan: &IcebergTableScan) -> IcebergTableScan {
+ IcebergTableScan::new_with_predicate(
+ scan.table().clone(),
+ scan.snapshot_id(),
+ scan.schema(),
+ scan.predicates().cloned(),
+ scan.limit(),
+ )
+}
+
+/// 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>()
+ .or_else(|| plan.children().into_iter().find_map(find_node::<T>))
+}
+
+/// The plan nodes and providers can be named and inspected from outside this
+/// crate, and rebuilt from their parts, as a codec that serializes them does.
+#[tokio::test]
+async fn test_plan_nodes_are_inspectable() -> Result<(), Box<dyn Error>> {
+ let iceberg_catalog = get_iceberg_catalog().await;
+ let namespace = NamespaceIdent::new("test_plan_nodes".to_string());
+ set_test_namespace(&iceberg_catalog, &namespace).await?;
+ let creation = get_table_creation(temp_path(), "my_table", None)?;
+ iceberg_catalog.create_table(&namespace, creation).await?;
+ let ident = TableIdent::new(namespace.clone(), "my_table".to_string());
+ let client: Arc<dyn Catalog> = Arc::new(iceberg_catalog);
+
+ let ctx = SessionContext::new();
+ let catalog = IcebergCatalogProvider::try_new(client.clone()).await?;
+ ctx.register_catalog("catalog", Arc::new(catalog));
+ let provider = ctx
+ .table_provider("catalog.test_plan_nodes.my_table")
+ .await?;
+ let provider = provider
+ .downcast_ref::<IcebergTableProvider>()
+ .expect("a catalog-backed provider");
+ assert_eq!(provider.table_ident(), &ident);
+ assert!(Arc::ptr_eq(provider.catalog(), &client));
+ let rebuilt = IcebergTableProvider::try_new(
+ provider.catalog().clone(),
+ provider.table_ident().namespace().clone(),
+ provider.table_ident().name(),
+ )
+ .await?;
+ assert_eq!(rebuilt.table_ident(), &ident);
+ assert_eq!(rebuilt.schema(), provider.schema());
+
+ // Write path: a commit above a write, both holding the table, and the
+ // commit going through the provider's catalog. The plan that runs is
+ // rebuilt from their accessors and children alone. The optimizer drops
+ // the coalesce above a single-partition write, so the rebuilt commit
+ // always gets one, as a codec would.
+ let insert = ctx
+ .sql("INSERT INTO catalog.test_plan_nodes.my_table VALUES (1, 'alan'),
(2, 'turing')")
+ .await?
+ .create_physical_plan()
+ .await?;
+ let commit = insert
+ .downcast_ref::<IcebergCommitExec>()
+ .expect("the insert plan is rooted at a commit");
+ assert_eq!(commit.table().identifier(), &ident);
+ assert!(Arc::ptr_eq(commit.catalog(), &client));
+ let write = find_node::<IcebergWriteExec>(&insert).expect("a write below
the commit");
+ assert_eq!(write.table().identifier(), &ident);
+ let rebuilt_write: Arc<dyn ExecutionPlan> = Arc::new(IcebergWriteExec::new(
+ write.table().clone(),
+ write.children()[0].clone(),
+ ));
+ let rebuilt_commit = IcebergCommitExec::new(
+ commit.table().clone(),
+ commit.catalog().clone(),
+ Arc::new(CoalescePartitionsExec::new(rebuilt_write)),
+ );
+ expect![[r#"
+ +-------+
+ | count |
+ +-------+
+ | 2 |
+ +-------+"#]]
+ .assert_eq(&run(&rebuilt_commit, &ctx).await?);
+
+ // Read path: a scan pinned to a snapshot, rebuilt from its accessors,
+ // returns the same rows.
+ let table = client.load_table(&ident).await?;
+ let snapshot_id = table.metadata().current_snapshot_id().unwrap();
+ let pinned =
+ IcebergStaticTableProvider::try_new_from_table_snapshot(table,
snapshot_id)
+ .await?;
+ assert_eq!(pinned.snapshot_id(), Some(snapshot_id));
+ ctx.register_table("pinned", Arc::new(pinned.clone()))?;
+ // A later write, so a scan reading the current snapshot rather than the
+ // pinned one would return its row too.
+ ctx.sql("INSERT INTO catalog.test_plan_nodes.my_table VALUES (3,
'hopper')")
+ .await?
+ .collect()
+ .await?;
+ let latest_snapshot_id = client
+ .load_table(&ident)
+ .await?
+ .metadata()
+ .current_snapshot_id()
+ .unwrap();
+ assert_ne!(latest_snapshot_id, snapshot_id);
+ let plan = ctx
+ .sql("SELECT foo2 FROM pinned WHERE foo1 = 1")
+ .await?
+ .create_physical_plan()
+ .await?;
+ let scan = find_node::<IcebergTableScan>(&plan).expect("a scan");
+ assert_eq!(
+ scan.predicates().map(ToString::to_string).as_deref(),
+ Some("foo1 = 1")
+ );
+ let rebuilt = rebuild_scan(scan);
+ assert_eq!(rebuilt.schema(), scan.schema());
+ assert_eq!(rebuilt.projection(), scan.projection());
+ let expected = run(scan, &ctx).await?;
+ expect![[r#"
+ +------+------+
+ | foo1 | foo2 |
+ +------+------+
+ | 1 | alan |
+ +------+------+"#]]
+ .assert_eq(&expected);
+ assert_eq!(run(&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");
+ 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?;
+ expect![[r#"
+ +------+------+
+ | foo1 | foo2 |
+ +------+------+
+ | 1 | alan |
+ +------+------+"#]]
+ .assert_eq(&expected);
Review Comment:
We should not be using raw assertions over ascii graphics here, if these
changes it will be very hard to maintain. I'd recommend to rework this using
`insta` for managing snapshot assertion automatically.
##########
crates/datafusion/tests/integration_datafusion_test.rs:
##########
@@ -977,3 +992,395 @@ async fn test_insert_into_partitioned() -> Result<(),
Box<dyn Error>> {
Ok(())
}
+
+/// Executes `plan`, which must have a single partition, and returns its rows.
+async fn run_batches(
+ plan: &dyn ExecutionPlan,
+ ctx: &SessionContext,
+) -> Result<Vec<RecordBatch>, Box<dyn Error>> {
+ assert_eq!(plan.properties().partitioning.partition_count(), 1);
+ let stream = plan.execute(0, ctx.task_ctx())?;
+ Ok(collect(stream).await?)
+}
+
+/// Executes `plan`, which must have a single partition, and renders its rows
+/// as a table.
+async fn run(
+ plan: &dyn ExecutionPlan,
+ ctx: &SessionContext,
+) -> Result<String, Box<dyn Error>> {
+ Ok(pretty_format_batches(&run_batches(plan, ctx).await?)?.to_string())
+}
+
+/// Rebuilds `scan` from its accessors alone, as a codec would.
+fn rebuild_scan(scan: &IcebergTableScan) -> IcebergTableScan {
+ IcebergTableScan::new_with_predicate(
+ scan.table().clone(),
+ scan.snapshot_id(),
+ scan.schema(),
+ scan.predicates().cloned(),
+ scan.limit(),
+ )
+}
+
+/// 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>()
+ .or_else(|| plan.children().into_iter().find_map(find_node::<T>))
+}
+
+/// The plan nodes and providers can be named and inspected from outside this
+/// crate, and rebuilt from their parts, as a codec that serializes them does.
+#[tokio::test]
+async fn test_plan_nodes_are_inspectable() -> Result<(), Box<dyn Error>> {
+ let iceberg_catalog = get_iceberg_catalog().await;
+ let namespace = NamespaceIdent::new("test_plan_nodes".to_string());
+ set_test_namespace(&iceberg_catalog, &namespace).await?;
+ let creation = get_table_creation(temp_path(), "my_table", None)?;
+ iceberg_catalog.create_table(&namespace, creation).await?;
+ let ident = TableIdent::new(namespace.clone(), "my_table".to_string());
+ let client: Arc<dyn Catalog> = Arc::new(iceberg_catalog);
+
+ let ctx = SessionContext::new();
+ let catalog = IcebergCatalogProvider::try_new(client.clone()).await?;
+ ctx.register_catalog("catalog", Arc::new(catalog));
+ let provider = ctx
+ .table_provider("catalog.test_plan_nodes.my_table")
+ .await?;
+ let provider = provider
+ .downcast_ref::<IcebergTableProvider>()
+ .expect("a catalog-backed provider");
+ assert_eq!(provider.table_ident(), &ident);
+ assert!(Arc::ptr_eq(provider.catalog(), &client));
+ let rebuilt = IcebergTableProvider::try_new(
+ provider.catalog().clone(),
+ provider.table_ident().namespace().clone(),
+ provider.table_ident().name(),
+ )
+ .await?;
+ assert_eq!(rebuilt.table_ident(), &ident);
+ assert_eq!(rebuilt.schema(), provider.schema());
+
+ // Write path: a commit above a write, both holding the table, and the
+ // commit going through the provider's catalog. The plan that runs is
+ // rebuilt from their accessors and children alone. The optimizer drops
+ // the coalesce above a single-partition write, so the rebuilt commit
+ // always gets one, as a codec would.
+ let insert = ctx
+ .sql("INSERT INTO catalog.test_plan_nodes.my_table VALUES (1, 'alan'),
(2, 'turing')")
+ .await?
+ .create_physical_plan()
+ .await?;
+ let commit = insert
+ .downcast_ref::<IcebergCommitExec>()
+ .expect("the insert plan is rooted at a commit");
+ assert_eq!(commit.table().identifier(), &ident);
+ assert!(Arc::ptr_eq(commit.catalog(), &client));
+ let write = find_node::<IcebergWriteExec>(&insert).expect("a write below
the commit");
+ assert_eq!(write.table().identifier(), &ident);
+ let rebuilt_write: Arc<dyn ExecutionPlan> = Arc::new(IcebergWriteExec::new(
+ write.table().clone(),
+ write.children()[0].clone(),
+ ));
+ let rebuilt_commit = IcebergCommitExec::new(
+ commit.table().clone(),
+ commit.catalog().clone(),
+ Arc::new(CoalescePartitionsExec::new(rebuilt_write)),
+ );
+ expect![[r#"
+ +-------+
+ | count |
+ +-------+
+ | 2 |
+ +-------+"#]]
+ .assert_eq(&run(&rebuilt_commit, &ctx).await?);
+
+ // Read path: a scan pinned to a snapshot, rebuilt from its accessors,
+ // returns the same rows.
+ let table = client.load_table(&ident).await?;
+ let snapshot_id = table.metadata().current_snapshot_id().unwrap();
+ let pinned =
+ IcebergStaticTableProvider::try_new_from_table_snapshot(table,
snapshot_id)
+ .await?;
+ assert_eq!(pinned.snapshot_id(), Some(snapshot_id));
+ ctx.register_table("pinned", Arc::new(pinned.clone()))?;
+ // A later write, so a scan reading the current snapshot rather than the
+ // pinned one would return its row too.
+ ctx.sql("INSERT INTO catalog.test_plan_nodes.my_table VALUES (3,
'hopper')")
+ .await?
+ .collect()
+ .await?;
+ let latest_snapshot_id = client
+ .load_table(&ident)
+ .await?
+ .metadata()
+ .current_snapshot_id()
+ .unwrap();
+ assert_ne!(latest_snapshot_id, snapshot_id);
+ let plan = ctx
+ .sql("SELECT foo2 FROM pinned WHERE foo1 = 1")
+ .await?
+ .create_physical_plan()
+ .await?;
+ let scan = find_node::<IcebergTableScan>(&plan).expect("a scan");
+ assert_eq!(
+ scan.predicates().map(ToString::to_string).as_deref(),
+ Some("foo1 = 1")
+ );
+ let rebuilt = rebuild_scan(scan);
+ assert_eq!(rebuilt.schema(), scan.schema());
+ assert_eq!(rebuilt.projection(), scan.projection());
+ let expected = run(scan, &ctx).await?;
+ expect![[r#"
+ +------+------+
+ | foo1 | foo2 |
+ +------+------+
+ | 1 | alan |
+ +------+------+"#]]
+ .assert_eq(&expected);
+ assert_eq!(run(&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");
+ 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?;
+ expect![[r#"
+ +------+------+
+ | foo1 | foo2 |
+ +------+------+
+ | 1 | alan |
+ +------+------+"#]]
+ .assert_eq(&expected);
+ assert_eq!(run(&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(
+ pinned.table().clone(),
+ Some(snapshot_id),
+ foo2_only,
+ None,
+ None,
+ );
+ expect![[r#"
+ +--------+
+ | foo2 |
+ +--------+
+ | alan |
+ | turing |
+ +--------+"#]]
+ .assert_eq(&run(&partial, &ctx).await?);
+
+ // Metadata tables: a scan rebuilt from a metadata scan's parts reads the
+ // same rows.
+ let plan = ctx
+ .sql("SELECT * FROM catalog.test_plan_nodes.\"my_table$snapshots\"")
+ .await?
+ .create_physical_plan()
+ .await?;
+ let metadata_scan = find_node::<IcebergMetadataScan>(&plan).expect("a
metadata scan");
+ let provider = metadata_scan.provider();
+ assert_eq!(provider.table().identifier(), &ident);
+ let rebuilt = IcebergMetadataScan::new(IcebergMetadataTableProvider::new(
+ provider.table().clone(),
+ provider.metadata_type().clone(),
+ ));
+ let batches = run_batches(metadata_scan, &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.
+ let batch = concat_batches(&batches[0].schema(), &batches)?;
+ let order = sort_to_indices(batch.column_by_name("parent_id").unwrap(),
None, None)?;
+ let batch = take_record_batch(&batch, &order)?;
+ let column = |name: &str| batch.column_by_name(name).unwrap().clone();
+ let longs = |name: &str| -> Result<Vec<Option<i64>>, Box<dyn Error>> {
+ Ok(cast(&column(name), &DataType::Int64)?
+ .as_primitive::<Int64Type>()
+ .iter()
+ .collect())
+ };
+ let strings = |name: &str| -> Vec<Option<String>> {
+ column(name)
+ .as_string::<i32>()
+ .iter()
+ .map(|value| value.map(str::to_string))
+ .collect()
+ };
+ let metadata = provider.table().metadata();
+ let snapshots = [snapshot_id, latest_snapshot_id].map(|id| {
+ metadata
+ .snapshot_by_id(id)
+ .expect("a snapshot of the table")
+ });
+ assert_eq!(
+ batch
+ .schema()
+ .fields()
+ .iter()
+ .map(|field| field.name().as_str())
+ .collect::<Vec<_>>(),
+ [
+ "committed_at",
+ "snapshot_id",
+ "parent_id",
+ "operation",
+ "manifest_list",
+ "summary"
+ ]
+ );
+ assert_eq!(
+ longs("committed_at")?,
+ snapshots.map(|snapshot| Some(snapshot.timestamp_ms() * 1000))
+ );
+ assert_eq!(
+ longs("snapshot_id")?,
+ [Some(snapshot_id), Some(latest_snapshot_id)]
+ );
+ assert_eq!(longs("parent_id")?, [None, Some(snapshot_id)]);
+ assert_eq!(
+ strings("operation"),
+ [Some("append".to_string()), Some("append".to_string())]
+ );
+ assert_eq!(
+ strings("manifest_list"),
+ snapshots.map(|snapshot| Some(snapshot.manifest_list().to_string()))
+ );
+ // The summaries print their keys in no set order, so sort them.
+ let summaries = column("summary");
+ let summaries = summaries.as_map();
+ let summaries = (0..summaries.len())
+ .map(|row| {
+ let entries = summaries.value(row);
+ let keys = entries.column(0).as_string::<i32>();
+ let values = entries.column(1).as_string::<i32>();
+ (0..entries.len())
+ .map(|entry| format!("{}: {}", keys.value(entry),
values.value(entry)))
+ .collect::<BTreeSet<_>>()
+ })
+ .collect::<Vec<_>>();
+ assert_eq!(
+ summaries,
+ snapshots.map(|snapshot| {
+ snapshot
+ .summary()
+ .additional_properties
+ .iter()
+ .map(|(key, value)| format!("{key}: {value}"))
+ .collect::<BTreeSet<_>>()
+ })
+ );
+ // The rebuilt scan holds the same table, so lists its snapshots in the
+ // same order.
+ assert_eq!(
+ pretty_format_batches(&run_batches(&rebuilt,
&ctx).await?)?.to_string(),
+ pretty_format_batches(&batches)?.to_string()
+ );
+
+ Ok(())
+}
+
+/// A catalog-backed provider keeps the schema it was built with, but scans the
+/// table as it is now. After a column is added to the table and written to, a
+/// scan with no projection still returns only the provider's columns, matching
+/// its schema.
+#[tokio::test]
+async fn test_scan_after_schema_evolution_reads_provider_columns()
+-> Result<(), Box<dyn Error>> {
+ let iceberg_catalog = get_iceberg_catalog().await;
+ let namespace = NamespaceIdent::new("test_schema_evolution".to_string());
+ set_test_namespace(&iceberg_catalog, &namespace).await?;
+ let creation = get_table_creation(temp_path(), "my_table", None)?;
+ iceberg_catalog.create_table(&namespace, creation).await?;
+ let ident = TableIdent::new(namespace.clone(), "my_table".to_string());
+ let client: Arc<dyn Catalog> = Arc::new(iceberg_catalog);
+
+ let provider = Arc::new(
+ IcebergTableProvider::try_new(client.clone(), namespace,
"my_table").await?,
+ );
+ let ctx = SessionContext::new();
+ ctx.register_table("t", provider.clone())?;
+ ctx.sql("INSERT INTO t VALUES (1, 'alan')")
+ .await?
+ .collect()
+ .await?;
+
+ let table = client.load_table(&ident).await?;
+ let tx = Transaction::new(&table);
+ tx.update_schema()
+ .add_column(AddColumn::optional(
+ "foo3",
+ Type::Primitive(PrimitiveType::Int),
+ ))
+ .apply(tx)?
+ .commit(client.as_ref())
+ .await?;
+
+ // 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(
+ client.clone(),
+ ident.namespace().clone(),
+ ident.name(),
+ )
+ .await?;
+ expect![[r#"
+ Schema {
+ fields: [
+ Field {
+ name: "foo1",
+ data_type: Int32,
+ metadata: {
+ "PARQUET:field_id": "1",
+ },
+ },
+ Field {
+ name: "foo2",
+ data_type: Utf8,
+ metadata: {
+ "PARQUET:field_id": "2",
+ },
+ },
+ Field {
+ name: "foo3",
+ data_type: Int32,
+ nullable: true,
+ metadata: {
+ "PARQUET:field_id": "3",
+ },
+ },
+ ],
+ metadata: {},
+ }
+ "#]]
+ .assert_debug_eq(&evolved.schema());
Review Comment:
Same as above, is going to be hard to maintain these kind of assertions.
##########
crates/datafusion/src/physical_plan/scan.rs:
##########
@@ -64,14 +64,110 @@ impl IcebergTableScan {
projection: Option<&Vec<usize>>,
filters: &[Expr],
limit: Option<usize>,
- ) -> Self {
+ ) -> Result<Self> {
let output_schema = match projection {
- None => schema.clone(),
- Some(projection) => Arc::new(schema.project(projection).unwrap()),
+ None => schema,
+ Some(projection) => Arc::new(schema.project(projection)?),
};
- let plan_properties = Self::compute_properties(output_schema.clone());
- let projection = get_column_names(schema.clone(), projection);
- let predicates = convert_filters_to_predicate(filters);
+ Ok(Self::new_with_predicate(
+ table,
+ snapshot_id,
+ output_schema,
+ convert_filters_to_predicate(filters),
+ limit,
+ ))
+ }
+
+ /// 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.
+ ///
+ /// 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::physical_plan::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>>(())
+ /// ```
+ pub fn new_with_predicate(
Review Comment:
This method is part of the public API now, however, it seems likely that
this will needs to be extended with new arguments, forcing a breaking change
and a major release just because of this signature.
My recommendation here is to create something like an
`IcebergTableScanBuilder` using a builder pattern to guarantee public API
stability across contributions.
##########
crates/datafusion/src/physical_plan/scan.rs:
##########
@@ -64,14 +64,110 @@ impl IcebergTableScan {
projection: Option<&Vec<usize>>,
filters: &[Expr],
limit: Option<usize>,
- ) -> Self {
+ ) -> Result<Self> {
let output_schema = match projection {
- None => schema.clone(),
- Some(projection) => Arc::new(schema.project(projection).unwrap()),
+ None => schema,
+ Some(projection) => Arc::new(schema.project(projection)?),
};
- let plan_properties = Self::compute_properties(output_schema.clone());
- let projection = get_column_names(schema.clone(), projection);
- let predicates = convert_filters_to_predicate(filters);
+ Ok(Self::new_with_predicate(
+ table,
+ snapshot_id,
+ output_schema,
+ convert_filters_to_predicate(filters),
+ limit,
+ ))
+ }
+
+ /// 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.
+ ///
+ /// 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::physical_plan::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>>(())
Review Comment:
The example shown here seems unreasonably long, with some duplicated lines
and inconsistent use of doc visibility in the example (imports that are not
hidden but actual sample code that is hidden).
I recommend to shrink to make it more readable for people through the public
API.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
