NoahKusaba commented on code in PR #20:
URL: https://github.com/apache/datafusion-iceberg/pull/20#discussion_r4208962382
##########
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:
I actually didn't know that, thanks for letting me know.
Is there a book / article I should read about established rust repository
best practices that you would recommend? Something that covers stuff like this
doc.rs thing or the try_new thing?
--
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]
