Re: [PR] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [iceberg] [datafusion-comet]

["mbutrovich (via GitHub)" <gi...@apache.org>]

mbutrovich commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2854464577


##########
native/core/src/execution/operators/iceberg_scan.rs:
##########
@@ -251,38 +258,34 @@ where
             Poll::Ready(Some(Ok(batch))) => {
                 let file_schema = batch.schema();
 
-                // Check if we need to create a new adapter for this file's schema
-                let needs_new_adapter = match &self.cached_adapter {
-                    Some((cached_schema, _)) => !Arc::ptr_eq(cached_schema, &file_schema),
-                    None => true,
+                // Reuse cached projection expressions if file schema hasn't changed,
+                // otherwise create a new adapter and build new expressions
+                let projection_exprs = match &self.cached {
+                    Some(cached) if cached.file_schema.as_ref() == file_schema.as_ref() => {

Review Comment:
   This is using a deep `Schema ==` instead of `Arc::ptr_eq`. Since iceberg-rust reuses the same `Arc<Schema>` for all batches from a file, `ptr_eq` was faster.



##########
native/core/src/execution/operators/iceberg_scan.rs:
##########
@@ -313,3 +316,48 @@ impl DisplayAs for IcebergScanExec {
         )
     }
 }
+
+/// Build projection expressions that adapt batches from a file schema to the target schema.
+///
+/// The returned expressions can be cached and reused across multiple batches
+/// that share the same file schema, avoiding repeated expression construction.
+fn build_projection_expressions(
+    target_schema: &SchemaRef,
+    adapter: &Arc<dyn PhysicalExprAdapter>,
+) -> DFResult<Vec<Arc<dyn PhysicalExpr>>> {
+    target_schema
+        .fields()
+        .iter()
+        .enumerate()
+        .map(|(i, _field)| {
+            let col_expr: Arc<dyn PhysicalExpr> = Arc::new(Column::new_with_schema(
+                target_schema.field(i).name(),
+                target_schema.as_ref(),
+            )?);
+            adapter.rewrite(col_expr)
+        })
+        .collect::<DFResult<Vec<_>>>()
+}
+
+/// Adapt a batch to match the target schema using pre-built projection expressions.
+///
+/// The caller provides pre-built `projection_exprs` (from [`build_projection_expressions`])
+/// which can be cached and reused across multiple batches with the same file schema.
+fn adapt_batch_with_expressions(
+    batch: RecordBatch,
+    target_schema: &SchemaRef,
+    projection_exprs: &[Arc<dyn PhysicalExpr>],
+) -> DFResult<RecordBatch> {
+    // If schemas match, no adaptation needed
+    if batch.schema().as_ref() == target_schema.as_ref() {

Review Comment:
   This is using a deep `Schema ==` instead of `Arc::ptr_eq`. Since iceberg-rust reuses the same `Arc<Schema>` for all batches from a file, `ptr_eq` was faster.
   
   Also, this check might be redundant if I'm understanding control flow. If the caller has cached projection expressions, it should apply them unconditionally. If we do need the check, use `Arc::ptr_eq` instead.



##########
native/core/src/execution/operators/iceberg_scan.rs:
##########
@@ -313,3 +316,48 @@ impl DisplayAs for IcebergScanExec {
         )
     }
 }
+
+/// Build projection expressions that adapt batches from a file schema to the target schema.
+///
+/// The returned expressions can be cached and reused across multiple batches
+/// that share the same file schema, avoiding repeated expression construction.
+fn build_projection_expressions(
+    target_schema: &SchemaRef,
+    adapter: &Arc<dyn PhysicalExprAdapter>,
+) -> DFResult<Vec<Arc<dyn PhysicalExpr>>> {
+    target_schema
+        .fields()
+        .iter()
+        .enumerate()
+        .map(|(i, _field)| {
+            let col_expr: Arc<dyn PhysicalExpr> = Arc::new(Column::new_with_schema(
+                target_schema.field(i).name(),
+                target_schema.as_ref(),
+            )?);
+            adapter.rewrite(col_expr)
+        })
+        .collect::<DFResult<Vec<_>>>()
+}
+
+/// Adapt a batch to match the target schema using pre-built projection expressions.
+///
+/// The caller provides pre-built `projection_exprs` (from [`build_projection_expressions`])
+/// which can be cached and reused across multiple batches with the same file schema.
+fn adapt_batch_with_expressions(
+    batch: RecordBatch,
+    target_schema: &SchemaRef,
+    projection_exprs: &[Arc<dyn PhysicalExpr>],
+) -> DFResult<RecordBatch> {
+    // If schemas match, no adaptation needed
+    if batch.schema().as_ref() == target_schema.as_ref() {
+        return Ok(batch);
+    }
+
+    // Evaluate expressions against batch
+    let columns: Vec<ArrayRef> = projection_exprs

Review Comment:
   This might be a future optimization if I'm understanding control flow correctly, but for default value/schema change columns we're allocating new `ArrayRef` every time. It seems like there are caching opportunities here. Maybe file an issue to optimize default value/identity columns/schema change columns in the expression adapter.



-- 
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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org