alamb commented on code in PR #24125:
URL: https://github.com/apache/datafusion/pull/24125#discussion_r3731842298
##########
datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt:
##########
@@ -125,20 +125,23 @@ explain analyze select s from full_schema;
----
Plan with Metrics DataSourceExec: <slt:ignore>metrics=[output_rows=3,
<slt:ignore>bytes_scanned=219<slt:ignore>]
-# `get_field` on a schema-narrowed struct becomes `get_field(CAST(s), 'x')`;
-# the read clips to the cast target (every field the *narrow* schema
-# declares), not further down to just `x`. The fair "nothing was clipped"
-# baseline is therefore reading every physical leaf of `s`
-# (`select s from full_schema` above), not the same `get_field` query against
-# `full_schema` -- that one needs no cast at all and takes `get_field`'s own,
-# more precise, single-leaf pushdown path.
+# `get_field` on a schema-narrowed struct is rewritten to
+# `CAST(get_field(s, 'x'))` rather than `get_field(CAST(s), 'x')`, so it takes
+# `get_field`'s own single-leaf pushdown path: the read clips all the way down
Review Comment:
I am not sure what 'get_fields own single-leaf pushdown path is' (It seems
like a bunch of implementation detail -- can we just clarify that this query
should read fewer bytes because it is selecting a field of s (not all the
fields) ?
##########
datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt:
##########
@@ -125,20 +125,23 @@ explain analyze select s from full_schema;
----
Plan with Metrics DataSourceExec: <slt:ignore>metrics=[output_rows=3,
<slt:ignore>bytes_scanned=219<slt:ignore>]
-# `get_field` on a schema-narrowed struct becomes `get_field(CAST(s), 'x')`;
-# the read clips to the cast target (every field the *narrow* schema
-# declares), not further down to just `x`. The fair "nothing was clipped"
-# baseline is therefore reading every physical leaf of `s`
-# (`select s from full_schema` above), not the same `get_field` query against
-# `full_schema` -- that one needs no cast at all and takes `get_field`'s own,
-# more precise, single-leaf pushdown path.
+# `get_field` on a schema-narrowed struct is rewritten to
+# `CAST(get_field(s, 'x'))` rather than `get_field(CAST(s), 'x')`, so it takes
+# `get_field`'s own single-leaf pushdown path: the read clips all the way down
+# to `x`, not just to the fields the *narrow* schema declares. That is why
+# this reads fewer bytes than `select s from narrow` above, which still needs
+# every narrow leaf.
query TT
explain analyze select s['x'] from narrow;
----
-Plan with Metrics DataSourceExec: <slt:ignore>metrics=[output_rows=3,
<slt:ignore>bytes_scanned=146<slt:ignore>]
+Plan with Metrics DataSourceExec: <slt:ignore>metrics=[output_rows=3,
<slt:ignore>bytes_scanned=75<slt:ignore>]
# Mixed access -- the whole (narrowed) column and a subfield of it -- still
-# reads only the narrow schema's leaves.
+# reads only the narrow schema's leaves. The whole-column read goes through
Review Comment:
I am not sure what this comment is trying to say -- it seems like it is
trying to explain implementation details. I think we can remove it from here
unless it is adding crititcal context past this PR
##########
datafusion/physical-expr-adapter/src/schema_rewriter.rs:
##########
@@ -1426,6 +1572,368 @@ mod tests {
// datafusion/core/tests/parquet/schema_adapter.rs provide better
coverage for this functionality.
}
+ /// Build `get_field(column, 'field')` against `schema`.
+ fn get_field_expr(
+ schema: &Schema,
+ column: &str,
+ field: &str,
+ ) -> Arc<dyn PhysicalExpr> {
+ let index = schema.index_of(column).unwrap();
+ Arc::new(
+ ScalarFunctionExpr::try_new(
+
Arc::new(datafusion_expr::ScalarUDF::from(GetFieldFunc::new())),
+ vec![
+ Arc::new(Column::new(column, index)),
+ Arc::new(Literal::new(ScalarValue::from(field))),
+ ],
+ schema,
+ Arc::new(datafusion_common::config::ConfigOptions::default()),
+ )
+ .unwrap(),
+ )
+ }
+
+ fn struct_schemas(
+ physical_fields: Vec<Field>,
+ logical_fields: Vec<Field>,
+ ) -> (SchemaRef, SchemaRef) {
+ let physical = Arc::new(Schema::new(vec![Field::new(
+ "s",
+ DataType::Struct(physical_fields.into()),
+ true,
+ )]));
+ let logical = Arc::new(Schema::new(vec![Field::new(
+ "s",
+ DataType::Struct(logical_fields.into()),
+ true,
+ )]));
+ (logical, physical)
+ }
+
+ /// `s['x']` where the file stores `x` as `Int32` and the table declares
Review Comment:
do these unit tests add any additional coverage compared with the `.slt`
coverage? I think the slt coverage is adequate and we could remove these tests
and make the PR much smaller
##########
datafusion/physical-expr-adapter/src/schema_rewriter.rs:
##########
@@ -282,13 +311,130 @@ impl DefaultPhysicalExprAdapterRewriter {
return Ok(Transformed::yes(transformed));
}
+ if let Some(transformed) = self.try_narrow_struct_cast(&expr)? {
+ return Ok(Transformed::yes(transformed));
+ }
+
if let Some(column) = expr.downcast_ref::<Column>() {
return self.rewrite_column(Arc::clone(&expr), column);
}
Ok(Transformed::no(expr))
}
+ /// Rewrite `get_field(cast(s AS Struct<..>), 'f')` into
+ /// `cast(get_field(s, 'f') AS <type of f>)`.
+ ///
+ /// Expressions are rewritten bottom-up, so by the time we reach a
+ /// `get_field` node its struct argument has already been wrapped in a cast
+ /// by [`Self::rewrite_column`] whenever the logical and physical struct
+ /// types differ. Casting the whole struct just to read one field is
+ /// wasteful, and — more importantly — it hides the underlying column from
+ /// consumers that pattern match on `get_field(column, 'f')`. The Parquet
Review Comment:
the silently dropping the predicate part I think is an implementation detail
that may not be relevant in the future.
##########
datafusion/physical-expr-adapter/src/schema_rewriter.rs:
##########
@@ -282,13 +311,130 @@ impl DefaultPhysicalExprAdapterRewriter {
return Ok(Transformed::yes(transformed));
}
+ if let Some(transformed) = self.try_narrow_struct_cast(&expr)? {
+ return Ok(Transformed::yes(transformed));
+ }
+
if let Some(column) = expr.downcast_ref::<Column>() {
return self.rewrite_column(Arc::clone(&expr), column);
}
Ok(Transformed::no(expr))
}
+ /// Rewrite `get_field(cast(s AS Struct<..>), 'f')` into
Review Comment:
> did you consider teaching the pushdown side (PushdownChecker / row-filter
builder) to see through the cast — i.e. recognize get_field(cast(col), 'f') —
instead of narrowing it here?
I agree that this fix seems very specific -- it almost seems like it is
focusing on the symptom rather than the underlying problem (the physical /
logical mismatch that @zhuqi-lucas is pointing out)
It seems like if this is a rewrite that should be done, shouldn't we be
doing at a higher level 🤔
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -890,6 +890,105 @@ set datafusion.execution.parquet.pushdown_filters = false;
statement ok
DROP TABLE t_struct_filter;
+##########
+# Regression test for https://github.com/apache/datafusion/issues/24109
+#
+# When the declared table schema differs from the physical file schema, the
Review Comment:
I think the old behavior is not very relevant after this PR -- maybe this
could just focus on what this case covers-- namely that the declared table
schema differs from the physical schema
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -890,6 +890,105 @@ set datafusion.execution.parquet.pushdown_filters = false;
statement ok
DROP TABLE t_struct_filter;
+##########
+# Regression test for https://github.com/apache/datafusion/issues/24109
+#
+# When the declared table schema differs from the physical file schema, the
+# expression adapter inserts a cast. Casting the whole struct column hid the
+# column from the row filter builder: the scan reported `s['x'] = 200` as
+# fully handled (so `FilterExec` was removed from the plan) but then could not
+# build the row filter, silently dropping the predicate and returning all rows.
+##########
+
+statement ok
+set datafusion.execution.parquet.pushdown_filters = true;
+
+statement ok
+COPY (
+ SELECT
+ column1 as id,
+ named_struct('x', arrow_cast(column2, 'Int32')) as s
+ FROM VALUES (1, 100), (2, 200), (3, 300)
+) TO 'test_files/scratch/parquet_filter_pushdown/struct_schema_cast.parquet'
+STORED AS PARQUET;
+
+# `x` is stored as Int32 in the file but declared as BIGINT here, which forces
+# the scan to adapt the struct column.
+statement ok
+CREATE EXTERNAL TABLE t_struct_schema_cast (id BIGINT, s STRUCT<x BIGINT>)
+STORED AS PARQUET
+LOCATION
'test_files/scratch/parquet_filter_pushdown/struct_schema_cast.parquet';
+
+query II
+SELECT id, s['x'] FROM t_struct_schema_cast WHERE s['x'] = 200;
+----
+2 200
+
+# Conjunction of a struct-field filter and a primitive filter.
+query II
+SELECT id, s['x'] FROM t_struct_schema_cast WHERE s['x'] > 100 AND id > 2;
+----
+3 300
+
+# Control: the same file read through a schema that matches it exactly, so no
+# cast is inserted. This path has always worked.
+statement ok
+CREATE EXTERNAL TABLE t_struct_no_schema_cast (id BIGINT, s STRUCT<x INT>)
+STORED AS PARQUET
+LOCATION
'test_files/scratch/parquet_filter_pushdown/struct_schema_cast.parquet';
+
+query II
+SELECT id, s['x'] FROM t_struct_no_schema_cast WHERE s['x'] = 200;
Review Comment:
can we also run the query too:
```sql
SELECT id, s['x'] FROM t_struct_schema_cast WHERE s['x'] > 100 AND id > 2;
```
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -890,6 +890,105 @@ set datafusion.execution.parquet.pushdown_filters = false;
statement ok
DROP TABLE t_struct_filter;
+##########
+# Regression test for https://github.com/apache/datafusion/issues/24109
+#
+# When the declared table schema differs from the physical file schema, the
+# expression adapter inserts a cast. Casting the whole struct column hid the
+# column from the row filter builder: the scan reported `s['x'] = 200` as
+# fully handled (so `FilterExec` was removed from the plan) but then could not
+# build the row filter, silently dropping the predicate and returning all rows.
+##########
+
+statement ok
+set datafusion.execution.parquet.pushdown_filters = true;
+
+statement ok
+COPY (
+ SELECT
+ column1 as id,
+ named_struct('x', arrow_cast(column2, 'Int32')) as s
+ FROM VALUES (1, 100), (2, 200), (3, 300)
+) TO 'test_files/scratch/parquet_filter_pushdown/struct_schema_cast.parquet'
+STORED AS PARQUET;
+
+# `x` is stored as Int32 in the file but declared as BIGINT here, which forces
+# the scan to adapt the struct column.
+statement ok
+CREATE EXTERNAL TABLE t_struct_schema_cast (id BIGINT, s STRUCT<x BIGINT>)
+STORED AS PARQUET
+LOCATION
'test_files/scratch/parquet_filter_pushdown/struct_schema_cast.parquet';
+
+query II
+SELECT id, s['x'] FROM t_struct_schema_cast WHERE s['x'] = 200;
+----
+2 200
+
+# Conjunction of a struct-field filter and a primitive filter.
+query II
+SELECT id, s['x'] FROM t_struct_schema_cast WHERE s['x'] > 100 AND id > 2;
+----
+3 300
+
+# Control: the same file read through a schema that matches it exactly, so no
+# cast is inserted. This path has always worked.
+statement ok
+CREATE EXTERNAL TABLE t_struct_no_schema_cast (id BIGINT, s STRUCT<x INT>)
+STORED AS PARQUET
+LOCATION
'test_files/scratch/parquet_filter_pushdown/struct_schema_cast.parquet';
+
+query II
+SELECT id, s['x'] FROM t_struct_no_schema_cast WHERE s['x'] = 200;
+----
+2 200
+
+# A field that the file does not have reads as null, so it filters nothing in.
+statement ok
+CREATE EXTERNAL TABLE t_struct_missing_field (id BIGINT, s STRUCT<x BIGINT,
missing BIGINT>)
+STORED AS PARQUET
+LOCATION
'test_files/scratch/parquet_filter_pushdown/struct_schema_cast.parquet';
+
+query II
+SELECT id, s['x'] FROM t_struct_missing_field WHERE s['missing'] = 200;
Review Comment:
how about also running the two queries above and verifying that they still
get the right answer even when there are new fields inserted?
##########
datafusion/physical-expr-adapter/src/schema_rewriter.rs:
##########
@@ -1426,6 +1572,368 @@ mod tests {
// datafusion/core/tests/parquet/schema_adapter.rs provide better
coverage for this functionality.
}
+ /// Build `get_field(column, 'field')` against `schema`.
+ fn get_field_expr(
+ schema: &Schema,
+ column: &str,
+ field: &str,
+ ) -> Arc<dyn PhysicalExpr> {
+ let index = schema.index_of(column).unwrap();
+ Arc::new(
+ ScalarFunctionExpr::try_new(
+
Arc::new(datafusion_expr::ScalarUDF::from(GetFieldFunc::new())),
+ vec![
+ Arc::new(Column::new(column, index)),
+ Arc::new(Literal::new(ScalarValue::from(field))),
+ ],
+ schema,
+ Arc::new(datafusion_common::config::ConfigOptions::default()),
+ )
+ .unwrap(),
+ )
+ }
+
+ fn struct_schemas(
+ physical_fields: Vec<Field>,
+ logical_fields: Vec<Field>,
+ ) -> (SchemaRef, SchemaRef) {
+ let physical = Arc::new(Schema::new(vec![Field::new(
+ "s",
+ DataType::Struct(physical_fields.into()),
+ true,
+ )]));
+ let logical = Arc::new(Schema::new(vec![Field::new(
+ "s",
+ DataType::Struct(logical_fields.into()),
+ true,
+ )]));
+ (logical, physical)
+ }
+
+ /// `s['x']` where the file stores `x` as `Int32` and the table declares
+ /// `Int64` must cast the extracted field, not the whole struct, so that
+ /// the column stays visible under the `get_field`.
+ ///
+ /// See <https://github.com/apache/datafusion/issues/24109>.
+ #[test]
+ fn test_narrow_struct_cast_to_field_access() {
+ let (logical_schema, physical_schema) = struct_schemas(
+ vec![Field::new("x", DataType::Int32, true)],
+ vec![Field::new("x", DataType::Int64, true)],
+ );
+
+ let adapter = DefaultPhysicalExprAdapterFactory
Review Comment:
there is a lot of boiler plate here (factor creation rewrite, cast, etc) --
maybe it could be factored into a helper so it is clearer what is being tested
and what is setup
##########
datafusion/physical-expr-adapter/src/schema_rewriter.rs:
##########
@@ -282,13 +311,130 @@ impl DefaultPhysicalExprAdapterRewriter {
return Ok(Transformed::yes(transformed));
}
+ if let Some(transformed) = self.try_narrow_struct_cast(&expr)? {
+ return Ok(Transformed::yes(transformed));
+ }
+
if let Some(column) = expr.downcast_ref::<Column>() {
return self.rewrite_column(Arc::clone(&expr), column);
}
Ok(Transformed::no(expr))
}
+ /// Rewrite `get_field(cast(s AS Struct<..>), 'f')` into
+ /// `cast(get_field(s, 'f') AS <type of f>)`.
Review Comment:
Can we also describe here the rationale for **why** we would want to do this
rewrite? It is not obvious I think from these commens
--
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]
