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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
