Repository navigation
fix: [branch-1.1] don't treat planner-inserted casts as Parquet schema conversions (#7159) - #7264
Merged
Conversation
…apache#7159) * fix: don't treat planner-inserted casts as Parquet schema conversions The native Parquet scan's expression adapter rewrites every DataFusion `CastExpr` 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 casts 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, i.e. 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. `replace_with_spark_cast` now leaves a cast alone unless it reconciles a file column with its table-schema type (`is_schema_adaptation_cast`). DataFusion's default adapter 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. Casts the planner inserted therefore stay DataFusion `CastExpr`s on every Spark version, instead of being rewritten into Spark `Cast`s on Spark 4.x. * fix: find schema-adaptation casts by explicit target, not by name Spark allows a Parquet column named "", so an empty target name does not mark a planner cast: an INT64 column named "" read as int skipped the read-conversion check and returned its values where Spark raises SchemaColumnConvertNotSupportedException. DataFusion's default adapter builds its schema casts with CastExpr::new_with_target_field, which sets the explicit-target flag that CastExpr::has_explicit_metadata() returns, also for an empty name, and with_new_children keeps it. Comet's planner builds the casts it puts in a predicate with CastExpr::new, which doesn't set it. replace_with_spark_cast now checks that flag, and is_schema_adaptation_cast is gone. Tests: an INT64 column named "" read as int is rejected under both type-promotion settings, the naming-convention test is replaced by one that pins the explicit-target flag, and the three substring filter tests are now one table-driven test on the existing scan_parquet_file helper, which still fails with the original INT32 -> bigint error without the fix. Adapted for branch-1.1: the scan test's ScalarFunc drops null_short_circuit, which branch-1.1's proto doesn't have. (cherry picked from commit 01fc37a)
github-actions
Bot
added
bug
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Backport of #7159 (squash commit 01fc37a) to
branch-1.1, for #7158.Rationale for this change
branch-1.1has the bug #7159 fixes onmain. Withspark.comet.parquet.rowFilterPushdown.enabled=trueon Spark 3.x, a pushed-down filter that contains a cast Comet's planner inserted, such as the Int64 positions ofsubstring(s, 1, 2), fails withParquetSchemaConvert { physical_type: "INT32", spark_type: "bigint" }. The native scan's expression adapter treats every cast in the predicate as a Parquet schema conversion. See #7158 for the details.What changes are included in this PR?
A cherry-pick of #7159 with
-x, adapted in one test line: the substring scan test builds its protoScalarFuncwithoutnull_short_circuit, whichbranch-1.1's proto doesn't have. The fix itself is identical to the source commit.replace_with_spark_castnow rewrites only the casts DataFusion's schema adapter builds, which set the explicit-target flag thatCastExpr::has_explicit_metadata()returns, and leaves the planner's casts alone.Backports to other branches:
branch-1.2, #7262.How are these changes tested?
The tests come with the cherry-pick, in
native/core/src/parquet/schema_adapter.rs. On this branch all 96schema_adaptertests pass, andcargo fmt --checkandcargo clippy --all-targets --workspace -- -D warningsare clean.