You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "snmvaughan (via GitHub)" <gi...@apache.org> on 2026/10/22 21:54:58 UTC
[PR] fix: don't treat planner-inserted casts as Parquet schema conversions [datafusion-comet]
snmvaughan opened a new pull request, #7159:
URL: https://github.com/apache/datafusion-comet/pull/7159
## Which issue does this PR close?
Closes #7158.
## Rationale for this change
The native Parquet scan's expression adapter (`SparkPhysicalExprAdapter`) rewrites every DataFusion `CastExpr` it finds in a pushed-down predicate as though it were a file-versus-table schema conversion, and applies Spark's Parquet read rules to it. A predicate can also contain `CastExpr`s that have nothing to do with the file schema: Comet's planner builds them with `CastExpr::new` to coerce scalar-function arguments to the types the function requires. `substring(s, 1, 2)` is one. Spark sends Int32 literals, `SparkSubstring` takes Int64 positions, so the planner wraps them in `CAST(1 AS Int64)` and `CAST(2 AS Int64)`.
On Spark 3.x (`allow_type_promotion = false`) the adapter reports that cast as a disallowed `INT32 -> bigint` Parquet promotion. The check is deferred to runtime, so it only fires when the predicate is evaluated against a decoded batch, that is with `spark.comet.parquet.rowFilterPushdown.enabled=true`:
```
Error evaluating filter predicate: External(ParquetSchemaConvert { file_path: "", column: "[]", physical_type: "INT32", spark_type: "bigint" })
```
The file's schema matches the requested schema exactly. The empty column name is the tell: the cast is not over a column.
With the flag off (the default) the predicate is only used for row-group and page pruning, so the problem stays hidden. It surfaced as TPC-H Q22, the only TPC-H query that uses `substring`, failing on Spark 3.4 when the TPC-H suite ran with row-level filter pushdown enabled. On Spark 4.x `allow_type_promotion` is true, so this particular cast is accepted, but the planner's cast is still silently swapped for a Spark `Cast`.
## What changes are included in this PR?
`replace_with_spark_cast` now leaves a cast alone unless it reconciles a file column with its table-schema type (new `is_schema_adaptation_cast`). DataFusion's `DefaultPhysicalExprAdapter` builds those with `CastExpr::new_with_target_field`, passing the table schema's field, so the target carries the column's name. The planner's `CastExpr::new` casts have an unnamed target field.
One behavior change: casts the planner inserted into a pushed-down predicate now stay DataFusion `CastExpr`s (with the planner's `safe: false` options) on every Spark version, instead of being rewritten into Spark `Cast`s on Spark 4.x. That matches how the same expression is evaluated when it is not pushed into the scan.
The check keys on the target field's name because `CastExpr` offers nothing else to tell the two apart in the DataFusion version used here. A unit test pins that naming convention, so a DataFusion upgrade that changes it fails loudly rather than silently reintroducing the problem.
## How are these changes tested?
New tests in `schema_adapter.rs`:
- `schema_adaptation_cast_is_told_apart_by_target_field_name` pins the naming convention the check relies on.
- `pushed_substring_filter_*` build `substring(s, 1, 2) = 'ab'` through `PhysicalPlanner` from the proto `CometSubstring` serializes, then scan a Parquet file whose schema equals the requested schema: without row-level pushdown (control), with it using Spark 3.x settings (this fails without the change, with the error above), and with it using Spark 4.x settings.
The existing tests for genuine schema mismatches (for example INT32 read as bigint) are unchanged and still pass, because the adapter names those casts after the column. The `datafusion-comet` lib tests pass (`cargo test -p datafusion-comet --lib`), and `cargo fmt --check` and `cargo clippy --all-targets -- -D warnings` are clean.
--
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] fix: don't treat planner-inserted casts as Parquet schema conversions [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #7159:
URL: https://github.com/apache/datafusion-comet/pull/7159#discussion_r4354855054
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -429,6 +429,23 @@ fn is_string_or_binary(dt: &DataType) -> bool {
)
}
+/// Whether `cast` reconciles a column's Parquet (file) type with its table-schema type, as opposed
+/// to being part of the pushed-down expression itself.
+///
+/// DataFusion's `DefaultPhysicalExprAdapter` builds the former with
+/// `CastExpr::new_with_target_field`, passing the table schema's field, so the target is named
+/// after the column (or nested field). Comet's planner builds the latter with `CastExpr::new`, for
+/// example to coerce a scalar function's arguments to the types the function requires:
+/// `substring(s, 1, 2)` casts its Int32 literals to Int64. `CastExpr::new` produces an unnamed
+/// target field.
+///
+/// Only the former are Parquet read conversions that Spark's vectorized reader validates. Applying
+/// those rules to the latter rejects valid queries, for example as a disallowed `INT32 -> bigint`
+/// promotion on Spark 3.x when row-level filter pushdown evaluates the predicate.
+fn is_schema_adaptation_cast(cast: &datafusion::physical_expr::expressions::CastExpr) -> bool {
+ !cast.target_field().name().is_empty()
+}
Review Comment:
A concrete fix for the empty-name case sunchao raised. `CastExpr::has_explicit_metadata()` exists in the pinned DataFusion 55.2.0 (`physical-expr/src/expressions/cast.rs`, it returns the `explicit_target` flag that only `new_with_target_field` sets, and `with_new_children` preserves it). The default adapter builds its schema casts with `new_with_target_field` and Comet's planner uses `CastExpr::new`, so `!cast.has_explicit_metadata()` at the call site would separate the two without relying on the field name, and `is_schema_adaptation_cast` could go. I read this from the sources and did not run it.
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -2350,6 +2375,220 @@ pub(crate) mod test {
Ok(())
}
+ /// `is_schema_adaptation_cast` tells the casts DataFusion's default adapter inserts apart from
+ /// the casts Comet's planner puts in a predicate by whether the target field is named. If a
+ /// DataFusion upgrade changes how `CastExpr::new` names its target, this fails and the check
+ /// has to be revisited (newer DataFusion versions also expose an explicit-target marker).
Review Comment:
This comment says newer DataFusion versions expose an explicit-target marker, but 55.2.0 already has `has_explicit_metadata()`. The PR description makes the same claim. The test `schema_adaptation_cast_is_told_apart_by_target_field_name` also pins a naming convention that DataFusion pins itself. If the helper is replaced, a regression test with a `Field::new("", ..)` target built through `new_with_target_field` would be more useful here, together with a scan-level case where a file `Int64` column named `""` read as `Int32` must still raise `ParquetSchemaConvert`.
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -2350,6 +2375,220 @@ pub(crate) mod test {
Ok(())
}
+ /// `is_schema_adaptation_cast` tells the casts DataFusion's default adapter inserts apart from
+ /// the casts Comet's planner puts in a predicate by whether the target field is named. If a
+ /// DataFusion upgrade changes how `CastExpr::new` names its target, this fails and the check
+ /// has to be revisited (newer DataFusion versions also expose an explicit-target marker).
+ #[test]
+ fn schema_adaptation_cast_is_told_apart_by_target_field_name() {
+ use super::is_schema_adaptation_cast;
+ use datafusion::physical_expr::expressions::CastExpr;
+
+ // Built by the planner, from a bare type.
+ let planner_cast = CastExpr::new(
+ Arc::new(Literal::new(ScalarValue::Int32(Some(1)))),
+ DataType::Int64,
+ None,
+ );
+ assert!(!is_schema_adaptation_cast(&planner_cast));
+
+ // Built by the default adapter, from the table schema's field.
+ let adapter_cast = CastExpr::new_with_target_field(
+ Arc::new(Column::new("a", 0)),
+ Arc::new(Field::new("a", DataType::Int64, true)),
+ None,
+ );
+ assert!(is_schema_adaptation_cast(&adapter_cast));
+ }
+
+ /// Scans a Parquet file with a pushed-down `substring(s, 1, 2) = 'ab'` data filter, the way
+ /// `CometNativeScan` sends it, and returns the values the scan produced.
+ ///
+ /// The file schema is identical to the requested schema, so there is no schema evolution of any
+ /// kind. The predicate comes from the real planner, fed the proto `CometSubstring` emits (Int32
+ /// literals for the position and length). `SparkSubstring` takes Int64 positions, so the planner
+ /// coerces those literals with `CastExpr::new`.
+ async fn scan_with_substring_filter(
Review Comment:
`scan_with_substring_filter` rebuilds the Parquet writer, `ParquetSource`, `FileScanConfigBuilder`, adapter factory and `DataSourceExec` setup that `scan_parquet_file` (line 2634) already provides, and that helper is already used with `row_filters = true` and a predicate in the tests around lines 3559 to 3694. The three `pushed_substring_filter_*` tests also differ only in `(row_filter_pushdown, allow_type_promotion)` and the expected values, and only the middle one fails without the fix, so one loop with a comment on the controls would do. The `try_pushdown_filters` route matches `parquet_exec.rs`, so that part may need to stay. I have not run `scan_parquet_file` with a `CAST(id AS Int64) = lit` predicate to see whether it reproduces the failure.
--
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] fix: don't treat planner-inserted casts as Parquet schema conversions [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove merged PR #7159:
URL: https://github.com/apache/datafusion-comet/pull/7159
--
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] fix: don't treat planner-inserted casts as Parquet schema conversions [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #7159:
URL: https://github.com/apache/datafusion-comet/pull/7159#discussion_r4354015278
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -429,6 +429,23 @@ fn is_string_or_binary(dt: &DataType) -> bool {
)
}
+/// Whether `cast` reconciles a column's Parquet (file) type with its table-schema type, as opposed
+/// to being part of the pushed-down expression itself.
+///
+/// DataFusion's `DefaultPhysicalExprAdapter` builds the former with
+/// `CastExpr::new_with_target_field`, passing the table schema's field, so the target is named
+/// after the column (or nested field). Comet's planner builds the latter with `CastExpr::new`, for
+/// example to coerce a scalar function's arguments to the types the function requires:
+/// `substring(s, 1, 2)` casts its Int32 literals to Int64. `CastExpr::new` produces an unnamed
+/// target field.
+///
+/// Only the former are Parquet read conversions that Spark's vectorized reader validates. Applying
+/// those rules to the latter rejects valid queries, for example as a disallowed `INT32 -> bigint`
+/// promotion on Spark 3.x when row-level filter pushdown evaluates the predicate.
+fn is_schema_adaptation_cast(cast: &datafusion::physical_expr::expressions::CastExpr) -> bool {
+ !cast.target_field().name().is_empty()
Review Comment:
[P2] Preserve schema validation for empty-named columns. Spark supports a Parquet column named `""`, so an empty target name does not identify a planner cast. For a file field `Field("", Int64)` containing `[1, 2]`, requested as `Field("", Int32)`, the default adapter creates a genuine schema cast with an empty target name. This predicate now skips `check_conversion`, and the native scan returns `[1, 2]` instead of Spark's schema-conversion error. This occurs with row-filter pushdown disabled and with either promotion setting. Could this use DataFusion 55.2.0's existing explicit-target accessor, such as `has_explicit_metadata()`, and retain a regression test for empty field names?
Evidence: Ran a disposable native test through the existing `scan_parquet` helper. With `allow_type_promotion=false` and `true`, column `a` produced `ParquetSchemaConvert { physical_type: "INT64", spark_type: "int" }`, while column `""` returned `Int32 [1, 2]`. The native run reported 92 passed and only this regression test failed. Stock Spark 3.5.9 successfully wrote and read an empty-named BIGINT column containing 1, then rejected an IntegerType read with `SchemaColumnConvertNotSupportedException: column: [], physicalType: INT64, logicalType: int`. The base implementation reaches `check_conversion` without the new name guard. Reproduction source and logs are retained in `/tmp/comet-review-7159-a1350a73/`.
--
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] fix: don't treat planner-inserted casts as Parquet schema conversions [datafusion-comet]
Posted by "snmvaughan (via GitHub)" <gi...@apache.org>.
snmvaughan commented on PR #7159:
URL: https://github.com/apache/datafusion-comet/pull/7159#issuecomment-6367779480
Thanks @andygrove, @sunchao and @comphead. a5cbcd5 addresses the review:
- `replace_with_spark_cast` checks `CastExpr::has_explicit_metadata()` instead of the target field's name, so an empty-named column keeps the read-conversion check. `is_schema_adaptation_cast` is gone.
- `empty_named_column_long_read_as_int_errors`: an `INT64` column named `""` read as `int` raises `ParquetSchemaConvert` under both type-promotion settings. It fails on the previous head.
- `schema_casts_are_told_apart_by_explicit_target` replaces the naming-convention test.
- The substring tests are one table-driven test on `scan_parquet_file`.
--
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] fix: don't treat planner-inserted casts as Parquet schema conversions [datafusion-comet]
Posted by "snmvaughan (via GitHub)" <gi...@apache.org>.
snmvaughan commented on code in PR #7159:
URL: https://github.com/apache/datafusion-comet/pull/7159#discussion_r4360898491
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -429,6 +429,23 @@ fn is_string_or_binary(dt: &DataType) -> bool {
)
}
+/// Whether `cast` reconciles a column's Parquet (file) type with its table-schema type, as opposed
+/// to being part of the pushed-down expression itself.
+///
+/// DataFusion's `DefaultPhysicalExprAdapter` builds the former with
+/// `CastExpr::new_with_target_field`, passing the table schema's field, so the target is named
+/// after the column (or nested field). Comet's planner builds the latter with `CastExpr::new`, for
+/// example to coerce a scalar function's arguments to the types the function requires:
+/// `substring(s, 1, 2)` casts its Int32 literals to Int64. `CastExpr::new` produces an unnamed
+/// target field.
+///
+/// Only the former are Parquet read conversions that Spark's vectorized reader validates. Applying
+/// those rules to the latter rejects valid queries, for example as a disallowed `INT32 -> bigint`
+/// promotion on Spark 3.x when row-level filter pushdown evaluates the predicate.
+fn is_schema_adaptation_cast(cast: &datafusion::physical_expr::expressions::CastExpr) -> bool {
+ !cast.target_field().name().is_empty()
Review Comment:
Thanks for catching this, and for checking it against Spark 3.5.9. I reproduced it: with the name check, an `INT64` column named `""` read as `int` returned `[1, 2]`. Fixed in a5cbcd5. `replace_with_spark_cast` now checks `CastExpr::has_explicit_metadata()` instead of the target name, and the new `empty_named_column_long_read_as_int_errors` covers the empty-named column under both type-promotion settings. It fails on the previous head and passes now.
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -429,6 +429,23 @@ fn is_string_or_binary(dt: &DataType) -> bool {
)
}
+/// Whether `cast` reconciles a column's Parquet (file) type with its table-schema type, as opposed
+/// to being part of the pushed-down expression itself.
+///
+/// DataFusion's `DefaultPhysicalExprAdapter` builds the former with
+/// `CastExpr::new_with_target_field`, passing the table schema's field, so the target is named
+/// after the column (or nested field). Comet's planner builds the latter with `CastExpr::new`, for
+/// example to coerce a scalar function's arguments to the types the function requires:
+/// `substring(s, 1, 2)` casts its Int32 literals to Int64. `CastExpr::new` produces an unnamed
+/// target field.
+///
+/// Only the former are Parquet read conversions that Spark's vectorized reader validates. Applying
+/// those rules to the latter rejects valid queries, for example as a disallowed `INT32 -> bigint`
+/// promotion on Spark 3.x when row-level filter pushdown evaluates the predicate.
+fn is_schema_adaptation_cast(cast: &datafusion::physical_expr::expressions::CastExpr) -> bool {
+ !cast.target_field().name().is_empty()
+}
Review Comment:
Thanks, that's what a5cbcd5 does. I checked it in DataFusion 55.2.0: the default adapter builds its schema casts with `new_with_target_field` (`schema_rewriter.rs`), Comet only calls `CastExpr::new`, and `with_new_children` keeps the flag. `is_schema_adaptation_cast` is gone.
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -2350,6 +2375,220 @@ pub(crate) mod test {
Ok(())
}
+ /// `is_schema_adaptation_cast` tells the casts DataFusion's default adapter inserts apart from
+ /// the casts Comet's planner puts in a predicate by whether the target field is named. If a
+ /// DataFusion upgrade changes how `CastExpr::new` names its target, this fails and the check
+ /// has to be revisited (newer DataFusion versions also expose an explicit-target marker).
Review Comment:
Right, 55.2.0 already has it. The comment and the naming-convention test are gone. `schema_casts_are_told_apart_by_explicit_target` pins the flag instead, including for a `Field::new("", ..)` target built with `new_with_target_field` and after `with_new_children`, and the scan-level case is `empty_named_column_long_read_as_int_errors`. I've corrected the PR description too.
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -2350,6 +2375,220 @@ pub(crate) mod test {
Ok(())
}
+ /// `is_schema_adaptation_cast` tells the casts DataFusion's default adapter inserts apart from
+ /// the casts Comet's planner puts in a predicate by whether the target field is named. If a
+ /// DataFusion upgrade changes how `CastExpr::new` names its target, this fails and the check
+ /// has to be revisited (newer DataFusion versions also expose an explicit-target marker).
+ #[test]
+ fn schema_adaptation_cast_is_told_apart_by_target_field_name() {
+ use super::is_schema_adaptation_cast;
+ use datafusion::physical_expr::expressions::CastExpr;
+
+ // Built by the planner, from a bare type.
+ let planner_cast = CastExpr::new(
+ Arc::new(Literal::new(ScalarValue::Int32(Some(1)))),
+ DataType::Int64,
+ None,
+ );
+ assert!(!is_schema_adaptation_cast(&planner_cast));
+
+ // Built by the default adapter, from the table schema's field.
+ let adapter_cast = CastExpr::new_with_target_field(
+ Arc::new(Column::new("a", 0)),
+ Arc::new(Field::new("a", DataType::Int64, true)),
+ None,
+ );
+ assert!(is_schema_adaptation_cast(&adapter_cast));
+ }
+
+ /// Scans a Parquet file with a pushed-down `substring(s, 1, 2) = 'ab'` data filter, the way
+ /// `CometNativeScan` sends it, and returns the values the scan produced.
+ ///
+ /// The file schema is identical to the requested schema, so there is no schema evolution of any
+ /// kind. The predicate comes from the real planner, fed the proto `CometSubstring` emits (Int32
+ /// literals for the position and length). `SparkSubstring` takes Int64 positions, so the planner
+ /// coerces those literals with `CastExpr::new`.
+ async fn scan_with_substring_filter(
Review Comment:
Good call. `scan_parquet_file` does reproduce it: with the check disabled, the row-filter case fails with the original `ParquetSchemaConvert { column: "[]", physical_type: "INT32", spark_type: "bigint" }`. So the custom scan setup is gone, and the three tests are one table-driven `pushed_substring_filter_keeps_planner_casts`, with comments on the two controls.
--
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