You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "comphead (via GitHub)" <gi...@apache.org> on 2026/02/16 16:32:34 UTC
[PR] chore: [df52] migration [datafusion-comet]
comphead opened a new pull request, #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
Closes #.
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
## What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
## How are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [iceberg] [datafusion-comet]
Posted by "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
Re: [PR] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2835119622
##########
spark/src/test/scala/org/apache/comet/CometMathExpressionSuite.scala:
##########
@@ -92,7 +92,8 @@ class CometMathExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe
DataGenOptions(generateNegativeZero = generateNegativeZero))
}
- test("width_bucket") {
+ // https://github.com/apache/datafusion/issues/20320
Review Comment:
Same comment about the closed PR.
##########
spark/src/main/spark-4.0/org/apache/comet/shims/CometExprShim.scala:
##########
@@ -104,9 +104,14 @@ trait CometExprShim extends CommonStringExprs {
}
case wb: WidthBucket =>
- val childExprs = wb.children.map(exprToProtoInternal(_, inputs, binding))
- val optExpr = scalarFunctionExprToProto("width_bucket", childExprs: _*)
- optExprWithInfo(optExpr, wb, wb.children: _*)
+ withInfo(
+ wb,
+ "WidthBucket not supported, track https://github.com/apache/datafusion/issues/20320")
+ None
+// https://github.com/apache/datafusion/issues/20320
Review Comment:
Same comment about the closed PR.
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2835121892
##########
spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala:
##########
@@ -709,7 +709,7 @@ class CometExecSuite extends CometTestBase {
assert(metrics.contains("input_rows"))
assert(metrics("input_rows").value == 5L)
assert(metrics.contains("output_batches"))
- assert(metrics("output_batches").value == 5L)
+ assert(metrics("output_batches").value == 1L)
Review Comment:
Why did this change?
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2834850991
##########
spark/src/test/scala/org/apache/comet/CometCastSuite.scala:
##########
@@ -1053,7 +1053,8 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
}
test("cast TimestampType to LongType") {
- castTest(generateTimestampsExtended(), DataTypes.LongType)
+ // currently fails on timestamps outside chrono
Review Comment:
passed
--
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: [df52] migration [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2819302829
##########
native/spark-expr/src/utils.rs:
##########
@@ -71,6 +72,49 @@ pub fn array_with_timezone(
to_type: Option<&DataType>,
) -> Result<ArrayRef, ArrowError> {
match array.data_type() {
+ DataType::Timestamp(TimeUnit::Millisecond, None) => {
+ assert!(!timezone.is_empty());
+ match to_type {
+ Some(DataType::Utf8) | Some(DataType::Date32) => Ok(array),
+ Some(DataType::Timestamp(_, Some(_))) => {
+ timestamp_ntz_to_timestamp(array, timezone.as_str(), Some(timezone.as_str()))
+ }
+ Some(DataType::Timestamp(TimeUnit::Microsecond, None)) => {
+ // Convert from Timestamp(Millisecond, None) to Timestamp(Microsecond, None)
+ let millis_array = as_primitive_array::<TimestampMillisecondType>(&array);
+ let micros_array: TimestampMicrosecondArray = millis_array
+ .iter()
+ .map(|opt| opt.map(|v| v * 1000))
+ .collect();
+ Ok(Arc::new(micros_array))
+ }
+ _ => {
+ // Not supported
+ panic!(
Review Comment:
We should return an Err here
##########
native/spark-expr/src/utils.rs:
##########
@@ -71,6 +72,49 @@ pub fn array_with_timezone(
to_type: Option<&DataType>,
) -> Result<ArrayRef, ArrowError> {
match array.data_type() {
+ DataType::Timestamp(TimeUnit::Millisecond, None) => {
+ assert!(!timezone.is_empty());
+ match to_type {
+ Some(DataType::Utf8) | Some(DataType::Date32) => Ok(array),
+ Some(DataType::Timestamp(_, Some(_))) => {
+ timestamp_ntz_to_timestamp(array, timezone.as_str(), Some(timezone.as_str()))
+ }
+ Some(DataType::Timestamp(TimeUnit::Microsecond, None)) => {
+ // Convert from Timestamp(Millisecond, None) to Timestamp(Microsecond, None)
+ let millis_array = as_primitive_array::<TimestampMillisecondType>(&array);
+ let micros_array: TimestampMicrosecondArray = millis_array
+ .iter()
+ .map(|opt| opt.map(|v| v * 1000))
+ .collect();
+ Ok(Arc::new(micros_array))
+ }
+ _ => {
+ // Not supported
+ panic!(
+ "Cannot convert from {:?} to {:?}",
+ array.data_type(),
+ to_type.unwrap()
+ )
+ }
+ }
+ }
+ DataType::Timestamp(TimeUnit::Microsecond, None) => {
+ assert!(!timezone.is_empty());
+ match to_type {
+ Some(DataType::Utf8) | Some(DataType::Date32) => Ok(array),
+ Some(DataType::Timestamp(_, Some(_))) => {
+ timestamp_ntz_to_timestamp(array, timezone.as_str(), Some(timezone.as_str()))
+ }
+ _ => {
+ // Not supported
+ panic!(
Review Comment:
We should return an Err here
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2835177156
##########
spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala:
##########
@@ -709,7 +709,7 @@ class CometExecSuite extends CometTestBase {
assert(metrics.contains("input_rows"))
assert(metrics("input_rows").value == 5L)
assert(metrics.contains("output_batches"))
- assert(metrics("output_batches").value == 5L)
+ assert(metrics("output_batches").value == 1L)
Review Comment:
Assuming it is because https://github.com/apache/datafusion/pull/18972
--
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: [df52] migration [datafusion-comet]
Posted by "sqlbenchmark (via GitHub)" <gi...@apache.org>.
sqlbenchmark commented on PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#issuecomment-3909965495
## Comet TPC-H Benchmark Results
**Baseline:** `main` (`e22f35c`)
**PR:** `749a407` - DataFusion 52 migration
**Scale Factor:** SF100
**Iterations:** 1
### Query Times
| Query | Baseline Avg (s) | Baseline Best (s) | PR Avg (s) | PR Best (s) | Change (Avg) | Change (Best) |
|-------|-----------------|-------------------|-----------|------------|-------------|--------------|
| Q1 | 10.64 | 10.64 | 10.75 | 10.75 | :white_circle: +1.1% | :white_circle: +1.1% |
| Q2 | 5.90 | 5.90 | 5.76 | 5.76 | :white_circle: -2.4% | :white_circle: -2.4% |
| Q3 | 9.79 | 9.79 | 9.81 | 9.81 | :white_circle: +0.2% | :white_circle: +0.2% |
| Q4 | 11.62 | 11.62 | 13.99 | 13.99 | :red_circle: +20.4% | :red_circle: +20.4% |
| Q5 | 19.16 | 19.16 | 18.81 | 18.81 | :white_circle: -1.8% | :white_circle: -1.8% |
| Q6 | 2.52 | 2.52 | 2.52 | 2.52 | :white_circle: -0.1% | :white_circle: -0.1% |
| Q7 | 12.44 | 12.44 | 11.79 | 11.79 | :green_circle: -5.2% | :green_circle: -5.2% |
| Q8 | 25.43 | 25.43 | 24.39 | 24.39 | :white_circle: -4.1% | :white_circle: -4.1% |
| Q9 | 39.40 | 39.40 | 37.94 | 37.94 | :white_circle: -3.7% | :white_circle: -3.7% |
| Q10 | 10.37 | 10.37 | 10.62 | 10.62 | :white_circle: +2.4% | :white_circle: +2.4% |
| Q11 | 4.50 | 4.50 | 4.80 | 4.80 | :red_circle: +6.5% | :red_circle: +6.5% |
| Q12 | 7.07 | 7.07 | 6.41 | 6.41 | :green_circle: -9.3% | :green_circle: -9.3% |
| Q13 | 7.36 | 7.36 | 7.40 | 7.40 | :white_circle: +0.6% | :white_circle: +0.6% |
| Q14 | 3.59 | 3.59 | 3.27 | 3.27 | :green_circle: -8.9% | :green_circle: -8.9% |
| Q15 | 7.21 | 7.21 | 7.32 | 7.32 | :white_circle: +1.4% | :white_circle: +1.4% |
| Q16 | 4.73 | 4.73 | 4.93 | 4.93 | :white_circle: +4.0% | :white_circle: +4.0% |
| Q17 | 32.63 | 32.63 | 32.55 | 32.55 | :white_circle: -0.2% | :white_circle: -0.2% |
| Q18 | 33.85 | 33.85 | 34.62 | 34.62 | :white_circle: +2.3% | :white_circle: +2.3% |
| Q19 | 6.85 | 6.85 | 6.01 | 6.01 | :green_circle: -12.2% | :green_circle: -12.2% |
| Q20 | 6.75 | 6.75 | 6.12 | 6.12 | :green_circle: -9.3% | :green_circle: -9.3% |
| Q21 | 46.71 | 46.71 | 51.97 | 51.97 | :red_circle: +11.3% | :red_circle: +11.3% |
| Q22 | 4.92 | 4.92 | 5.41 | 5.41 | :red_circle: +9.9% | :red_circle: +9.9% |
| **Total** | **313.45** | **313.45** | **317.18** | **317.18** | :white_circle: +1.2% | :white_circle: +1.2% |
<details>
<summary><b>Spark Configuration</b></summary>
| Setting | Value |
|---------|-------|
| Spark Master | `local[*]` |
| Driver Memory | 32G |
| Driver Cores | 8 |
| Executor Memory | 32G |
| Executor Cores | 8 |
| Off-Heap Enabled | true |
| Off-Heap Size | 24g |
| Shuffle Manager | `CometShuffleManager` |
| Comet Replace SMJ | true |
</details>
---
*Automated benchmark run by dfbench*
--
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: [df52] migration [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2819304750
##########
native/core/src/parquet/parquet_exec.rs:
##########
@@ -166,27 +203,24 @@ fn get_options(
(table_parquet_options, spark_parquet_options)
}
-fn get_file_config_builder(
- schema: SchemaRef,
- partition_schema: Option<SchemaRef>,
- file_groups: Vec<FileGroup>,
- object_store_url: ObjectStoreUrl,
- file_source: Arc<dyn FileSource>,
-) -> FileScanConfigBuilder {
- match partition_schema {
- Some(partition_schema) => {
- let partition_fields: Vec<Field> = partition_schema
- .fields()
- .iter()
- .map(|field| {
- Field::new(field.name(), field.data_type().clone(), field.is_nullable())
- })
- .collect_vec();
- FileScanConfigBuilder::new(object_store_url, Arc::clone(&schema), file_source)
- .with_file_groups(file_groups)
- .with_table_partition_cols(partition_fields)
+/// Wraps a `SendableRecordBatchStream` to print each batch as it flows through.
+/// Returns a new `SendableRecordBatchStream` that yields the same batches.
+pub fn dbg_batch_stream(stream: SendableRecordBatchStream) -> SendableRecordBatchStream {
Review Comment:
this function is unused now?
--
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: [df52] migration [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2819296741
##########
native/spark-expr/src/conversion_funcs/cast.rs:
##########
@@ -707,22 +707,38 @@ pub fn spark_cast(
data_type: &DataType,
cast_options: &SparkCastOptions,
) -> DataFusionResult<ColumnarValue> {
- match arg {
- ColumnarValue::Array(array) => Ok(ColumnarValue::Array(cast_array(
- array,
- data_type,
- cast_options,
- )?)),
+ // let input_type = match &arg {
+ // ColumnarValue::Array(array) => array.data_type().clone(),
+ // ColumnarValue::Scalar(scalar) => scalar.data_type(),
+ // };
+
+ let result = match arg {
+ ColumnarValue::Array(array) => {
+ let result_array = cast_array(array, data_type, cast_options)?;
+ ColumnarValue::Array(result_array)
+ }
ColumnarValue::Scalar(scalar) => {
// Note that normally CAST(scalar) should be fold in Spark JVM side. However, for
// some cases e.g., scalar subquery, Spark will not fold it, so we need to handle it
// here.
let array = scalar.to_array()?;
let scalar =
ScalarValue::try_from_array(&cast_array(array, data_type, cast_options)?, 0)?;
- Ok(ColumnarValue::Scalar(scalar))
+ ColumnarValue::Scalar(scalar)
}
- }
+ };
+
+ // let result_type = match &result {
+ // ColumnarValue::Array(array) => array.data_type().clone(),
+ // ColumnarValue::Scalar(scalar) => scalar.data_type(),
+ // };
+
+ // println!(
+ // "spark_cast: {} -> {} (requested: {})",
+ // input_type, result_type, data_type
+ // );
Review Comment:
can this be removed now?
--
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: [df52] migration [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2819297807
##########
native/spark-expr/src/conversion_funcs/cast.rs:
##########
@@ -930,7 +951,19 @@ fn cast_array(
)))
}
};
- Ok(spark_cast_postprocess(cast_result?, from_type, to_type))
+ let x = cast_result?;
+ // println!("cast_array BEFORE postprocess:");
+ // println!(" from_type: {}", from_type);
+ // println!(" to_type: {}", to_type);
+ // println!(" intermediate data_type: {}", x.data_type());
Review Comment:
debug logging can be removed?
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2830929790
##########
spark/src/test/scala/org/apache/comet/CometCastSuite.scala:
##########
@@ -1053,7 +1053,8 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
}
test("cast TimestampType to LongType") {
- castTest(generateTimestampsExtended(), DataTypes.LongType)
+ // currently fails on timestamps outside chrono
Review Comment:
double checking if it still the issue after timestamp conversion
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2835104691
##########
dev/diffs/3.5.8.diff:
##########
@@ -2795,7 +2795,7 @@ index d675503a8ba..f220892396e 100644
+ }
Review Comment:
I still don't see any changes other than whitespace here. Can we remove this from the PR?
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [iceberg] [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2854686142
##########
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:
make sense.
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [iceberg] [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2854691199
##########
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:
so we sure that all batches shares the same schema pointer?
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2824703298
##########
native/core/src/execution/operators/iceberg_scan.rs:
##########
@@ -221,15 +218,12 @@ impl IcebergScanMetrics {
/// Wrapper around iceberg-rust's stream that performs schema adaptation.
/// Handles batches from multiple files that may have different Arrow schemas
-/// (metadata, field IDs, etc.). Caches schema adapters by source schema to avoid
-/// recreating them for every batch from the same file.
+/// (metadata, field IDs, etc.).
struct IcebergStreamWrapper<S> {
inner: S,
schema: SchemaRef,
- /// Cached schema adapter with its source schema. Created when schema changes.
- cached_adapter: Option<(SchemaRef, Arc<dyn SchemaMapper>)>,
Review Comment:
added back
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2822869169
##########
native/core/src/parquet/cast_column.rs:
##########
@@ -0,0 +1,639 @@
+// 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 arrow::{
+ array::{
+ make_array, Array, ArrayRef, LargeListArray, ListArray, MapArray, StructArray,
+ TimestampMicrosecondArray, TimestampMillisecondArray,
+ },
+ compute::CastOptions,
+ datatypes::{DataType, FieldRef, Schema, TimeUnit},
+ record_batch::RecordBatch,
+};
+
+use crate::parquet::parquet_support::{spark_parquet_convert, SparkParquetOptions};
+use datafusion::common::format::DEFAULT_CAST_OPTIONS;
+use datafusion::common::Result as DataFusionResult;
+use datafusion::common::ScalarValue;
+use datafusion::logical_expr::ColumnarValue;
+use datafusion::physical_expr::PhysicalExpr;
+use std::{
+ any::Any,
+ fmt::{self, Display},
+ hash::Hash,
+ sync::Arc,
+};
+
+/// Returns true if two DataTypes are structurally equivalent (same data layout)
+/// but may differ in field names within nested types.
+fn types_differ_only_in_field_names(physical: &DataType, logical: &DataType) -> bool {
+ match (physical, logical) {
+ (DataType::List(pf), DataType::List(lf)) => {
+ pf.is_nullable() == lf.is_nullable()
+ && (pf.data_type() == lf.data_type()
+ || types_differ_only_in_field_names(pf.data_type(), lf.data_type()))
+ }
+ (DataType::LargeList(pf), DataType::LargeList(lf)) => {
+ pf.is_nullable() == lf.is_nullable()
+ && (pf.data_type() == lf.data_type()
+ || types_differ_only_in_field_names(pf.data_type(), lf.data_type()))
+ }
+ (DataType::Map(pf, p_sorted), DataType::Map(lf, l_sorted)) => {
+ p_sorted == l_sorted
+ && pf.is_nullable() == lf.is_nullable()
+ && (pf.data_type() == lf.data_type()
+ || types_differ_only_in_field_names(pf.data_type(), lf.data_type()))
+ }
+ (DataType::Struct(pfields), DataType::Struct(lfields)) => {
+ // For Struct types, field names are semantically meaningful (they
+ // identify different columns), so we require name equality here.
+ // This distinguishes from List/Map wrapper field names ("item" vs
+ // "element") which are purely cosmetic.
+ pfields.len() == lfields.len()
+ && pfields.iter().zip(lfields.iter()).all(|(pf, lf)| {
+ pf.name() == lf.name()
+ && pf.is_nullable() == lf.is_nullable()
+ && (pf.data_type() == lf.data_type()
+ || types_differ_only_in_field_names(pf.data_type(), lf.data_type()))
+ })
+ }
+ _ => false,
+ }
+}
+
+/// Recursively relabel an array so its DataType matches `target_type`.
+/// This only changes metadata (field names, nullability flags in nested fields);
+/// it does NOT change the underlying buffer data.
+fn relabel_array(array: ArrayRef, target_type: &DataType) -> ArrayRef {
+ if array.data_type() == target_type {
+ return array;
+ }
+ match target_type {
+ DataType::List(target_field) => {
+ let list = array.as_any().downcast_ref::<ListArray>().unwrap();
+ let values = relabel_array(Arc::clone(list.values()), target_field.data_type());
+ Arc::new(ListArray::new(
+ Arc::clone(target_field),
+ list.offsets().clone(),
+ values,
+ list.nulls().cloned(),
+ ))
+ }
+ DataType::LargeList(target_field) => {
+ let list = array.as_any().downcast_ref::<LargeListArray>().unwrap();
+ let values = relabel_array(Arc::clone(list.values()), target_field.data_type());
+ Arc::new(LargeListArray::new(
+ Arc::clone(target_field),
+ list.offsets().clone(),
+ values,
+ list.nulls().cloned(),
+ ))
+ }
+ DataType::Map(target_entries_field, sorted) => {
+ let map = array.as_any().downcast_ref::<MapArray>().unwrap();
+ let entries = relabel_array(
+ Arc::new(map.entries().clone()),
+ target_entries_field.data_type(),
+ );
+ let entries_struct = entries.as_any().downcast_ref::<StructArray>().unwrap();
+ Arc::new(MapArray::new(
+ Arc::clone(target_entries_field),
+ map.offsets().clone(),
+ entries_struct.clone(),
+ map.nulls().cloned(),
+ *sorted,
+ ))
+ }
+ DataType::Struct(target_fields) => {
+ let struct_arr = array.as_any().downcast_ref::<StructArray>().unwrap();
+ let columns: Vec<ArrayRef> = target_fields
+ .iter()
+ .zip(struct_arr.columns())
+ .map(|(tf, col)| relabel_array(Arc::clone(col), tf.data_type()))
+ .collect();
+ Arc::new(StructArray::new(
+ target_fields.clone(),
+ columns,
+ struct_arr.nulls().cloned(),
+ ))
+ }
+ // Primitive types - shallow swap is safe
+ _ => {
+ let data = array.to_data();
+ let new_data = data
+ .into_builder()
+ .data_type(target_type.clone())
+ .build()
+ .expect("relabel_array: data layout must be compatible");
+ make_array(new_data)
+ }
+ }
+}
+
+/// Casts a Timestamp(Microsecond) array to Timestamp(Millisecond) by dividing values by 1000.
+/// Preserves the timezone from the target type.
+fn cast_timestamp_micros_to_millis_array(
+ array: &ArrayRef,
+ target_tz: Option<Arc<str>>,
+) -> ArrayRef {
+ let micros_array = array
+ .as_any()
+ .downcast_ref::<TimestampMicrosecondArray>()
+ .expect("Expected TimestampMicrosecondArray");
+
+ let millis_values: TimestampMillisecondArray = micros_array
+ .iter()
+ .map(|opt| opt.map(|v| v / 1000))
+ .collect();
Review Comment:
```suggestion
let millis_values: TimestampMillisecondArray =
arrow::compute::kernels::arity::unary(micros_array, |v| v / 1000);
```
--
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: [df52] migration [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2819550773
##########
native/core/src/parquet/parquet_exec.rs:
##########
@@ -166,27 +203,24 @@ fn get_options(
(table_parquet_options, spark_parquet_options)
}
-fn get_file_config_builder(
- schema: SchemaRef,
- partition_schema: Option<SchemaRef>,
- file_groups: Vec<FileGroup>,
- object_store_url: ObjectStoreUrl,
- file_source: Arc<dyn FileSource>,
-) -> FileScanConfigBuilder {
- match partition_schema {
- Some(partition_schema) => {
- let partition_fields: Vec<Field> = partition_schema
- .fields()
- .iter()
- .map(|field| {
- Field::new(field.name(), field.data_type().clone(), field.is_nullable())
- })
- .collect_vec();
- FileScanConfigBuilder::new(object_store_url, Arc::clone(&schema), file_source)
- .with_file_groups(file_groups)
- .with_table_partition_cols(partition_fields)
+/// Wraps a `SendableRecordBatchStream` to print each batch as it flows through.
+/// Returns a new `SendableRecordBatchStream` that yields the same batches.
+pub fn dbg_batch_stream(stream: SendableRecordBatchStream) -> SendableRecordBatchStream {
Review Comment:
Found it useful for outputting a stream, preserving the original stream to the output. it is not used now, but would be handy for debug purposes.
--
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: [df52] migration [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#issuecomment-3909833801
@sqlbenchmark run tpch
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2822877037
##########
native/spark-expr/src/utils.rs:
##########
@@ -71,6 +72,49 @@ pub fn array_with_timezone(
to_type: Option<&DataType>,
) -> Result<ArrayRef, ArrowError> {
match array.data_type() {
+ DataType::Timestamp(TimeUnit::Millisecond, None) => {
+ assert!(!timezone.is_empty());
+ match to_type {
+ Some(DataType::Utf8) | Some(DataType::Date32) => Ok(array),
+ Some(DataType::Timestamp(_, Some(_))) => {
+ timestamp_ntz_to_timestamp(array, timezone.as_str(), Some(timezone.as_str()))
+ }
+ Some(DataType::Timestamp(TimeUnit::Microsecond, None)) => {
+ // Convert from Timestamp(Millisecond, None) to Timestamp(Microsecond, None)
+ let millis_array = as_primitive_array::<TimestampMillisecondType>(&array);
+ let micros_array: TimestampMicrosecondArray = millis_array
+ .iter()
+ .map(|opt| opt.map(|v| v * 1000))
+ .collect();
Review Comment:
This could use unary kernel instead
--
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: [df52] migration [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#issuecomment-3910056037
> > The changes in this PR are somehow causing large differences in memory usage compared to a recent build from the main branch.
> > Metric Baseline DF52 Delta
> > peak_JVMHeapMemory 6.38 GB 10.60 GB +4.22 GB (+66%)
> > peak_JVMOffHeapMemory 148.01 MB 147.28 MB -0.73 MB (~0%)
> > peak_OnHeapExecutionMemory 0 B 0 B --
> > peak_OffHeapExecutionMemory 14.20 GB 5.30 GB -8.90 GB (-63%)
> > peak_OnHeapUnifiedMemory 74.86 MB 63.63 MB -11.24 MB (-15%)
> > peak_OffHeapUnifiedMemory 14.20 GB 5.30 GB -8.90 GB (-63%)
> > DF52 shows a significant shift from off-heap execution memory to JVM heap. Peak off-heap execution dropped from 14.20 GB to 5.30 GB (-63%), while JVM heap rose from 6.38 GB to 10.60 GB (+66%). Net combined peak memory actually decreased (~20.6 GB to ~15.9 GB).
>
> The total mem used seems to be less? +4 for Heap and -8 for on Heap, so -4 total. The JVM growth looks weird as DF52 has very few JVM changes. Is the baseline for latest main?
The baseline was a build from Friday.
Claude is suggesting this as the root cause:
"The memory shift is primarily driven by DataFusion 52's internal changes to how operators manage buffers (batch coalescing, expression evaluation), combined with these buffers likely not being tracked through Comet's CometUnifiedMemoryPool → JNI → Spark accounting path. The result is that native memory still exists but is "invisible" to Spark's off-heap accounting, while Spark's JVM-side operations expand to fill the perceived available space."
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2822885361
##########
native/core/src/execution/operators/iceberg_scan.rs:
##########
@@ -221,15 +218,12 @@ impl IcebergScanMetrics {
/// Wrapper around iceberg-rust's stream that performs schema adaptation.
/// Handles batches from multiple files that may have different Arrow schemas
-/// (metadata, field IDs, etc.). Caches schema adapters by source schema to avoid
-/// recreating them for every batch from the same file.
+/// (metadata, field IDs, etc.).
struct IcebergStreamWrapper<S> {
inner: S,
schema: SchemaRef,
- /// Cached schema adapter with its source schema. Created when schema changes.
- cached_adapter: Option<(SchemaRef, Arc<dyn SchemaMapper>)>,
Review Comment:
Is caching no longer possible with DF52?
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2823463388
##########
dev/diffs/3.5.8.diff:
##########
@@ -2795,7 +2795,7 @@ index d675503a8ba..f220892396e 100644
+ }
Review Comment:
I think this file just has a whitespace change, can you get the version of the file from main to reduce the diff?
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2829205398
##########
native/core/src/execution/operators/scan.rs:
##########
@@ -582,3 +588,77 @@ impl InputBatch {
InputBatch::Batch(columns, num_rows)
}
}
+
+/// Adapts a Map array to match a target schema's Map type.
+/// This handles the common case where the field names differ (e.g., Parquet uses "key_value"
+/// while Spark uses "entries") but the key/value types are the same.
+/// Returns None if the types are not compatible or not Map types.
+fn adapt_map_to_schema(column: &ArrayRef, target_type: &DataType) -> Option<ArrayRef> {
Review Comment:
does this need to handle nested maps?
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2835128784
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -15,276 +15,482 @@
// specific language governing permissions and limitations
// under the License.
-//! Custom schema adapter that uses Spark-compatible conversions
-
+use crate::parquet::cast_column::CometCastColumnExpr;
use crate::parquet::parquet_support::{spark_parquet_convert, SparkParquetOptions};
-use arrow::array::{RecordBatch, RecordBatchOptions};
-use arrow::datatypes::{Schema, SchemaRef};
-use datafusion::common::ColumnStatistics;
-use datafusion::datasource::schema_adapter::{SchemaAdapter, SchemaAdapterFactory, SchemaMapper};
+use arrow::array::{ArrayRef, RecordBatch};
+use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+use datafusion::common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion::common::Result as DataFusionResult;
+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::{
+ replace_columns_with_literals, DefaultPhysicalExprAdapterFactory, PhysicalExprAdapter,
+ PhysicalExprAdapterFactory,
+};
use std::collections::HashMap;
use std::sync::Arc;
-/// An implementation of DataFusion's `SchemaAdapterFactory` that uses a Spark-compatible
-/// `cast` implementation.
+/// 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 SparkSchemaAdapterFactory {
- /// Spark cast options
+pub struct SparkPhysicalExprAdapterFactory {
+ /// Spark-specific parquet options for type conversions
parquet_options: SparkParquetOptions,
- default_values: Option<HashMap<usize, ScalarValue>>,
+ /// Default values for columns that may be missing from the physical schema.
+ /// The key is the Column (containing name and index).
+ default_values: Option<HashMap<Column, ScalarValue>>,
}
-impl SparkSchemaAdapterFactory {
+impl SparkPhysicalExprAdapterFactory {
+ /// Create a new factory with the given options.
pub fn new(
- options: SparkParquetOptions,
- default_values: Option<HashMap<usize, ScalarValue>>,
+ parquet_options: SparkParquetOptions,
+ default_values: Option<HashMap<Column, ScalarValue>>,
) -> Self {
Self {
- parquet_options: options,
+ parquet_options,
default_values,
}
}
}
-impl SchemaAdapterFactory for SparkSchemaAdapterFactory {
- /// Create a new factory for mapping batches from a file schema to a table
- /// schema.
- ///
- /// This is a convenience for [`DefaultSchemaAdapterFactory::create`] with
- /// the same schema for both the projected table schema and the table
- /// schema.
+/// Remap physical schema field names to match logical schema field names using
+/// case-insensitive matching. This allows the DefaultPhysicalExprAdapter (which
+/// uses exact name matching) to correctly find columns when the parquet file has
Review Comment:
Spark has some configs around case sensitivity for column names, I think. Do we need an issue to respect that, or is it something we cannot respect?
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2835118310
##########
spark/src/main/spark-3.5/org/apache/comet/shims/CometExprShim.scala:
##########
@@ -82,9 +82,14 @@ trait CometExprShim extends CommonStringExprs {
}
case wb: WidthBucket =>
- val childExprs = wb.children.map(exprToProtoInternal(_, inputs, binding))
- val optExpr = scalarFunctionExprToProto("width_bucket", childExprs: _*)
- optExprWithInfo(optExpr, wb, wb.children: _*)
+ withInfo(
+ wb,
+ "WidthBucket not supported, track https://github.com/apache/datafusion/issues/20320")
+ None
+// https://github.com/apache/datafusion/issues/20320
Review Comment:
The linked issue is closed. We should have a Comet issue linked here so we know what to do with this.
--
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: [df52] migration [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#issuecomment-3909951581
> The changes in this PR are somehow causing large differences in memory usage compared to a recent build from the main branch.
>
> Metric Baseline DF52 Delta
> peak_JVMHeapMemory 6.38 GB 10.60 GB +4.22 GB (+66%)
> peak_JVMOffHeapMemory 148.01 MB 147.28 MB -0.73 MB (~0%)
> peak_OnHeapExecutionMemory 0 B 0 B --
> peak_OffHeapExecutionMemory 14.20 GB 5.30 GB -8.90 GB (-63%)
> peak_OnHeapUnifiedMemory 74.86 MB 63.63 MB -11.24 MB (-15%)
> peak_OffHeapUnifiedMemory 14.20 GB 5.30 GB -8.90 GB (-63%)
> DF52 shows a significant shift from off-heap execution memory to JVM heap. Peak off-heap execution dropped from 14.20 GB to 5.30 GB (-63%), while JVM heap rose from 6.38 GB to 10.60 GB (+66%). Net combined peak memory actually decreased (~20.6 GB to ~15.9 GB).
The total mem used seems to be less? +4 for Heap and -8 for on Heap, so -4 total. The JVM growth looks weird as DF52 has very few JVM changes. Is the baseline for latest main?
--
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: [df52] migration [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#issuecomment-3910028761
not sure how true this is yet, but Claude claims that we there may be a regression related to Iceberg scans:
```
In iceberg_scan.rs, the old code cached (SchemaRef, Arc<dyn SchemaMapper>) and reused it across batches with the same schema. The new adapt_batch_with_expressions() creates a new SparkPhysicalExprAdapterFactory + adapter + expression trees for every batch that needs adaptation. This adds per-batch allocation churn, though the actual data buffers are shared via Arc.
```
--
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: [df52] migration [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#issuecomment-3909774128
@sqlbenchmark run tpch
--
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: [df52] migration [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#issuecomment-3910015613
@sqlbenchmark run tpch --iterations 3
--
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: [df52] migration [datafusion-comet]
Posted by "sqlbenchmark (via GitHub)" <gi...@apache.org>.
sqlbenchmark commented on PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#issuecomment-3910251376
## Comet TPC-H Benchmark Results
**Baseline:** `main` (`e22f35c`)
**PR:** `749a407` - DataFusion 52 migration
**Scale Factor:** SF100
**Iterations:** 3
### Query Times
| Query | Baseline Avg (s) | Baseline Best (s) | PR Avg (s) | PR Best (s) | Change (Avg) | Change (Best) |
|-------|-----------------|-------------------|-----------|------------|-------------|--------------|
| Q1 | 10.03 | 9.46 | 10.02 | 9.43 | :white_circle: -0.2% | :white_circle: -0.3% |
| Q2 | 5.43 | 5.11 | 5.34 | 4.98 | :white_circle: -1.7% | :white_circle: -2.7% |
| Q3 | 9.96 | 9.79 | 10.01 | 9.82 | :white_circle: +0.5% | :white_circle: +0.3% |
| Q4 | 10.95 | 10.72 | 13.36 | 12.60 | :red_circle: +21.9% | :red_circle: +17.6% |
| Q5 | 19.33 | 19.03 | 19.36 | 18.80 | :white_circle: +0.2% | :white_circle: -1.2% |
| Q6 | 2.46 | 2.32 | 2.51 | 2.32 | :white_circle: +1.8% | :white_circle: -0.1% |
| Q7 | 12.46 | 12.06 | 12.24 | 11.74 | :white_circle: -1.7% | :white_circle: -2.7% |
| Q8 | 25.65 | 25.10 | 25.47 | 24.41 | :white_circle: -0.7% | :white_circle: -2.7% |
| Q9 | 41.14 | 39.22 | 40.66 | 38.29 | :white_circle: -1.2% | :white_circle: -2.3% |
| Q10 | 10.63 | 10.18 | 11.11 | 10.84 | :white_circle: +4.5% | :red_circle: +6.5% |
| Q11 | 4.41 | 4.38 | 4.75 | 4.52 | :red_circle: +7.7% | :white_circle: +3.2% |
| Q12 | 7.37 | 7.25 | 6.60 | 6.44 | :green_circle: -10.5% | :green_circle: -11.1% |
| Q13 | 7.43 | 7.29 | 7.51 | 7.23 | :white_circle: +1.2% | :white_circle: -0.8% |
| Q14 | 3.56 | 3.51 | 3.39 | 3.27 | :white_circle: -4.9% | :green_circle: -7.0% |
| Q15 | 7.23 | 7.07 | 7.35 | 7.13 | :white_circle: +1.7% | :white_circle: +0.9% |
| Q16 | 4.39 | 4.25 | 4.36 | 4.27 | :white_circle: -0.7% | :white_circle: +0.4% |
| Q17 | 33.93 | 32.87 | 34.23 | 32.74 | :white_circle: +0.9% | :white_circle: -0.4% |
| Q18 | 34.33 | 33.82 | 35.75 | 35.14 | :white_circle: +4.2% | :white_circle: +3.9% |
| Q19 | 7.00 | 6.73 | 6.48 | 6.25 | :green_circle: -7.4% | :green_circle: -7.1% |
| Q20 | 6.68 | 6.52 | 6.47 | 6.33 | :white_circle: -3.1% | :white_circle: -3.0% |
| Q21 | 46.87 | 46.40 | 52.85 | 51.53 | :red_circle: +12.8% | :red_circle: +11.1% |
| Q22 | 5.02 | 5.00 | 5.52 | 5.42 | :red_circle: +10.0% | :red_circle: +8.4% |
| **Total** | **316.29** | **308.08** | **325.36** | **313.50** | :white_circle: +2.9% | :white_circle: +1.8% |
<details>
<summary><b>Spark Configuration</b></summary>
| Setting | Value |
|---------|-------|
| Spark Master | `local[*]` |
| Driver Memory | 32G |
| Driver Cores | 8 |
| Executor Memory | 32G |
| Executor Cores | 8 |
| Off-Heap Enabled | true |
| Off-Heap Size | 24g |
| Shuffle Manager | `CometShuffleManager` |
| Comet Replace SMJ | true |
</details>
---
*Automated benchmark run by dfbench*
--
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: [df52] migration [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#issuecomment-3910404829
I ran the queries individually and compared memory usage between main and this PR.
Key findings from Claude analysis of the results:
1. The memory shift is NOT consistent — it's highly query-dependent. Some queries see off-heap decrease (Q4, Q10, Q11), others see large increases (Q7, Q12, Q13). There is no single directional trend.
2. Off-heap and JVM heap sometimes move inversely. Q11 is the clearest example: off-heap dropped 56.4% while JVM heap increased 127%. Q10 shows the same pattern (off-heap -72.7%, heap +36.5%). DF52 appears to shift work between native and JVM memory for certain query shapes.
3. Join-heavy queries are most affected. The queries with the largest memory changes (Q4, Q7, Q10, Q11, Q12, Q13, Q21) all involve complex joins, correlated subqueries, or GROUP BY with HAVING. Simpler scan-and-aggregate queries (Q1, Q6) are stable. This points to changes in DataFusion 52's hash join/aggregate memory management.
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2830977351
##########
native/core/src/execution/operators/scan.rs:
##########
@@ -582,3 +588,77 @@ impl InputBatch {
InputBatch::Batch(columns, num_rows)
}
}
+
+/// Adapts a Map array to match a target schema's Map type.
+/// This handles the common case where the field names differ (e.g., Parquet uses "key_value"
+/// while Spark uses "entries") but the key/value types are the same.
+/// Returns None if the types are not compatible or not Map types.
+fn adapt_map_to_schema(column: &ArrayRef, target_type: &DataType) -> Option<ArrayRef> {
Review Comment:
removed it, as it repeated another cast
--
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: [df52] migration [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2819299162
##########
native/spark-expr/src/utils.rs:
##########
@@ -127,6 +171,7 @@ pub fn array_with_timezone(
}
fn datetime_cast_err(value: i64) -> ArrowError {
+ println!("{}", std::backtrace::Backtrace::force_capture());
Review Comment:
debug logging can be removed?
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2829208906
##########
spark/src/test/scala/org/apache/comet/CometCastSuite.scala:
##########
@@ -1053,7 +1053,8 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
}
test("cast TimestampType to LongType") {
- castTest(generateTimestampsExtended(), DataTypes.LongType)
+ // currently fails on timestamps outside chrono
Review Comment:
is there a follow on issue for this?
--
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: [df52] migration [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#issuecomment-3909895669
The changes in this PR are somehow causing large differences in memory usage compared to a recent build from the main branch.
| Metric | Baseline | DF52 | Delta |
|---|---|---|---|
| peak_JVMHeapMemory | 6.38 GB | 10.60 GB | +4.22 GB (+66%) |
| peak_JVMOffHeapMemory | 148.01 MB | 147.28 MB | -0.73 MB (~0%) |
| peak_OnHeapExecutionMemory | 0 B | 0 B | -- |
| peak_OffHeapExecutionMemory | 14.20 GB | 5.30 GB | -8.90 GB (-63%) |
| peak_OnHeapUnifiedMemory | 74.86 MB | 63.63 MB | -11.24 MB (-15%) |
| peak_OffHeapUnifiedMemory | 14.20 GB | 5.30 GB | -8.90 GB (-63%) |
--
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: [df52] migration [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#issuecomment-3910545337
I ran a memory profiling comparison to main with replaceSortMergeJoin=false as well.
| Query | BL Time (s) | DF52 Time (s) | Time Delta | BL OffHeap Exec | DF52 OffHeap Exec | OffHeap Delta | BL JVM Heap | DF52 JVM Heap | Heap Delta |
|-------|-------------|---------------|------------|-----------------|-------------------|---------------|-------------|---------------|------------|
| Q1 | 82.6 | 82.5 | -0.2% | 514 MB | 514 MB | 0.0% | 5.53 GB | 4.15 GB | -25.0% |
| Q2 | 15.9 | 15.6 | -1.8% | n/a | n/a | n/a | n/a | 2.84 GB | n/a |
| Q3 | 30.0 | 30.1 | +0.3% | 3.75 GB | 1.69 GB | -54.9% | 4.37 GB | 3.55 GB | -18.8% |
| Q4 | 24.9 | 25.1 | +0.9% | 16.00 GB | 16.00 GB | 0.0% | 2.52 GB | 2.15 GB | -14.8% |
| Q5 | 58.2 | 57.3 | -1.4% | 3.51 GB | 3.00 GB | -14.5% | 3.72 GB | 5.55 GB | +49.3% |
| Q6 | 4.8 | 5.0 | +5.3% | n/a | n/a | n/a | n/a | n/a | n/a |
| Q7 | 28.5 | 27.7 | -2.5% | 2.62 GB | n/a | n/a | 6.55 GB | 6.76 GB | +3.3% |
| Q8 | 42.8 | 41.9 | -2.1% | 657 MB | n/a | n/a | 7.01 GB | 6.12 GB | -12.6% |
| Q9 | 86.1 | 86.6 | +0.6% | 3.00 GB | 3.00 GB | 0.0% | 5.39 GB | 6.56 GB | +21.7% |
| Q10 | 24.1 | 25.0 | +3.7% | 3.04 GB | 1.43 GB | -53.0% | 2.29 GB | 1.25 GB | -45.5% |
| Q11 | 17.2 | 17.4 | +1.2% | 1.00 GB | 2.00 GB | +100.0% | 2.82 GB | 3.79 GB | +34.5% |
| Q12 | 16.7 | 16.3 | -2.7% | 644 MB | n/a | n/a | 4.37 GB | n/a | n/a |
| Q13 | 25.2 | 24.8 | -1.7% | 5.16 GB | 5.13 GB | -0.5% | 3.05 GB | 4.06 GB | +33.1% |
| Q14 | 8.6 | 8.7 | +0.9% | n/a | n/a | n/a | n/a | n/a | n/a |
| Q15 | 19.6 | 19.5 | -0.4% | 460 MB | 520 MB | +12.9% | 1.81 GB | 2.13 GB | +17.2% |
| Q16 | 11.4 | 10.6 | -6.7% | n/a | 2.00 GB | n/a | n/a | 3.69 GB | n/a |
| Q17 | 77.7 | 78.0 | +0.4% | 2.03 GB | 2.02 GB | -0.8% | 4.44 GB | 5.24 GB | +17.9% |
| Q18 | 82.7 | 82.1 | -0.8% | 3.19 GB | 4.75 GB | +49.0% | 5.96 GB | 5.91 GB | -0.9% |
| Q19 | 9.6 | 9.6 | -0.5% | n/a | n/a | n/a | n/a | n/a | n/a |
| Q20 | 14.4 | 14.5 | +0.8% | n/a | n/a | n/a | 3.44 GB | n/a | n/a |
| Q21 | 79.1 | 78.6 | -0.7% | 5.13 GB | 5.25 GB | +2.4% | 4.58 GB | 4.93 GB | +7.7% |
| Q22 | 12.6 | 12.6 | -0.4% | n/a | n/a | n/a | 1.95 GB | n/a | n/a |
| Total | 772.6 | 769.4 | -0.4% | | | | | | |
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2823286441
##########
native/core/src/execution/operators/iceberg_scan.rs:
##########
@@ -221,15 +218,12 @@ impl IcebergScanMetrics {
/// Wrapper around iceberg-rust's stream that performs schema adaptation.
/// Handles batches from multiple files that may have different Arrow schemas
-/// (metadata, field IDs, etc.). Caches schema adapters by source schema to avoid
-/// recreating them for every batch from the same file.
+/// (metadata, field IDs, etc.).
struct IcebergStreamWrapper<S> {
inner: S,
schema: SchemaRef,
- /// Cached schema adapter with its source schema. Created when schema changes.
- cached_adapter: Option<(SchemaRef, Arc<dyn SchemaMapper>)>,
Review Comment:
checking this
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2830810113
##########
native/core/src/execution/operators/scan.rs:
##########
@@ -582,3 +588,77 @@ impl InputBatch {
InputBatch::Batch(columns, num_rows)
}
}
+
+/// Adapts a Map array to match a target schema's Map type.
+/// This handles the common case where the field names differ (e.g., Parquet uses "key_value"
+/// while Spark uses "entries") but the key/value types are the same.
+/// Returns None if the types are not compatible or not Map types.
+fn adapt_map_to_schema(column: &ArrayRef, target_type: &DataType) -> Option<ArrayRef> {
Review Comment:
I'm actually thinking this should be part of schema rewrite rather than standalone cast. checking 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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#issuecomment-3928604107
@andygrove @mbutrovich
May I ask you for the second round of review?
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2835128784
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -15,276 +15,482 @@
// specific language governing permissions and limitations
// under the License.
-//! Custom schema adapter that uses Spark-compatible conversions
-
+use crate::parquet::cast_column::CometCastColumnExpr;
use crate::parquet::parquet_support::{spark_parquet_convert, SparkParquetOptions};
-use arrow::array::{RecordBatch, RecordBatchOptions};
-use arrow::datatypes::{Schema, SchemaRef};
-use datafusion::common::ColumnStatistics;
-use datafusion::datasource::schema_adapter::{SchemaAdapter, SchemaAdapterFactory, SchemaMapper};
+use arrow::array::{ArrayRef, RecordBatch};
+use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+use datafusion::common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion::common::Result as DataFusionResult;
+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::{
+ replace_columns_with_literals, DefaultPhysicalExprAdapterFactory, PhysicalExprAdapter,
+ PhysicalExprAdapterFactory,
+};
use std::collections::HashMap;
use std::sync::Arc;
-/// An implementation of DataFusion's `SchemaAdapterFactory` that uses a Spark-compatible
-/// `cast` implementation.
+/// 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 SparkSchemaAdapterFactory {
- /// Spark cast options
+pub struct SparkPhysicalExprAdapterFactory {
+ /// Spark-specific parquet options for type conversions
parquet_options: SparkParquetOptions,
- default_values: Option<HashMap<usize, ScalarValue>>,
+ /// Default values for columns that may be missing from the physical schema.
+ /// The key is the Column (containing name and index).
+ default_values: Option<HashMap<Column, ScalarValue>>,
}
-impl SparkSchemaAdapterFactory {
+impl SparkPhysicalExprAdapterFactory {
+ /// Create a new factory with the given options.
pub fn new(
- options: SparkParquetOptions,
- default_values: Option<HashMap<usize, ScalarValue>>,
+ parquet_options: SparkParquetOptions,
+ default_values: Option<HashMap<Column, ScalarValue>>,
) -> Self {
Self {
- parquet_options: options,
+ parquet_options,
default_values,
}
}
}
-impl SchemaAdapterFactory for SparkSchemaAdapterFactory {
- /// Create a new factory for mapping batches from a file schema to a table
- /// schema.
- ///
- /// This is a convenience for [`DefaultSchemaAdapterFactory::create`] with
- /// the same schema for both the projected table schema and the table
- /// schema.
+/// Remap physical schema field names to match logical schema field names using
+/// case-insensitive matching. This allows the DefaultPhysicalExprAdapter (which
+/// uses exact name matching) to correctly find columns when the parquet file has
Review Comment:
Spark has some configs around case sensitivity for column names, I think. Do we need an issue to respect that, or is it something we cannot respect?
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2835120386
##########
spark/src/test/scala/org/apache/comet/CometMathExpressionSuite.scala:
##########
@@ -105,7 +106,8 @@ class CometMathExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe
}
}
- test("width_bucket - edge cases") {
+ // https://github.com/apache/datafusion/issues/20320
Review Comment:
Same comment about the closed PR.
##########
spark/src/test/scala/org/apache/comet/CometMathExpressionSuite.scala:
##########
@@ -122,7 +124,8 @@ class CometMathExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe
}
}
- test("width_bucket - NaN values") {
+ // https://github.com/apache/datafusion/issues/20320
Review Comment:
Same comment about the closed PR.
##########
spark/src/test/scala/org/apache/comet/CometMathExpressionSuite.scala:
##########
@@ -134,7 +137,8 @@ class CometMathExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe
}
}
- test("width_bucket - with range data") {
+ // https://github.com/apache/datafusion/issues/20320
Review Comment:
Same comment about the closed PR.
--
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] deps: DataFusion 52.0.0 migration (SchemaAdapter changes, etc.) [iceberg] [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #3536:
URL: https://github.com/apache/datafusion-comet/pull/3536#discussion_r2854726641
##########
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:
Filed #3602
--
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