You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "adriangb (via GitHub)" <gi...@apache.org> on 2026/01/06 17:58:07 UTC
[PR] Migrate SchemaAdapter to PhysicalExprAdapter [datafusion-comet]
adriangb opened a new pull request, #3047:
URL: https://github.com/apache/datafusion-comet/pull/3047
## Summary
This PR prepares for DataFusion 52.0 by migrating from the deprecated `SchemaAdapter` approach to the new `PhysicalExprAdapter` approach.
**Key changes:**
- Add `SparkPhysicalExprAdapterFactory` and `SparkPhysicalExprAdapter` that work at planning time (expression rewriting) instead of runtime (batch transformation)
- Replace `CastColumnExpr` with Spark-compatible `Cast` expressions
- Update `parquet_exec.rs` to use `with_expr_adapter()` instead of `with_schema_adapter_factory()`
- Update Iceberg scan to use `adapt_batch_with_expressions()`
- Mark old `SparkSchemaAdapterFactory` as deprecated
**The new approach:**
1. `PhysicalExprAdapterFactory.create()` returns `PhysicalExprAdapter`
2. `PhysicalExprAdapter.rewrite()` transforms expressions at planning time
3. Casts are injected as expressions that execute when the plan runs
See [DataFusion upgrading guide](https://github.com/apache/datafusion/blob/main/docs/source/library-user-guide/upgrading.md) for more context on this migration.
## Test plan
- [ ] Verify build compiles successfully
- [ ] Run existing schema adapter tests
- [ ] Test Parquet roundtrip with type casting
- [ ] Test Iceberg scan with schema evolution
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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
Re: [PR] chore: migrate SchemaAdapter to PhysicalExprAdapter for DataFusion 52 compatibility [datafusion-comet]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #3047:
URL: https://github.com/apache/datafusion-comet/pull/3047#issuecomment-4064629189
Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days.
--
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
Re: [PR] chore: migrate SchemaAdapter to PhysicalExprAdapter for DataFusion 52 compatibility [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on PR #3047:
URL: https://github.com/apache/datafusion-comet/pull/3047#issuecomment-3747533064
Hi @adriangb I went through proposed changes https://github.com/apache/datafusion/pull/19716 I still missing something. There is a use case for Comet native readers(non-iceberg), which backed by DF readers in https://github.com/apache/datafusion-comet/blob/c4da3c91d672e6bc5d41959f6d60a671cd3dfd8a/native/core/src/parquet/parquet_exec.rs#L107
The schema adapter, used for schema conversion via `map_schema` and batch runtime modifications `map_batch`.
`map_schema` looks straight forward however `map_batch` was a callback as part of parquet file opener https://github.com/apache/datafusion/blob/e571b49e0983892597a8f92e5d1502b17a15b180/datafusion/datasource-parquet/src/opener.rs#L465
In https://github.com/apache/datafusion/pull/19716 there is a proposed `BatchAdapter` however I can't find how this adapter would be called, what is the triggering mechanism. With `map_batch` the file opener was a caller and all the logic below worked with modified RB, I'm not sure how to achieve same with `BatchAdapter` though?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: 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
Re: [PR] Migrate SchemaAdapter to PhysicalExprAdapter [datafusion-comet]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #3047:
URL: https://github.com/apache/datafusion-comet/pull/3047#discussion_r2665779336
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -16,19 +16,241 @@
// under the License.
//! Custom schema adapter that uses Spark-compatible conversions
+//!
+//! This module provides both:
+//! - The deprecated `SchemaAdapter` approach (for backwards compatibility)
+//! - The new `PhysicalExprAdapter` approach (recommended, works at planning time)
use crate::parquet::parquet_support::{spark_parquet_convert, SparkParquetOptions};
-use arrow::array::{RecordBatch, RecordBatchOptions};
+use arrow::array::{ArrayRef, RecordBatch, RecordBatchOptions};
use arrow::datatypes::{Schema, SchemaRef};
-use datafusion::common::ColumnStatistics;
+use datafusion::common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion::common::{ColumnStatistics, Result as DataFusionResult};
use datafusion::datasource::schema_adapter::{SchemaAdapter, SchemaAdapterFactory, SchemaMapper};
+use datafusion::physical_expr::expressions::Column;
+use datafusion::physical_expr::PhysicalExpr;
use datafusion::physical_plan::ColumnarValue;
use datafusion::scalar::ScalarValue;
+use datafusion_comet_spark_expr::{Cast, SparkCastOptions};
+use datafusion_physical_expr_adapter::{
+ DefaultPhysicalExprAdapterFactory, PhysicalExprAdapter, PhysicalExprAdapterFactory,
+ replace_columns_with_literals,
+};
use std::collections::HashMap;
use std::sync::Arc;
+// ============================================================================
+// New PhysicalExprAdapter Implementation (Recommended)
+// ============================================================================
+
+/// Factory for creating Spark-compatible physical expression adapters.
+///
+/// This factory creates adapters that rewrite expressions at planning time
+/// to inject Spark-compatible casts where needed.
+#[derive(Clone, Debug)]
+pub struct SparkPhysicalExprAdapterFactory {
+ /// Spark-specific parquet options for type conversions
+ parquet_options: SparkParquetOptions,
+ /// Default values for columns that may be missing from the physical schema.
+ /// The key is the column index in the logical schema.
+ default_values: Option<HashMap<usize, ScalarValue>>,
+}
+
+impl SparkPhysicalExprAdapterFactory {
+ /// Create a new factory with the given options.
+ pub fn new(
+ parquet_options: SparkParquetOptions,
+ default_values: Option<HashMap<usize, ScalarValue>>,
+ ) -> Self {
+ Self {
+ parquet_options,
+ default_values,
+ }
+ }
+}
+
+impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory {
+ fn create(
+ &self,
+ logical_file_schema: SchemaRef,
+ physical_file_schema: SchemaRef,
+ ) -> Arc<dyn PhysicalExprAdapter> {
+ let default_factory = DefaultPhysicalExprAdapterFactory;
+ let default_adapter = default_factory.create(
+ Arc::clone(&logical_file_schema),
+ Arc::clone(&physical_file_schema),
+ );
+
+ Arc::new(SparkPhysicalExprAdapter {
+ logical_file_schema,
+ physical_file_schema,
+ parquet_options: self.parquet_options.clone(),
+ default_values: self.default_values.clone(),
+ default_adapter,
+ })
+ }
+}
+
+/// Spark-compatible physical expression adapter.
+///
+/// This adapter rewrites expressions at planning time to:
+/// 1. Replace references to missing columns with default values or nulls
+/// 2. Replace standard DataFusion cast expressions with Spark-compatible casts
+/// 3. Handle case-insensitive column matching
+#[derive(Debug)]
+struct SparkPhysicalExprAdapter {
+ /// The logical schema expected by the query
+ logical_file_schema: SchemaRef,
+ /// The physical schema of the actual file being read
+ physical_file_schema: SchemaRef,
+ /// Spark-specific options for type conversions
+ parquet_options: SparkParquetOptions,
+ /// Default values for missing columns (keyed by logical schema index)
+ default_values: Option<HashMap<usize, ScalarValue>>,
+ /// The default DataFusion adapter to delegate standard handling to
+ default_adapter: Arc<dyn PhysicalExprAdapter>,
+}
+
+impl PhysicalExprAdapter for SparkPhysicalExprAdapter {
+ fn rewrite(&self, expr: Arc<dyn PhysicalExpr>) -> DataFusionResult<Arc<dyn PhysicalExpr>> {
+ // Step 1: Handle default values for missing columns
+ let expr = self.replace_missing_with_defaults(expr)?;
+
+ // Step 2: Delegate to default adapter for standard handling
+ // This handles: missing columns → nulls, type mismatches → CastColumnExpr
+ let expr = self.default_adapter.rewrite(expr)?;
+
+ // Step 3: Replace CastColumnExpr with Spark-compatible Cast expressions
+ expr.transform(|e| self.replace_with_spark_cast(e)).data()
+ }
+}
+
+impl SparkPhysicalExprAdapter {
+ /// Replace CastColumnExpr (DataFusion's cast) with Spark's Cast expression.
+ fn replace_with_spark_cast(
+ &self,
+ expr: Arc<dyn PhysicalExpr>,
+ ) -> DataFusionResult<Transformed<Arc<dyn PhysicalExpr>>> {
+ // Check for CastColumnExpr and replace with spark_expr::Cast
+ // CastColumnExpr is in datafusion_physical_expr::expressions
+ if let Some(cast) = expr
+ .as_any()
+ .downcast_ref::<datafusion::physical_expr::expressions::CastColumnExpr>()
+ {
+ let child = cast.expr().clone();
+ let target_type = cast.target_field().data_type().clone();
+
+ // Create Spark-compatible cast options
+ let mut cast_options = SparkCastOptions::new(
+ self.parquet_options.eval_mode,
+ &self.parquet_options.timezone,
+ self.parquet_options.allow_incompat,
+ );
+ cast_options.allow_cast_unsigned_ints = self.parquet_options.allow_cast_unsigned_ints;
+ cast_options.is_adapting_schema = true;
+
+ let spark_cast = Arc::new(Cast::new(child, target_type, cast_options));
+
+ return Ok(Transformed::yes(spark_cast as Arc<dyn PhysicalExpr>));
+ }
+
+ Ok(Transformed::no(expr))
+ }
+
+ /// Replace references to missing columns with default values.
+ fn replace_missing_with_defaults(
+ &self,
+ expr: Arc<dyn PhysicalExpr>,
+ ) -> DataFusionResult<Arc<dyn PhysicalExpr>> {
+ let Some(defaults) = &self.default_values else {
+ return Ok(expr);
+ };
+
+ if defaults.is_empty() {
+ return Ok(expr);
+ }
+
+ // Convert index-based defaults to name-based for replace_columns_with_literals
+ let name_based: HashMap<&str, &ScalarValue> = defaults
+ .iter()
+ .filter_map(|(idx, val)| {
+ self.logical_file_schema
+ .fields()
+ .get(*idx)
+ .map(|f| (f.name().as_str(), val))
+ })
+ .collect();
+
+ if name_based.is_empty() {
+ return Ok(expr);
+ }
+
+ replace_columns_with_literals(expr, &name_based)
+ }
+}
+
+/// Adapt a batch to match the target schema using expression evaluation.
+///
+/// This function is useful for cases like Iceberg scanning where batches
+/// are read directly and need to be adapted to the expected schema.
+pub fn adapt_batch_with_expressions(
+ batch: RecordBatch,
+ target_schema: &SchemaRef,
+ parquet_options: &SparkParquetOptions,
+) -> DataFusionResult<RecordBatch> {
+ let file_schema = batch.schema();
+
+ // If schemas match, no adaptation needed
+ if file_schema.as_ref() == target_schema.as_ref() {
+ return Ok(batch);
+ }
+
+ // Create adapter
+ let factory = SparkPhysicalExprAdapterFactory::new(parquet_options.clone(), None);
+ let adapter = factory.create(Arc::clone(target_schema), Arc::clone(&file_schema));
+
+ // Create column projection expressions for target schema
+ let projection_exprs: 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::<DataFusionResult<Vec<_>>>()?;
+
+ // Evaluate expressions against batch
+ let columns: Vec<ArrayRef> = projection_exprs
+ .iter()
+ .map(|expr| {
+ expr.evaluate(&batch)?
+ .into_array(batch.num_rows())
+ .map_err(|e| e.into())
+ })
+ .collect::<DataFusionResult<Vec<_>>>()?;
+
+ RecordBatch::try_new(Arc::clone(target_schema), columns).map_err(|e| e.into())
+}
+
+// ============================================================================
+// Legacy SchemaAdapter Implementation (Deprecated)
Review Comment:
Maybe just delete it?
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -16,19 +16,241 @@
// under the License.
//! Custom schema adapter that uses Spark-compatible conversions
+//!
+//! This module provides both:
+//! - The deprecated `SchemaAdapter` approach (for backwards compatibility)
+//! - The new `PhysicalExprAdapter` approach (recommended, works at planning time)
use crate::parquet::parquet_support::{spark_parquet_convert, SparkParquetOptions};
-use arrow::array::{RecordBatch, RecordBatchOptions};
+use arrow::array::{ArrayRef, RecordBatch, RecordBatchOptions};
use arrow::datatypes::{Schema, SchemaRef};
-use datafusion::common::ColumnStatistics;
+use datafusion::common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion::common::{ColumnStatistics, Result as DataFusionResult};
use datafusion::datasource::schema_adapter::{SchemaAdapter, SchemaAdapterFactory, SchemaMapper};
+use datafusion::physical_expr::expressions::Column;
+use datafusion::physical_expr::PhysicalExpr;
use datafusion::physical_plan::ColumnarValue;
use datafusion::scalar::ScalarValue;
+use datafusion_comet_spark_expr::{Cast, SparkCastOptions};
+use datafusion_physical_expr_adapter::{
+ DefaultPhysicalExprAdapterFactory, PhysicalExprAdapter, PhysicalExprAdapterFactory,
+ replace_columns_with_literals,
+};
use std::collections::HashMap;
use std::sync::Arc;
+// ============================================================================
+// New PhysicalExprAdapter Implementation (Recommended)
+// ============================================================================
+
+/// Factory for creating Spark-compatible physical expression adapters.
+///
+/// This factory creates adapters that rewrite expressions at planning time
+/// to inject Spark-compatible casts where needed.
+#[derive(Clone, Debug)]
+pub struct SparkPhysicalExprAdapterFactory {
+ /// Spark-specific parquet options for type conversions
+ parquet_options: SparkParquetOptions,
+ /// Default values for columns that may be missing from the physical schema.
+ /// The key is the column index in the logical schema.
+ default_values: Option<HashMap<usize, ScalarValue>>,
+}
+
+impl SparkPhysicalExprAdapterFactory {
+ /// Create a new factory with the given options.
+ pub fn new(
+ parquet_options: SparkParquetOptions,
+ default_values: Option<HashMap<usize, ScalarValue>>,
+ ) -> Self {
+ Self {
+ parquet_options,
+ default_values,
+ }
+ }
+}
+
+impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory {
+ fn create(
+ &self,
+ logical_file_schema: SchemaRef,
+ physical_file_schema: SchemaRef,
+ ) -> Arc<dyn PhysicalExprAdapter> {
+ let default_factory = DefaultPhysicalExprAdapterFactory;
+ let default_adapter = default_factory.create(
+ Arc::clone(&logical_file_schema),
+ Arc::clone(&physical_file_schema),
+ );
+
+ Arc::new(SparkPhysicalExprAdapter {
+ logical_file_schema,
+ physical_file_schema,
+ parquet_options: self.parquet_options.clone(),
+ default_values: self.default_values.clone(),
+ default_adapter,
+ })
+ }
+}
+
+/// Spark-compatible physical expression adapter.
+///
+/// This adapter rewrites expressions at planning time to:
+/// 1. Replace references to missing columns with default values or nulls
+/// 2. Replace standard DataFusion cast expressions with Spark-compatible casts
+/// 3. Handle case-insensitive column matching
+#[derive(Debug)]
+struct SparkPhysicalExprAdapter {
+ /// The logical schema expected by the query
+ logical_file_schema: SchemaRef,
+ /// The physical schema of the actual file being read
+ physical_file_schema: SchemaRef,
+ /// Spark-specific options for type conversions
+ parquet_options: SparkParquetOptions,
+ /// Default values for missing columns (keyed by logical schema index)
+ default_values: Option<HashMap<usize, ScalarValue>>,
+ /// The default DataFusion adapter to delegate standard handling to
+ default_adapter: Arc<dyn PhysicalExprAdapter>,
+}
+
+impl PhysicalExprAdapter for SparkPhysicalExprAdapter {
+ fn rewrite(&self, expr: Arc<dyn PhysicalExpr>) -> DataFusionResult<Arc<dyn PhysicalExpr>> {
+ // Step 1: Handle default values for missing columns
+ let expr = self.replace_missing_with_defaults(expr)?;
+
+ // Step 2: Delegate to default adapter for standard handling
+ // This handles: missing columns → nulls, type mismatches → CastColumnExpr
+ let expr = self.default_adapter.rewrite(expr)?;
+
+ // Step 3: Replace CastColumnExpr with Spark-compatible Cast expressions
+ expr.transform(|e| self.replace_with_spark_cast(e)).data()
+ }
+}
+
+impl SparkPhysicalExprAdapter {
+ /// Replace CastColumnExpr (DataFusion's cast) with Spark's Cast expression.
+ fn replace_with_spark_cast(
+ &self,
+ expr: Arc<dyn PhysicalExpr>,
+ ) -> DataFusionResult<Transformed<Arc<dyn PhysicalExpr>>> {
+ // Check for CastColumnExpr and replace with spark_expr::Cast
+ // CastColumnExpr is in datafusion_physical_expr::expressions
+ if let Some(cast) = expr
+ .as_any()
+ .downcast_ref::<datafusion::physical_expr::expressions::CastColumnExpr>()
+ {
+ let child = cast.expr().clone();
+ let target_type = cast.target_field().data_type().clone();
+
+ // Create Spark-compatible cast options
+ let mut cast_options = SparkCastOptions::new(
+ self.parquet_options.eval_mode,
+ &self.parquet_options.timezone,
+ self.parquet_options.allow_incompat,
+ );
+ cast_options.allow_cast_unsigned_ints = self.parquet_options.allow_cast_unsigned_ints;
+ cast_options.is_adapting_schema = true;
+
+ let spark_cast = Arc::new(Cast::new(child, target_type, cast_options));
+
+ return Ok(Transformed::yes(spark_cast as Arc<dyn PhysicalExpr>));
+ }
+
+ Ok(Transformed::no(expr))
+ }
+
+ /// Replace references to missing columns with default values.
+ fn replace_missing_with_defaults(
+ &self,
+ expr: Arc<dyn PhysicalExpr>,
+ ) -> DataFusionResult<Arc<dyn PhysicalExpr>> {
+ let Some(defaults) = &self.default_values else {
+ return Ok(expr);
+ };
+
+ if defaults.is_empty() {
+ return Ok(expr);
+ }
+
+ // Convert index-based defaults to name-based for replace_columns_with_literals
+ let name_based: HashMap<&str, &ScalarValue> = defaults
+ .iter()
+ .filter_map(|(idx, val)| {
+ self.logical_file_schema
+ .fields()
+ .get(*idx)
+ .map(|f| (f.name().as_str(), val))
+ })
+ .collect();
+
+ if name_based.is_empty() {
+ return Ok(expr);
+ }
+
+ replace_columns_with_literals(expr, &name_based)
+ }
+}
+
+/// Adapt a batch to match the target schema using expression evaluation.
+///
+/// This function is useful for cases like Iceberg scanning where batches
+/// are read directly and need to be adapted to the expected schema.
+pub fn adapt_batch_with_expressions(
+ batch: RecordBatch,
+ target_schema: &SchemaRef,
+ parquet_options: &SparkParquetOptions,
+) -> DataFusionResult<RecordBatch> {
+ let file_schema = batch.schema();
+
+ // If schemas match, no adaptation needed
+ if file_schema.as_ref() == target_schema.as_ref() {
+ return Ok(batch);
+ }
+
+ // Create adapter
+ let factory = SparkPhysicalExprAdapterFactory::new(parquet_options.clone(), None);
+ let adapter = factory.create(Arc::clone(target_schema), Arc::clone(&file_schema));
+
+ // Create column projection expressions for target schema
+ let projection_exprs: 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)
+ })
Review Comment:
```suggestion
.map(|(i, field)| {
let col_expr: Arc<dyn PhysicalExpr> = Arc::new(Column::new(
field.name(),
i,
));
adapter.rewrite(col_expr)
})
```
--
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
Re: [PR] chore: migrate SchemaAdapter to PhysicalExprAdapter for DataFusion 52 compatibility [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on PR #3047:
URL: https://github.com/apache/datafusion-comet/pull/3047#issuecomment-3747772071
I made changes in https://github.com/apache/datafusion-comet/pull/3052 to reflect the proposition, however I can see rust test failed
```
failures:
execution::planner::tests::test_nested_types_extract_missing_struct_names_non_overlap
execution::planner::tests::test_nested_types_list_of_struct_by_index
execution::planner::tests::test_nested_types_map_keys
execution::planner::tests::test_unpack_dictionary_primitive
execution::planner::tests::test_unpack_dictionary_string
```
I assume the reason is our `map_batch` which reconstructs the batch to comply to user defined schema was never called
--
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
Re: [PR] chore: migrate SchemaAdapter to PhysicalExprAdapter for DataFusion 52 compatibility [datafusion-comet]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #3047: chore: migrate SchemaAdapter to PhysicalExprAdapter for DataFusion 52 compatibility
URL: https://github.com/apache/datafusion-comet/pull/3047
--
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
Re: [PR] chore: migrate SchemaAdapter to PhysicalExprAdapter for DataFusion 52 compatibility [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on PR #3047:
URL: https://github.com/apache/datafusion-comet/pull/3047#issuecomment-3750596200
> It seems you were missing one swap for PhysicalExprAdapter, not sure if that's the problem or not. If there continue to be issues happy to set up a call for next week and we can go over it?
Would be awesome, to set up details, you can find me in DataFusion discord or Apache Slack
--
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
Re: [PR] chore: migrate SchemaAdapter to PhysicalExprAdapter for DataFusion 52 compatibility [datafusion-comet]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #3047:
URL: https://github.com/apache/datafusion-comet/pull/3047#issuecomment-3747814585
It seems you were missing one swap for PhysicalExprAdapter, not sure if that's the problem or not. If there continue to be issues happy to set up a call for next week and we can go over it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: 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
Re: [PR] chore: migrate SchemaAdapter to PhysicalExprAdapter for DataFusion 52 compatibility [datafusion-comet]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #3047:
URL: https://github.com/apache/datafusion-comet/pull/3047#issuecomment-3716351753
@comphead I've looked through the diff and the approach seems solid to me. I tried to get running locally but am running into what to me are exotic Java errors so unless it would be very helpful to you how about I hand this off? I'll note that although the dep is still DF 51 this was written against the DF 52 APIs.
--
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
Re: [PR] chore: migrate SchemaAdapter to PhysicalExprAdapter for DataFusion 52 compatibility [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on PR #3047:
URL: https://github.com/apache/datafusion-comet/pull/3047#issuecomment-3716453224
> @comphead I've looked through the diff and the approach seems solid to me. I tried to get running locally but am running into what to me are exotic Java errors so unless it would be very helpful to you how about I hand this off? I'll note that although the dep is still DF 51 this was written against the DF 52 APIs.
Its okay, thanks @adriangb for giving pointers, we already tracking this issue in https://github.com/apache/datafusion-comet/issues/2058. I'll check if it would be straightfwd to migrate SchemaAdapter to Comet just for now and this approach would be base to migrate later. Or sooner if SchemaAdapter not easily transferrable
--
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
Re: [PR] chore: migrate SchemaAdapter to PhysicalExprAdapter for DataFusion 52 compatibility [datafusion-comet]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #3047:
URL: https://github.com/apache/datafusion-comet/pull/3047#issuecomment-3716492067
I *think* you should have no issues migrating to `PhysicalExprAdapter`, it covers all of the use cases of `SchemaAdapter` (and more) with a simpler API to write against. Forward porting `SchemaAdapter` may be difficult considering it's not supported in `ParquetOpener`, etc. anymore.
--
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
Re: [PR] Migrate SchemaAdapter to PhysicalExprAdapter [datafusion-comet]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #3047:
URL: https://github.com/apache/datafusion-comet/pull/3047#issuecomment-3715734176
(I expect a lot of CI failures, I did not run this locally as I don't work with Java / Spark and did not want to set up the stack)
--
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
Re: [PR] chore: migrate SchemaAdapter to PhysicalExprAdapter for DataFusion 52 compatibility [datafusion-comet]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #3047:
URL: https://github.com/apache/datafusion-comet/pull/3047#issuecomment-3747603386
> Hi @adriangb I went through proposed changes [apache/datafusion#19716](https://github.com/apache/datafusion/pull/19716) I still missing something. There is a use case for Comet native readers(non-iceberg), which backed by DF readers in
>
> https://github.com/apache/datafusion-comet/blob/c4da3c91d672e6bc5d41959f6d60a671cd3dfd8a/native/core/src/parquet/parquet_exec.rs#L107
>
> The schema adapter, used for schema conversion via `map_schema` and batch runtime modifications `map_batch`. `map_schema` looks straight forward however `map_batch` was a callback as part of parquet file opener https://github.com/apache/datafusion/blob/e571b49e0983892597a8f92e5d1502b17a15b180/datafusion/datasource-parquet/src/opener.rs#L465
>
> In [apache/datafusion#19716](https://github.com/apache/datafusion/pull/19716) there is a proposed `BatchAdapter` however I can't find how this adapter would be called, what is the triggering mechanism. With `map_batch` the file opener was a caller and all the logic below worked with modified RB, I'm not sure how to achieve same with `BatchAdapter` though?
This is covered upstream in DataFusion, the only changes needed here I already handled in this PR: https://github.com/apache/datafusion-comet/pull/3047/changes#diff-d573ace4d519dc7995a332cd91ffcaf553475dbd80db1cda8f5707c35f8e4c9b
The whole point of the new API is that you get to define via expressions how you want to manipulate the schemas and batches (since expressions encapsulate both concepts of how to transform the data and how the schemas get transformed). Internally `ParquetOpener` does something similar to the `BatchaAdapter` being discussed in https://github.com/apache/datafusion/pull/19716:
https://github.com/apache/datafusion/blob/567ba75840494170cbe7e50c695110d447426c8c/datafusion/datasource-parquet/src/opener.rs#L401-L418
https://github.com/apache/datafusion/blob/567ba75840494170cbe7e50c695110d447426c8c/datafusion/datasource-parquet/src/opener.rs#L616-L622
https://github.com/apache/datafusion/blob/567ba75840494170cbe7e50c695110d447426c8c/datafusion/datasource-parquet/src/opener.rs#L631
But you shouldn't have to worry about that, providing the custom `PhysicalExprAdapterFactory` like I do in this PR should be enough.
--
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