You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "adriangb (via GitHub)" <gi...@apache.org> on 2025/11/30 08:44:33 UTC
[PR] PhysicalExprAdapterFactory handle struct columns and remove SchemaAdapterFactory from row filter [datafusion]
adriangb opened a new pull request, #18998:
URL: https://github.com/apache/datafusion/pull/18998
Chips away at https://github.com/apache/datafusion/issues/14993 and https://github.com/apache/datafusion/issues/16800
--
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] Cut Parquet over to PhysicalExprAdapter from SchemaAdapter [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#issuecomment-3593375230
๐ค `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running
Linux aal-dev 6.14.0-1018-gcp #19~24.04.1-Ubuntu SMP Wed Sep 24 23:23:09 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing remove-schema-row-filter (5710a0bf396b4ec6a5588e637460a245d63e0488) to 1f03ac502f48f205a6eb6a69d408f58ffd792879 [diff](https://github.com/apache/datafusion/compare/1f03ac502f48f205a6eb6a69d408f58ffd792879..5710a0bf396b4ec6a5588e637460a245d63e0488) using: tpch_mem clickbench_partitioned clickbench_extended
Results will be posted here when complete
--
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] PhysicalExprAdapterFactory handle struct columns and remove SchemaAdapterFactory from row filter [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#discussion_r2573516918
##########
datafusion/datasource-parquet/src/opener.rs:
##########
@@ -1275,143 +1273,4 @@ mod test {
}
}
}
-
- #[tokio::test]
- async fn test_custom_schema_adapter_no_rewriter() {
- // Make a hardcoded schema adapter that adds a new column "b" with default value 0.0
- // and converts the first column "a" from Int32 to UInt64.
Review Comment:
Same here I think this test is no longer relevant
##########
datafusion/core/tests/parquet/schema_adapter.rs:
##########
@@ -293,36 +293,8 @@ async fn test_custom_schema_adapter_and_custom_expression_adapter() {
];
assert_batches_eq!(expected, &batches);
- // Test using a custom schema adapter and no explicit physical expr adapter
- // This should use the custom schema adapter both for projections and predicate pushdown
Review Comment:
Removing this because it's no longer true. SchemaAdapter is not used for predicate pushdown.
##########
datafusion/physical-expr/src/expressions/cast.rs:
##########
@@ -740,6 +740,101 @@ mod tests {
Ok(())
}
+ #[test]
+ fn test_cast_timestamp_with_timezone_to_timestamp() -> Result<()> {
Review Comment:
These ended up not being necessary (the bug was elsewhere in my changes) but I think it's good coverage anyway so leaving it in 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] Cut Parquet over to PhysicalExprAdapter from SchemaAdapter [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#issuecomment-3593321626
> run benchmarks
BTW your job is scheduled. My job runner got borked due to something silly (https://github.com/alamb/datafusion-benchmarking/commit/213b4782071771e1e54dad978698ffe7f9d229be). The run should complete in a while
--
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] Cut Parquet over to PhysicalExprAdapter from SchemaAdapter [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#discussion_r2578279131
##########
datafusion/datasource-parquet/src/row_filter.rs:
##########
@@ -270,10 +233,12 @@ impl FilterCandidateBuilder {
.project(&required_indices_into_table_schema)?,
);
- let (schema_mapper, projection_into_file_schema) = self
- .schema_adapter_factory
- .create(Arc::clone(&projected_table_schema), self.table_schema)
- .map_schema(&self.file_schema)?;
+ // Compute the projection into the file schema by matching column names
+ let projection_into_file_schema: Vec<usize> = projected_table_schema
+ .fields()
+ .iter()
+ .filter_map(|f| self.file_schema.index_of(f.name()).ok())
+ .collect();
Review Comment:
I think (1) exists but I will confirm or add it. I will also add (2).
--
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] Cut Parquet over to PhysicalExprAdapter from SchemaAdapter [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#discussion_r2579997429
##########
datafusion/datasource-parquet/src/row_filter.rs:
##########
@@ -270,10 +233,12 @@ impl FilterCandidateBuilder {
.project(&required_indices_into_table_schema)?,
);
- let (schema_mapper, projection_into_file_schema) = self
- .schema_adapter_factory
- .create(Arc::clone(&projected_table_schema), self.table_schema)
- .map_schema(&self.file_schema)?;
+ // Compute the projection into the file schema by matching column names
+ let projection_into_file_schema: Vec<usize> = projected_table_schema
+ .fields()
+ .iter()
+ .filter_map(|f| self.file_schema.index_of(f.name()).ok())
+ .collect();
Review Comment:
66df4f27f67d285aaf50dd0e0208bca3b06a5c7c and 091cbbd9e2318172d6a0725c63bb8f8b0039f3bf
--
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] Cut Parquet over to PhysicalExprAdapter from SchemaAdapter [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#discussion_r2589975202
##########
datafusion/datasource-parquet/src/row_filter.rs:
##########
@@ -270,10 +233,12 @@ impl FilterCandidateBuilder {
.project(&required_indices_into_table_schema)?,
);
- let (schema_mapper, projection_into_file_schema) = self
- .schema_adapter_factory
- .create(Arc::clone(&projected_table_schema), self.table_schema)
- .map_schema(&self.file_schema)?;
+ // Compute the projection into the file schema by matching column names
+ let projection_into_file_schema: Vec<usize> = projected_table_schema
+ .fields()
+ .iter()
+ .filter_map(|f| self.file_schema.index_of(f.name()).ok())
+ .collect();
Review Comment:
How about 3cd7c84 ?
--
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] Cut Parquet over to PhysicalExprAdapter from SchemaAdapter [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#issuecomment-3593580776
๐ค: Benchmark completed
<details><summary>Details</summary>
<p>
```
Comparing HEAD and remove-schema-row-filter
--------------------
Benchmark clickbench_extended.json
--------------------
โโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโ
โ Query โ HEAD โ remove-schema-row-filter โ Change โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ QQuery 0 โ 2621.88 ms โ 2655.35 ms โ no change โ
โ QQuery 1 โ 1310.68 ms โ 1218.84 ms โ +1.08x faster โ
โ QQuery 2 โ 2459.78 ms โ 2417.48 ms โ no change โ
โ QQuery 3 โ 1140.63 ms โ 1147.50 ms โ no change โ
โ QQuery 4 โ 2341.78 ms โ 2317.89 ms โ no change โ
โ QQuery 5 โ 28444.58 ms โ 28356.08 ms โ no change โ
โ QQuery 6 โ 4139.85 ms โ 4242.22 ms โ no change โ
โ QQuery 7 โ 3488.21 ms โ 3476.70 ms โ no change โ
โโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโ
โ Benchmark Summary โ โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ Total Time (HEAD) โ 45947.39ms โ
โ Total Time (remove-schema-row-filter) โ 45832.06ms โ
โ Average Time (HEAD) โ 5743.42ms โ
โ Average Time (remove-schema-row-filter) โ 5729.01ms โ
โ Queries Faster โ 1 โ
โ Queries Slower โ 0 โ
โ Queries with No Change โ 7 โ
โ Queries with Failure โ 0 โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโ
--------------------
Benchmark clickbench_partitioned.json
--------------------
โโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโ
โ Query โ HEAD โ remove-schema-row-filter โ Change โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ QQuery 0 โ 2.20 ms โ 2.20 ms โ no change โ
โ QQuery 1 โ 48.91 ms โ 48.79 ms โ no change โ
โ QQuery 2 โ 138.02 ms โ 135.95 ms โ no change โ
โ QQuery 3 โ 165.97 ms โ 164.59 ms โ no change โ
โ QQuery 4 โ 1073.83 ms โ 1066.07 ms โ no change โ
โ QQuery 5 โ 1501.42 ms โ 1517.40 ms โ no change โ
โ QQuery 6 โ 2.13 ms โ 2.17 ms โ no change โ
โ QQuery 7 โ 55.93 ms โ 53.67 ms โ no change โ
โ QQuery 8 โ 1433.32 ms โ 1418.65 ms โ no change โ
โ QQuery 9 โ 1862.66 ms โ 1842.93 ms โ no change โ
โ QQuery 10 โ 385.91 ms โ 375.00 ms โ no change โ
โ QQuery 11 โ 447.05 ms โ 422.90 ms โ +1.06x faster โ
โ QQuery 12 โ 1335.70 ms โ 1373.84 ms โ no change โ
โ QQuery 13 โ 2152.49 ms โ 2125.07 ms โ no change โ
โ QQuery 14 โ 1287.97 ms โ 1302.17 ms โ no change โ
โ QQuery 15 โ 1224.50 ms โ 1216.62 ms โ no change โ
โ QQuery 16 โ 2734.26 ms โ 2703.42 ms โ no change โ
โ QQuery 17 โ 2680.01 ms โ 2680.94 ms โ no change โ
โ QQuery 18 โ 5038.98 ms โ 4977.56 ms โ no change โ
โ QQuery 19 โ 132.20 ms โ 127.47 ms โ no change โ
โ QQuery 20 โ 1981.07 ms โ 1999.59 ms โ no change โ
โ QQuery 21 โ 2287.76 ms โ 2338.12 ms โ no change โ
โ QQuery 22 โ 3920.27 ms โ 4012.02 ms โ no change โ
โ QQuery 23 โ 13216.89 ms โ 13123.84 ms โ no change โ
โ QQuery 24 โ 215.43 ms โ 213.78 ms โ no change โ
โ QQuery 25 โ 474.90 ms โ 485.86 ms โ no change โ
โ QQuery 26 โ 218.52 ms โ 228.08 ms โ no change โ
โ QQuery 27 โ 2822.31 ms โ 2860.85 ms โ no change โ
โ QQuery 28 โ 23568.19 ms โ 23842.02 ms โ no change โ
โ QQuery 29 โ 982.52 ms โ 981.22 ms โ no change โ
โ QQuery 30 โ 1360.61 ms โ 1369.62 ms โ no change โ
โ QQuery 31 โ 1419.96 ms โ 1372.23 ms โ no change โ
โ QQuery 32 โ 5349.19 ms โ 4871.98 ms โ +1.10x faster โ
โ QQuery 33 โ 6056.90 ms โ 5840.10 ms โ no change โ
โ QQuery 34 โ 6011.50 ms โ 6033.00 ms โ no change โ
โ QQuery 35 โ 1899.02 ms โ 1854.38 ms โ no change โ
โ QQuery 36 โ 121.54 ms โ 119.47 ms โ no change โ
โ QQuery 37 โ 52.05 ms โ 52.13 ms โ no change โ
โ QQuery 38 โ 121.62 ms โ 120.55 ms โ no change โ
โ QQuery 39 โ 197.54 ms โ 195.53 ms โ no change โ
โ QQuery 40 โ 42.40 ms โ 43.96 ms โ no change โ
โ QQuery 41 โ 40.63 ms โ 38.96 ms โ no change โ
โ QQuery 42 โ 33.94 ms โ 32.79 ms โ no change โ
โโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโ
โ Benchmark Summary โ โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ Total Time (HEAD) โ 96098.23ms โ
โ Total Time (remove-schema-row-filter) โ 95587.47ms โ
โ Average Time (HEAD) โ 2234.84ms โ
โ Average Time (remove-schema-row-filter) โ 2222.96ms โ
โ Queries Faster โ 2 โ
โ Queries Slower โ 0 โ
โ Queries with No Change โ 41 โ
โ Queries with Failure โ 0 โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโ
--------------------
Benchmark tpch_mem_sf1.json
--------------------
โโโโโโโโโโโโโโโโณโโโโโโโโโโโโณโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโ
โ Query โ HEAD โ remove-schema-row-filter โ Change โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ QQuery 1 โ 139.30 ms โ 127.06 ms โ +1.10x faster โ
โ QQuery 2 โ 28.81 ms โ 28.36 ms โ no change โ
โ QQuery 3 โ 37.31 ms โ 39.11 ms โ no change โ
โ QQuery 4 โ 28.97 ms โ 29.15 ms โ no change โ
โ QQuery 5 โ 86.51 ms โ 87.73 ms โ no change โ
โ QQuery 6 โ 19.66 ms โ 19.51 ms โ no change โ
โ QQuery 7 โ 224.04 ms โ 223.65 ms โ no change โ
โ QQuery 8 โ 34.59 ms โ 34.25 ms โ no change โ
โ QQuery 9 โ 100.77 ms โ 112.50 ms โ 1.12x slower โ
โ QQuery 10 โ 64.59 ms โ 63.38 ms โ no change โ
โ QQuery 11 โ 17.44 ms โ 18.47 ms โ 1.06x slower โ
โ QQuery 12 โ 53.25 ms โ 53.65 ms โ no change โ
โ QQuery 13 โ 46.49 ms โ 45.69 ms โ no change โ
โ QQuery 14 โ 13.83 ms โ 14.01 ms โ no change โ
โ QQuery 15 โ 24.88 ms โ 25.03 ms โ no change โ
โ QQuery 16 โ 25.30 ms โ 25.21 ms โ no change โ
โ QQuery 17 โ 153.48 ms โ 152.36 ms โ no change โ
โ QQuery 18 โ 279.38 ms โ 284.93 ms โ no change โ
โ QQuery 19 โ 38.06 ms โ 37.27 ms โ no change โ
โ QQuery 20 โ 51.36 ms โ 50.02 ms โ no change โ
โ QQuery 21 โ 327.71 ms โ 338.82 ms โ no change โ
โ QQuery 22 โ 17.27 ms โ 17.44 ms โ no change โ
โโโโโโโโโโโโโโโโดโโโโโโโโโโโโดโโโโโโโโโโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโ
โ Benchmark Summary โ โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ Total Time (HEAD) โ 1812.98ms โ
โ Total Time (remove-schema-row-filter) โ 1827.57ms โ
โ Average Time (HEAD) โ 82.41ms โ
โ Average Time (remove-schema-row-filter) โ 83.07ms โ
โ Queries Faster โ 1 โ
โ Queries Slower โ 2 โ
โ Queries with No Change โ 19 โ
โ Queries with Failure โ 0 โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโ
```
</p>
</details>
--
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] Cut Parquet over to PhysicalExprAdapter from SchemaAdapter [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#discussion_r2590672147
##########
datafusion/pruning/src/pruning_predicate.rs:
##########
@@ -1105,6 +1106,20 @@ fn rewrite_expr_to_prunable(
None,
));
Ok((left, op, right))
+ } else if let Some(cast_col) = column_expr_any.downcast_ref::<CastColumnExpr>() {
Review Comment:
This is a good example where the code for `CastExpr` and `CastColumnExpr` are almost identical and the duplication seems unfortunate
I took an initial shot at removing the redundancy here
- https://github.com/apache/datafusion/pull/19097
##########
datafusion/core/src/datasource/physical_plan/parquet.rs:
##########
@@ -1257,7 +1257,7 @@ mod tests {
("c3", c3.clone()),
]);
- // batch2: c3(int8), c2(int64), c1(string), c4(string)
+ // batch2: c3(date64), c2(int64), c1(string), c4(date64)
Review Comment:
๐ค batch2 only has 3 columns (c3(date64), c2(int64) and c1(string))
```suggestion
// batch2: c3(date64), c2(int64), c1(string)
```
##########
datafusion/core/tests/parquet/schema_adapter.rs:
##########
@@ -293,255 +194,128 @@ async fn test_custom_schema_adapter_and_custom_expression_adapter() {
];
assert_batches_eq!(expected, &batches);
- // Test using a custom schema adapter and no explicit physical expr adapter
- // This should use the custom schema adapter both for projections and predicate pushdown
+ // Test with a custom physical expr adapter
+ // PhysicalExprAdapterFactory now handles both predicates AND projections
+ // CustomPhysicalExprAdapterFactory fills missing columns with 'b' for Utf8
let listing_table_config =
ListingTableConfig::new(ListingTableUrl::parse("memory:///").unwrap())
.infer_options(&ctx.state())
.await
.unwrap()
.with_schema(table_schema.clone())
- .with_schema_adapter_factory(Arc::new(CustomSchemaAdapterFactory));
+ .with_expr_adapter_factory(Arc::new(CustomPhysicalExprAdapterFactory));
let table = ListingTable::try_new(listing_table_config).unwrap();
ctx.deregister_table("t").unwrap();
ctx.register_table("t", Arc::new(table)).unwrap();
let batches = ctx
- .sql("SELECT c2, c1 FROM t WHERE c1 = 2 AND c2 = 'a'")
Review Comment:
it took me a while to understand why this test changed -- it is because `CustomSchemaMapper` used to fill in the value `a` but `CustomPhysicalExprAdapterFactory` fills in `b`.
Therefore this change makes sense to me ๐
##########
datafusion/core/tests/parquet/schema_adapter.rs:
##########
@@ -59,97 +51,6 @@ async fn write_parquet(batch: RecordBatch, store: Arc<dyn ObjectStore>, path: &s
store.put(&Path::from(path), data.into()).await.unwrap();
}
-#[derive(Debug)]
-struct CustomSchemaAdapterFactory;
-
-impl SchemaAdapterFactory for CustomSchemaAdapterFactory {
- fn create(
- &self,
- projected_table_schema: SchemaRef,
- _table_schema: SchemaRef,
- ) -> Box<dyn SchemaAdapter> {
- Box::new(CustomSchemaAdapter {
- logical_file_schema: projected_table_schema,
- })
- }
-}
-
-#[derive(Debug)]
-struct CustomSchemaAdapter {
- logical_file_schema: SchemaRef,
-}
-
-impl SchemaAdapter for CustomSchemaAdapter {
- fn map_column_index(&self, index: usize, file_schema: &Schema) -> Option<usize> {
- for (idx, field) in file_schema.fields().iter().enumerate() {
- if field.name() == self.logical_file_schema.field(index).name() {
- return Some(idx);
- }
- }
- None
- }
-
- fn map_schema(
- &self,
- file_schema: &Schema,
- ) -> Result<(Arc<dyn SchemaMapper>, Vec<usize>)> {
- let projection = (0..file_schema.fields().len()).collect_vec();
- Ok((
- Arc::new(CustomSchemaMapper {
- logical_file_schema: Arc::clone(&self.logical_file_schema),
- }),
- projection,
- ))
- }
-}
-
-#[derive(Debug)]
-struct CustomSchemaMapper {
- logical_file_schema: SchemaRef,
-}
-
-impl SchemaMapper for CustomSchemaMapper {
- fn map_batch(&self, batch: RecordBatch) -> Result<RecordBatch> {
- let mut output_columns =
- Vec::with_capacity(self.logical_file_schema.fields().len());
- for field in self.logical_file_schema.fields() {
- if let Some(array) = batch.column_by_name(field.name()) {
- output_columns.push(cast_with_options(
- array,
- field.data_type(),
- &CastOptions::default(),
- )?);
- } else {
- // Create a new array with the default value for the field type
- let default_value = match field.data_type() {
- DataType::Int64 => ScalarValue::Int64(Some(0)),
- DataType::Utf8 => ScalarValue::Utf8(Some("a".to_string())),
- _ => unimplemented!("Unsupported data type: {}", field.data_type()),
- };
- output_columns
- .push(default_value.to_array_of_size(batch.num_rows()).unwrap());
- }
- }
- let batch = RecordBatch::try_new_with_options(
- Arc::clone(&self.logical_file_schema),
- output_columns,
- &RecordBatchOptions::new().with_row_count(Some(batch.num_rows())),
- )
- .unwrap();
- Ok(batch)
- }
-
- fn map_column_statistics(
- &self,
- _file_col_statistics: &[ColumnStatistics],
- ) -> Result<Vec<ColumnStatistics>> {
- Ok(vec![
- ColumnStatistics::new_unknown();
- self.logical_file_schema.fields().len()
- ])
- }
-}
-
// Implement a custom PhysicalExprAdapterFactory that fills in missing columns with the default value for the field type
Review Comment:
it might help to note here that "default value for the field type" really means:
* `1` for missing Int64 values
* `'b'` for missing Utf8 columns
##########
datafusion/datasource-parquet/src/opener.rs:
##########
@@ -320,13 +314,42 @@ impl FileOpener for ParquetOpener {
reader_metadata,
);
- let (schema_mapping, adapted_projections) =
- schema_adapter.map_schema(&physical_file_schema)?;
+ let mut projection =
+ ProjectionExprs::from_indices(&projection, &logical_file_schema);
+ if let Some(expr_adapter_factory) = expr_adapter_factory {
+ let adapter = expr_adapter_factory
+ .create(
+ Arc::clone(&logical_file_schema),
+ Arc::clone(&physical_file_schema),
+ )
+ .with_partition_values(
+ partition_fields
+ .iter()
+ .cloned()
+ .zip(partitioned_file.partition_values.clone())
+ .collect_vec(),
+ );
+ let projection_expressions = projection
Review Comment:
I found some of this somewhat hard to follow, maybe as a follow on some of it could move to helpers or methods on ProjectionExprs -- for example maybe a function like
```rust
impl ProjectionExprs {
// Update this ProjectionExpr by applying a closure to all the contained expressions
fn map<F>(self, f:F) -> Result<Self> { ... }
}
```
##########
datafusion/datasource-parquet/src/opener.rs:
##########
@@ -464,14 +486,32 @@ impl FileOpener for ParquetOpener {
file_metrics.predicate_cache_inner_records.clone();
let predicate_cache_records = file_metrics.predicate_cache_records.clone();
+ let stream_schema = Arc::clone(stream.schema());
+
+ // Rebase column indices to match the narrowed stream schema.
+ // The projection expressions have indices based on physical_file_schema,
+ // but the stream only contains the columns selected by the ProjectionMask.
+ let rebased_exprs = projection
Review Comment:
here is another example where ProjectionExprs::map or something like that could be used
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -526,43 +499,10 @@ impl FileSource for ParquetSource {
) -> datafusion_common::Result<Arc<dyn FileOpener>> {
let split_projection = self.projection.clone();
- let (expr_adapter_factory, schema_adapter_factory) = match (
Review Comment:
โค๏ธ
##########
datafusion/physical-expr/src/expressions/cast.rs:
##########
@@ -740,6 +740,101 @@ mod tests {
Ok(())
}
+ #[test]
+ fn test_cast_timestamp_with_timezone_to_timestamp() -> Result<()> {
Review Comment:
These seem like they could be just as easily tested with `.slt` / `arrow_cast` calls ๐ค
That would be less verbose and easier to maintain in my opinion
##########
datafusion/datasource-parquet/src/opener.rs:
##########
@@ -320,13 +314,42 @@ impl FileOpener for ParquetOpener {
reader_metadata,
);
- let (schema_mapping, adapted_projections) =
- schema_adapter.map_schema(&physical_file_schema)?;
+ let mut projection =
+ ProjectionExprs::from_indices(&projection, &logical_file_schema);
+ if let Some(expr_adapter_factory) = expr_adapter_factory {
+ let adapter = expr_adapter_factory
+ .create(
+ Arc::clone(&logical_file_schema),
+ Arc::clone(&physical_file_schema),
+ )
+ .with_partition_values(
+ partition_fields
+ .iter()
+ .cloned()
+ .zip(partitioned_file.partition_values.clone())
+ .collect_vec(),
+ );
+ let projection_expressions = projection
+ .as_ref()
+ .iter()
+ .cloned()
+ .map(|mut proj| {
+ proj.expr = adapter.rewrite(Arc::clone(&proj.expr))?;
+ Ok(proj)
+ })
+ .collect::<Result<Vec<_>>>()?;
+ projection = ProjectionExprs::new(projection_expressions);
+ }
+ let indices = projection
Review Comment:
this might be a nice function on ProjectionExprs (the inverse of `ProjectionExprs::from_indices`, it could be `ProjectionExprs::to_indices` or something)
##########
datafusion/core/tests/parquet/schema_adapter.rs:
##########
@@ -293,255 +194,128 @@ async fn test_custom_schema_adapter_and_custom_expression_adapter() {
];
assert_batches_eq!(expected, &batches);
- // Test using a custom schema adapter and no explicit physical expr adapter
- // This should use the custom schema adapter both for projections and predicate pushdown
+ // Test with a custom physical expr adapter
+ // PhysicalExprAdapterFactory now handles both predicates AND projections
+ // CustomPhysicalExprAdapterFactory fills missing columns with 'b' for Utf8
let listing_table_config =
ListingTableConfig::new(ListingTableUrl::parse("memory:///").unwrap())
.infer_options(&ctx.state())
.await
.unwrap()
.with_schema(table_schema.clone())
- .with_schema_adapter_factory(Arc::new(CustomSchemaAdapterFactory));
+ .with_expr_adapter_factory(Arc::new(CustomPhysicalExprAdapterFactory));
let table = ListingTable::try_new(listing_table_config).unwrap();
ctx.deregister_table("t").unwrap();
ctx.register_table("t", Arc::new(table)).unwrap();
let batches = ctx
- .sql("SELECT c2, c1 FROM t WHERE c1 = 2 AND c2 = 'a'")
+ .sql("SELECT c2, c1 FROM t WHERE c1 = 2 AND c2 = 'b'")
.await
.unwrap()
.collect()
.await
.unwrap();
+ // With CustomPhysicalExprAdapterFactory, missing column c2 is filled with 'b'
+ // in both the predicate (c2 = 'b' becomes 'b' = 'b' -> true) and the projection
let expected = [
"+----+----+",
"| c2 | c1 |",
"+----+----+",
- "| a | 2 |",
+ "| b | 2 |",
"+----+----+",
];
assert_batches_eq!(expected, &batches);
+}
+
+/// Test demonstrating how to implement a custom PhysicalExprAdapterFactory
+/// that fills missing columns with non-null default values.
+///
+/// This is the recommended migration path for users who previously used
+/// SchemaAdapterFactory to fill missing columns with default values.
+/// Instead of transforming batches after reading (SchemaAdapter::map_batch),
+/// the PhysicalExprAdapterFactory rewrites expressions to use literals for
+/// missing columns, achieving the same result more efficiently.
+#[tokio::test]
+async fn test_physical_expr_adapter_with_non_null_defaults() {
+ // File only has c1 column
+ let batch = record_batch!(("c1", Int32, [10, 20, 30])).unwrap();
+
+ let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+ let store_url = ObjectStoreUrl::parse("memory://").unwrap();
+ write_parquet(batch, store.clone(), "defaults_test.parquet").await;
+
+ // Table schema has additional columns c2 (Utf8) and c3 (Int64) that don't exist in file
+ let table_schema = Arc::new(Schema::new(vec![
+ Field::new("c1", DataType::Int64, false), // type differs from file (Int32 vs Int64)
+ Field::new("c2", DataType::Utf8, true), // missing from file
+ Field::new("c3", DataType::Int64, true), // missing from file
+ ]));
+
+ let mut cfg = SessionConfig::new()
+ .with_collect_statistics(false)
+ .with_parquet_pruning(false);
+ cfg.options_mut().execution.parquet.pushdown_filters = true;
+ let ctx = SessionContext::new_with_config(cfg);
+ ctx.register_object_store(store_url.as_ref(), Arc::clone(&store));
- // Do the same test but with a custom physical expr adapter
- // Now the default schema adapter will be used for projections, but the custom physical expr adapter will be used for predicate pushdown
+ // CustomPhysicalExprAdapterFactory fills:
+ // - missing Utf8 columns with 'b'
+ // - missing Int64 columns with 1
let listing_table_config =
ListingTableConfig::new(ListingTableUrl::parse("memory:///").unwrap())
.infer_options(&ctx.state())
.await
.unwrap()
.with_schema(table_schema.clone())
.with_expr_adapter_factory(Arc::new(CustomPhysicalExprAdapterFactory));
+
let table = ListingTable::try_new(listing_table_config).unwrap();
- ctx.deregister_table("t").unwrap();
ctx.register_table("t", Arc::new(table)).unwrap();
+
+ // Query all columns - missing columns should have default values
let batches = ctx
- .sql("SELECT c2, c1 FROM t WHERE c1 = 2 AND c2 = 'b'")
+ .sql("SELECT c1, c2, c3 FROM t ORDER BY c1")
.await
.unwrap()
.collect()
.await
.unwrap();
+
+ // c1 is cast from Int32 to Int64, c2 defaults to 'b', c3 defaults to 1
let expected = [
- "+----+----+",
- "| c2 | c1 |",
- "+----+----+",
- "| | 2 |",
- "+----+----+",
+ "+----+----+----+",
+ "| c1 | c2 | c3 |",
+ "+----+----+----+",
+ "| 10 | b | 1 |",
+ "| 20 | b | 1 |",
+ "| 30 | b | 1 |",
+ "+----+----+----+",
];
assert_batches_eq!(expected, &batches);
- // If we use both then the custom physical expr adapter will be used for predicate pushdown and the custom schema adapter will be used for projections
- let listing_table_config =
- ListingTableConfig::new(ListingTableUrl::parse("memory:///").unwrap())
- .infer_options(&ctx.state())
- .await
- .unwrap()
- .with_schema(table_schema.clone())
- .with_schema_adapter_factory(Arc::new(CustomSchemaAdapterFactory))
- .with_expr_adapter_factory(Arc::new(CustomPhysicalExprAdapterFactory));
- let table = ListingTable::try_new(listing_table_config).unwrap();
- ctx.deregister_table("t").unwrap();
- ctx.register_table("t", Arc::new(table)).unwrap();
+ // Verify predicates work with default values
+ // c3 = 1 should match all rows since default is 1
let batches = ctx
- .sql("SELECT c2, c1 FROM t WHERE c1 = 2 AND c2 = 'b'")
+ .sql("SELECT c1 FROM t WHERE c3 = 1 ORDER BY c1")
.await
.unwrap()
.collect()
.await
.unwrap();
+
let expected = [
Review Comment:
minor it would be nice to have this be formatted (maybe ignore rustfmt)?
--
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] Cut Parquet over to PhysicalExprAdapter from SchemaAdapter [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#discussion_r2576301787
##########
datafusion/datasource-parquet/src/row_filter.rs:
##########
@@ -270,10 +233,12 @@ impl FilterCandidateBuilder {
.project(&required_indices_into_table_schema)?,
);
- let (schema_mapper, projection_into_file_schema) = self
- .schema_adapter_factory
- .create(Arc::clone(&projected_table_schema), self.table_schema)
- .map_schema(&self.file_schema)?;
+ // Compute the projection into the file schema by matching column names
+ let projection_into_file_schema: Vec<usize> = projected_table_schema
+ .fields()
+ .iter()
+ .filter_map(|f| self.file_schema.index_of(f.name()).ok())
+ .collect();
Review Comment:
This assumes predicate column indices refer to the table/logical schema (not file schema), and that we always map those logical columns into file indices by name.
That's true today because DefaultPhysicalExprAdapter rewrites expressions but preserves logical column indices and we do name-based mapping here.
Do you think adding:
1. a regression test for the reorder case (table: (a,b,c), file: (b,c,a)), ensuring both pruning and row-filtering work
2. an explicit assert and comment in [FilterCandidateBuilder::build] verifying that each required logical idx resolves to a unique file index or is intentionally missing and handled by adapter.
would prevent regression?
--
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] Cut `Parquet` over to PhysicalExprAdapter, remove `SchemaAdapter` [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#issuecomment-3615651901
Apologies, I merged with failing `cargo fmt`: https://github.com/apache/datafusion/pull/19102
--
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] Cut Parquet over to PhysicalExprAdapter from SchemaAdapter [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#discussion_r2589978640
##########
datafusion/datasource-parquet/src/row_filter.rs:
##########
@@ -270,10 +233,12 @@ impl FilterCandidateBuilder {
.project(&required_indices_into_table_schema)?,
);
- let (schema_mapper, projection_into_file_schema) = self
- .schema_adapter_factory
- .create(Arc::clone(&projected_table_schema), self.table_schema)
- .map_schema(&self.file_schema)?;
+ // Compute the projection into the file schema by matching column names
+ let projection_into_file_schema: Vec<usize> = projected_table_schema
+ .fields()
+ .iter()
+ .filter_map(|f| self.file_schema.index_of(f.name()).ok())
+ .collect();
Review Comment:
How about 3cd7c84b11a94c3fd27884249b37ce4986908c20 ?
--
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] Cut `Parquet` over to PhysicalExprAdapter, remove `SchemaAdapter` [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#issuecomment-3614766707
> Will we remove `SchemaAdapter` entirely in this release?
I think we could but I see no rush. Will address comments tomorrow ๐
--
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] Cut `Parquet` over to PhysicalExprAdapter, remove `SchemaAdapter` [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#discussion_r2591394234
##########
datafusion/datasource-parquet/src/row_filter.rs:
##########
@@ -270,10 +233,12 @@ impl FilterCandidateBuilder {
.project(&required_indices_into_table_schema)?,
);
- let (schema_mapper, projection_into_file_schema) = self
- .schema_adapter_factory
- .create(Arc::clone(&projected_table_schema), self.table_schema)
- .map_schema(&self.file_schema)?;
+ // Compute the projection into the file schema by matching column names
+ let projection_into_file_schema: Vec<usize> = projected_table_schema
+ .fields()
+ .iter()
+ .filter_map(|f| self.file_schema.index_of(f.name()).ok())
+ .collect();
Review Comment:
3cd7c84b11a94c3fd27884249b37ce4986908c20 is solid regression shield for both paths!
--
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] Cut Parquet over to PhysicalExprAdapter from SchemaAdapter [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#discussion_r2581191718
##########
datafusion/datasource-parquet/src/row_filter.rs:
##########
@@ -270,10 +233,12 @@ impl FilterCandidateBuilder {
.project(&required_indices_into_table_schema)?,
);
- let (schema_mapper, projection_into_file_schema) = self
- .schema_adapter_factory
- .create(Arc::clone(&projected_table_schema), self.table_schema)
- .map_schema(&self.file_schema)?;
+ // Compute the projection into the file schema by matching column names
+ let projection_into_file_schema: Vec<usize> = projected_table_schema
+ .fields()
+ .iter()
+ .filter_map(|f| self.file_schema.index_of(f.name()).ok())
+ .collect();
Review Comment:
https://github.com/apache/datafusion/commit/66df4f27f67d285aaf50dd0e0208bca3b06a5c7c are excellent projection tests for:
flipped order (c,b,a) vs (a,b,c), and
missing column (a,c) vs (a,b,c)
The tests for pruning / row filter path are still missing.
--
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] Cut `Parquet` over to PhysicalExprAdapter, remove `SchemaAdapter` [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#issuecomment-3614567468
Will we remove `SchemaAdapter` entirely in this release?
--
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] Cut Parquet over to PhysicalExprAdapter from SchemaAdapter [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#discussion_r2589975202
##########
datafusion/datasource-parquet/src/row_filter.rs:
##########
@@ -270,10 +233,12 @@ impl FilterCandidateBuilder {
.project(&required_indices_into_table_schema)?,
);
- let (schema_mapper, projection_into_file_schema) = self
- .schema_adapter_factory
- .create(Arc::clone(&projected_table_schema), self.table_schema)
- .map_schema(&self.file_schema)?;
+ // Compute the projection into the file schema by matching column names
+ let projection_into_file_schema: Vec<usize> = projected_table_schema
+ .fields()
+ .iter()
+ .filter_map(|f| self.file_schema.index_of(f.name()).ok())
+ .collect();
Review Comment:
How about 3cd7c84 ?
--
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] Cut Parquet over to PhysicalExprAdapter from SchemaAdapter [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#discussion_r2578270996
##########
datafusion/core/src/datasource/physical_plan/parquet.rs:
##########
Review Comment:
The comment predates this PR but I agree it's wrong. I've updated 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] Cut Parquet over to PhysicalExprAdapter from SchemaAdapter [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#issuecomment-3592919520
run benchmarks
--
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] Cut Parquet over to PhysicalExprAdapter from SchemaAdapter [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#discussion_r2589975202
##########
datafusion/datasource-parquet/src/row_filter.rs:
##########
@@ -270,10 +233,12 @@ impl FilterCandidateBuilder {
.project(&required_indices_into_table_schema)?,
);
- let (schema_mapper, projection_into_file_schema) = self
- .schema_adapter_factory
- .create(Arc::clone(&projected_table_schema), self.table_schema)
- .map_schema(&self.file_schema)?;
+ // Compute the projection into the file schema by matching column names
+ let projection_into_file_schema: Vec<usize> = projected_table_schema
+ .fields()
+ .iter()
+ .filter_map(|f| self.file_schema.index_of(f.name()).ok())
+ .collect();
Review Comment:
How about 3cd7c84?
--
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] Cut `Parquet` over to PhysicalExprAdapter, remove `SchemaAdapter` [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#issuecomment-3615652868
Thanks for the reviews @kosiew !
--
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] Cut `Parquet` over to PhysicalExprAdapter, remove `SchemaAdapter` [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb merged PR #18998:
URL: https://github.com/apache/datafusion/pull/18998
--
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] Cut Parquet over to PhysicalExprAdapter from SchemaAdapter [datafusion]
Posted by "martin-g (via GitHub)" <gi...@apache.org>.
martin-g commented on code in PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#discussion_r2574852819
##########
datafusion/core/src/datasource/physical_plan/parquet.rs:
##########
Review Comment:
Why `c4(string)` ? It is a Date64
##########
datafusion/core/src/datasource/physical_plan/parquet.rs:
##########
@@ -1272,7 +1272,7 @@ mod tests {
.round_trip_to_batches(vec![batch1, batch2])
.await;
assert_contains!(read.unwrap_err().to_string(),
- "Cannot cast file schema field c3 of type Date64 to table schema field of type Int8");
+ "Cannot cast column 'c3' from 'Date64' (physical data type) to 'Int8' (logical data type)");
Review Comment:
This looks the opposite to me - Int should be the physical type and Date is the logical one
--
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] Cut Parquet over to PhysicalExprAdapter from SchemaAdapter [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#discussion_r2578276350
##########
datafusion/core/src/datasource/physical_plan/parquet.rs:
##########
@@ -1272,7 +1272,7 @@ mod tests {
.round_trip_to_batches(vec![batch1, batch2])
.await;
assert_contains!(read.unwrap_err().to_string(),
- "Cannot cast file schema field c3 of type Date64 to table schema field of type Int8");
+ "Cannot cast column 'c3' from 'Date64' (physical data type) to 'Int8' (logical data type)");
Review Comment:
I checked and I think this is right.
From above:
```rust
let batch2 = create_batch(vec![("c3", c4), ("c2", c2), ("c1", c1)]);
```
And:
```rust
let c4: ArrayRef = Arc::new(Date64Array::from(vec![
Some(86400000),
None,
Some(259200000),
]));
```
But in the schema:
```rust
Field::new("c3", DataType::Int8, true)
```
So in the physical data `c3` has the data type `Date64` and the logical type is `Int8`
--
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] Cut Parquet over to PhysicalExprAdapter from SchemaAdapter [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#discussion_r2588230985
##########
datafusion/datasource-parquet/src/row_filter.rs:
##########
@@ -270,10 +233,12 @@ impl FilterCandidateBuilder {
.project(&required_indices_into_table_schema)?,
);
- let (schema_mapper, projection_into_file_schema) = self
- .schema_adapter_factory
- .create(Arc::clone(&projected_table_schema), self.table_schema)
- .map_schema(&self.file_schema)?;
+ // Compute the projection into the file schema by matching column names
+ let projection_into_file_schema: Vec<usize> = projected_table_schema
+ .fields()
+ .iter()
+ .filter_map(|f| self.file_schema.index_of(f.name()).ok())
+ .collect();
Review Comment:
Yes
--
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] Cut Parquet over to PhysicalExprAdapter from SchemaAdapter [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #18998:
URL: https://github.com/apache/datafusion/pull/18998#discussion_r2582028179
##########
datafusion/datasource-parquet/src/row_filter.rs:
##########
@@ -270,10 +233,12 @@ impl FilterCandidateBuilder {
.project(&required_indices_into_table_schema)?,
);
- let (schema_mapper, projection_into_file_schema) = self
- .schema_adapter_factory
- .create(Arc::clone(&projected_table_schema), self.table_schema)
- .map_schema(&self.file_schema)?;
+ // Compute the projection into the file schema by matching column names
+ let projection_into_file_schema: Vec<usize> = projected_table_schema
+ .fields()
+ .iter()
+ .filter_map(|f| self.file_schema.index_of(f.name()).ok())
+ .collect();
Review Comment:
Makes sense, how about 4f00a2f07?
--
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