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/04/18 19:57:08 UTC
[PR] re-implement filter pushdown for parquet [datafusion]
adriangb opened a new pull request, #15769:
URL: https://github.com/apache/datafusion/pull/15769
Very much WIP, please do not review
--
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] re-implement filter pushdown for parquet [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2051060245
##########
datafusion/datasource-parquet/src/file_format.rs:
##########
@@ -453,29 +451,8 @@ impl FileFormat for ParquetFormat {
Ok(Arc::new(DataSinkExec::new(input, sink, order_requirements)) as _)
}
- fn supports_filters_pushdown(
- &self,
- file_schema: &Schema,
- table_schema: &Schema,
- filters: &[&Expr],
- ) -> Result<FilePushdownSupport> {
- if !self.options().global.pushdown_filters {
- return Ok(FilePushdownSupport::NoSupport);
- }
-
- let all_supported = filters.iter().all(|filter| {
- can_expr_be_pushed_down_with_schemas(filter, file_schema, table_schema)
- });
-
- Ok(if all_supported {
- FilePushdownSupport::Supported
- } else {
- FilePushdownSupport::NotSupportedForFilter
- })
- }
-
fn file_source(&self) -> Arc<dyn FileSource> {
- Arc::new(ParquetSource::default())
+ todo!() // need access of file schema?
Review Comment:
This poses an issue.
TLDR is that in order to know if it can absorb a filter as exact `ParquetSource` needs to know not only the filter but also the file schema it's applied to (in particular to get the type of the columns since it can't handle structs).
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -587,4 +604,40 @@ impl FileSource for ParquetSource {
}
}
}
+
+ fn try_pushdown_filters(
+ &self,
+ fd: FilterDescription,
+ config: &datafusion_common::config::ConfigOptions,
+ ) -> datafusion_common::Result<FilterPushdownResult<Arc<dyn FileSource>>> {
+ let mut conf = self.clone();
+ let filters = fd.filters.clone();
+ let predicate = match conf.predicate {
+ Some(predicate) => conjunction(std::iter::once(predicate).chain(filters.into_iter())),
+ None => conjunction(filters.into_iter()),
+ };
+ conf.predicate = Some(predicate);
+ let remaining_description = if config.execution.parquet.pushdown_filters {
+ let mut remaining_filters = fd.filters.clone();
+ for filter in &remaining_filters {
+ if can_expr_be_pushed_down_with_schemas(filter, &conf.file_schema) {
Review Comment:
This is where we need the file schema
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2055055898
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -589,4 +559,49 @@ impl FileSource for ParquetSource {
}
}
}
+
+ fn try_pushdown_filters(
+ &self,
+ fd: FilterDescription,
+ config: &datafusion_common::config::ConfigOptions,
+ ) -> datafusion_common::Result<FilterPushdownResult<Arc<dyn FileSource>>> {
+ let Some(file_schema) = self.file_schema.clone() else {
+ return Ok(filter_pushdown_not_supported(fd));
+ };
+ let config_pushdown_enabled = config.execution.parquet.pushdown_filters;
+ let table_pushdown_enabled = self.pushdown_filters();
+ if table_pushdown_enabled || config_pushdown_enabled {
+ let mut conf = self.clone();
+ let mut allowed_filters = vec![];
+ let mut remaining_filters = vec![];
+ for filter in &fd.filters {
Review Comment:
f2bec87e2
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2073017359
##########
datafusion/sqllogictest/test_files/push_down_filter.slt:
##########
@@ -218,43 +219,57 @@ LOCATION 'test_files/scratch/push_down_filter/t.parquet';
query TT
explain select a from t where a = '100';
----
-logical_plan TableScan: t projection=[a], full_filters=[t.a = Int32(100)]
+logical_plan
+01)Filter: t.a = Int32(100)
+02)--TableScan: t projection=[a], partial_filters=[t.a = Int32(100)]
# The predicate should not have a column cast when the value is a valid i32
query TT
explain select a from t where a != '100';
----
-logical_plan TableScan: t projection=[a], full_filters=[t.a != Int32(100)]
+logical_plan
+01)Filter: t.a != Int32(100)
+02)--TableScan: t projection=[a], partial_filters=[t.a != Int32(100)]
Review Comment:
There is a statement in 21st line in this file:
`set datafusion.explain.logical_plan_only = true;`
There is also counter-part of that: `set datafusion.explain.physical_plan_only = true`
You can set the display type of explain queries with these.
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2163561507
##########
datafusion/datasource/src/file_format.rs:
##########
@@ -94,7 +93,6 @@ pub trait FileFormat: Send + Sync + fmt::Debug {
&self,
state: &dyn Session,
conf: FileScanConfig,
- filters: Option<&Arc<dyn PhysicalExpr>>,
Review Comment:
Thank you @alamb, I'll add this to my TODO list.
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#issuecomment-2827757032
I'm not really sure how this degrades anything. The end result is the same, users won't see any difference.
What ListingTable does currently is misguided and wrong since it is not really at the logical level as you say, instead it pierces the logical / physical separation (see how it converts Expr to PhysicalExpr, etc). It even produces bugs (I believe the pushdown of struct fields may currently be broken, or at least the implementation is confusing and the test is completely wrong).
I think there is pushdown that makes sense at the logical level, namely partition pruning. And I left that for TableProvider to continue to do. But the pruning that relies on a PhysicalExpr seems to me like it should be happening at the physical layer not the logical.
We might be able to leave the stuff in TableProvider in place but we'll be dealing with duplication and confusing methods on DataSource, which is already a complex bit of code. When I first tried to implement it this way I ran into cases with duplicate pushdown and other confusing scenarios. Probably it could have been resolved but I felt like why make one of the most complex bits in DataFusion even more complex instead of simplifying it where possible.
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#issuecomment-2850482445
I think you can disable logical plans. That's what @alamb suggests, if I don't get it wrong
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2074120924
##########
datafusion/core/tests/sql/path_partition.rs:
##########
@@ -57,55 +53,6 @@ use object_store::{
use object_store::{Attributes, MultipartUpload, PutMultipartOpts, PutPayload};
use url::Url;
-#[tokio::test]
-async fn parquet_partition_pruning_filter() -> Result<()> {
Review Comment:
can you remind me again why this test was removed ?
##########
datafusion/datasource-parquet/src/mod.rs:
##########
@@ -244,7 +242,7 @@ impl ParquetExecBuilder {
inner: DataSourceExec::new(Arc::new(base_config.clone())),
base_config,
predicate,
- pruning_predicate: parquet.pruning_predicate,
+ pruning_predicate: None, // for backwards compat since `ParquetExec` is only for backwards compat anyway
Review Comment:
I filed a PR to track this
- https://github.com/apache/datafusion/issues/15950
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -589,4 +559,49 @@ impl FileSource for ParquetSource {
}
}
}
+
+ fn try_pushdown_filters(
+ &self,
+ fd: FilterDescription,
+ config: &datafusion_common::config::ConfigOptions,
+ ) -> datafusion_common::Result<FilterPushdownResult<Arc<dyn FileSource>>> {
+ let Some(file_schema) = self.file_schema.clone() else {
Review Comment:
Let's file that follow on ticket
##########
datafusion/sqllogictest/test_files/push_down_filter.slt:
##########
@@ -18,7 +18,7 @@
# Test push down filter
statement ok
-set datafusion.explain.logical_plan_only = true;
+set datafusion.explain.physical_plan_only = true;
Review Comment:
💯
##########
docs/source/library-user-guide/upgrading.md:
##########
@@ -46,6 +46,16 @@ schema. To upgrade structs which implement `PhysicalExpr` you need to implement
the `return_field` function. There are numerous examples in the `physical-expr`
crate.
+### `FileFormat::supports_filters_pushdown` replaced with `FileSource::try_pushdown_filters`
Review Comment:
👍
##########
datafusion/sqllogictest/test_files/push_down_filter.slt:
##########
@@ -179,9 +190,9 @@ LOCATION 'test_files/scratch/parquet/test_filter_with_limit/';
query TT
explain select * from test_filter_with_limit where value = 2 limit 1;
----
-logical_plan
-01)Limit: skip=0, fetch=1
-02)--TableScan: test_filter_with_limit projection=[part_key, value], full_filters=[test_filter_with_limit.value = Int32(2)], fetch=1
+physical_plan
+01)CoalescePartitionsExec: fetch=1
+02)--DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet/test_filter_with_limit/part-0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet/test_filter_with_limit/part-1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet/test_filter_with_limit/part-2.parquet]]}, projection=[part_key, value], limit=1, file_type=parquet, predicate=value@1 = 2, pruning_predicate=value_null_count@2 != row_count@3 AND value_min@0 <= 2 AND 2 <= value_max@1, required_guarantees=[value in (2)]
Review Comment:
it is good to see this the predicate pushed down
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#issuecomment-2851630545
Okay folks this is rebased and ready for another round of review
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#issuecomment-2848670746
> Is there anything I'm missing? Is this what you meant by `the mistake`?
I meant the need of this PR.
> Is there anything I'm missing?
https://github.com/apache/datafusion/pull/15769#discussion_r2070398720
a ticket would be good IMO as well
https://github.com/apache/datafusion/pull/15769#discussion_r2070804563
do you wanna give it a try to change the orders of rules?
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2163543672
##########
datafusion/datasource/src/file_format.rs:
##########
@@ -94,7 +93,6 @@ pub trait FileFormat: Send + Sync + fmt::Debug {
&self,
state: &dyn Session,
conf: FileScanConfig,
- filters: Option<&Arc<dyn PhysicalExpr>>,
Review Comment:
I feel ilke we have done this a few times now @xudong963 (broken some APIs you are relying on heavily)
I wonder if there is some way to make a tests / example that illustrates what you are doing? That would have some benefits:
1. it would illustrate exactly the way the API needs to be changed on upgrade
2. it would make it clearer when changes to APIs have impacts on you (and other's usecases)
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#issuecomment-2850732455
I think we should also inform people who might be possibly effected by these logical plan changes
@andygrove
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2053975043
##########
datafusion/sqllogictest/test_files/aggregate.slt:
##########
@@ -5006,7 +5006,7 @@ SELECT column5, avg(column1) FROM d GROUP BY column5;
query I??
SELECT column5, column1, avg(column1) OVER (PARTITION BY column5 ORDER BY column1 ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) as window_avg
-FROM d WHERE column1 IS NOT NULL;
+FROM d WHERE column1 IS NOT NULL ORDER BY column5 DESC;
Review Comment:
You probably need to rebase. partition by is similar to group by, so you won't get the deterministic result. You need `rowsort` for this kind of query
https://github.com/apache/datafusion/blob/449c744c8369ecc3436d39010ceb70b271389c5f/datafusion/sqllogictest/test_files/aggregate.slt#L5008-L5015
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2055056263
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -589,4 +559,49 @@ impl FileSource for ParquetSource {
}
}
}
+
+ fn try_pushdown_filters(
+ &self,
+ fd: FilterDescription,
+ config: &datafusion_common::config::ConfigOptions,
+ ) -> datafusion_common::Result<FilterPushdownResult<Arc<dyn FileSource>>> {
+ let Some(file_schema) = self.file_schema.clone() else {
+ return Ok(filter_pushdown_not_supported(fd));
+ };
+ let config_pushdown_enabled = config.execution.parquet.pushdown_filters;
+ let table_pushdown_enabled = self.pushdown_filters();
+ if table_pushdown_enabled || config_pushdown_enabled {
+ let mut conf = self.clone();
Review Comment:
8c1b98cf8
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2052755734
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -589,4 +559,49 @@ impl FileSource for ParquetSource {
}
}
}
+
+ fn try_pushdown_filters(
+ &self,
+ fd: FilterDescription,
+ config: &datafusion_common::config::ConfigOptions,
+ ) -> datafusion_common::Result<FilterPushdownResult<Arc<dyn FileSource>>> {
+ let Some(file_schema) = self.file_schema.clone() else {
Review Comment:
I think they always end up with a schema now, but the current APIs don't require it via the constructor and instead it gets passed in via `FileScanConfigBuilder`. I tried piping it into the constructor but makes things difficult, there's APIs that rely on `ParquetSource::default()` and such. So TLDR is it's a bit gross but this is the least chrun way to do it and we can always come back later and clean the rest up.
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#issuecomment-2850530740
> I think you can disable logical plans. That's what @alamb suggests, if I don't get it wrong
done!
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#issuecomment-2848631745
@adriangb do you have time to address the last suggestions? I understand the mistake here, and I think we should take this in asap
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2072445465
##########
datafusion/sqllogictest/test_files/push_down_filter.slt:
##########
@@ -218,43 +219,57 @@ LOCATION 'test_files/scratch/push_down_filter/t.parquet';
query TT
explain select a from t where a = '100';
----
-logical_plan TableScan: t projection=[a], full_filters=[t.a = Int32(100)]
+logical_plan
+01)Filter: t.a = Int32(100)
+02)--TableScan: t projection=[a], partial_filters=[t.a = Int32(100)]
# The predicate should not have a column cast when the value is a valid i32
query TT
explain select a from t where a != '100';
----
-logical_plan TableScan: t projection=[a], full_filters=[t.a != Int32(100)]
+logical_plan
+01)Filter: t.a != Int32(100)
+02)--TableScan: t projection=[a], partial_filters=[t.a != Int32(100)]
Review Comment:
@alamb do you know why these don't display the phyiscal plan already? Is something parsing them out?
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2052756433
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -253,18 +251,18 @@ use object_store::ObjectStore;
/// [`RecordBatch`]: arrow::record_batch::RecordBatch
/// [`SchemaAdapter`]: datafusion_datasource::schema_adapter::SchemaAdapter
/// [`ParquetMetadata`]: parquet::file::metadata::ParquetMetaData
-#[derive(Clone, Default, Debug)]
+#[derive(Clone, Debug, Default)]
pub struct ParquetSource {
/// Options for reading Parquet files
pub(crate) table_parquet_options: TableParquetOptions,
/// Optional metrics
pub(crate) metrics: ExecutionPlanMetricsSet,
+ /// The schema of the file.
+ /// In particular, this is the schema of the table without partition columns,
+ /// *not* the physical schema of the file.
+ pub(crate) file_schema: Option<SchemaRef>,
Review Comment:
This is the same schema that `FileScanConfig` passes into `ParquetSource`
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2052242110
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -589,4 +559,49 @@ impl FileSource for ParquetSource {
}
}
}
+
+ fn try_pushdown_filters(
+ &self,
+ fd: FilterDescription,
+ config: &datafusion_common::config::ConfigOptions,
+ ) -> datafusion_common::Result<FilterPushdownResult<Arc<dyn FileSource>>> {
+ let Some(file_schema) = self.file_schema.clone() else {
+ return Ok(filter_pushdown_not_supported(fd));
+ };
+ let config_pushdown_enabled = config.execution.parquet.pushdown_filters;
+ let table_pushdown_enabled = self.pushdown_filters();
+ if table_pushdown_enabled || config_pushdown_enabled {
Review Comment:
OR'ing this is correct?
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -253,18 +251,18 @@ use object_store::ObjectStore;
/// [`RecordBatch`]: arrow::record_batch::RecordBatch
/// [`SchemaAdapter`]: datafusion_datasource::schema_adapter::SchemaAdapter
/// [`ParquetMetadata`]: parquet::file::metadata::ParquetMetaData
-#[derive(Clone, Default, Debug)]
+#[derive(Clone, Debug, Default)]
pub struct ParquetSource {
/// Options for reading Parquet files
pub(crate) table_parquet_options: TableParquetOptions,
/// Optional metrics
pub(crate) metrics: ExecutionPlanMetricsSet,
+ /// The schema of the file.
+ /// In particular, this is the schema of the table without partition columns,
+ /// *not* the physical schema of the file.
+ pub(crate) file_schema: Option<SchemaRef>,
/// Optional predicate for row filtering during parquet scan
pub(crate) predicate: Option<Arc<dyn PhysicalExpr>>,
- /// Optional predicate for pruning row groups (derived from `predicate`)
Review Comment:
good to see these are unifying
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -253,18 +251,18 @@ use object_store::ObjectStore;
/// [`RecordBatch`]: arrow::record_batch::RecordBatch
/// [`SchemaAdapter`]: datafusion_datasource::schema_adapter::SchemaAdapter
/// [`ParquetMetadata`]: parquet::file::metadata::ParquetMetaData
-#[derive(Clone, Default, Debug)]
+#[derive(Clone, Debug, Default)]
pub struct ParquetSource {
/// Options for reading Parquet files
pub(crate) table_parquet_options: TableParquetOptions,
/// Optional metrics
pub(crate) metrics: ExecutionPlanMetricsSet,
+ /// The schema of the file.
+ /// In particular, this is the schema of the table without partition columns,
+ /// *not* the physical schema of the file.
+ pub(crate) file_schema: Option<SchemaRef>,
Review Comment:
There is also another schema in FileScanConfig. Are they both reflects the file schema, not physical schema? and can we somehow unify them?
##########
datafusion/core/src/datasource/listing/table.rs:
##########
@@ -982,18 +980,6 @@ impl TableProvider for ListingTable {
return Ok(TableProviderFilterPushDown::Exact);
}
- // if we can't push it down completely with only the filename-based/path-based
- // column names, then we should check if we can do parquet predicate pushdown
- let supports_pushdown = self.options.format.supports_filters_pushdown(
- &self.file_schema,
- &self.table_schema,
- &[filter],
- )?;
-
- if supports_pushdown == FilePushdownSupport::Supported {
- return Ok(TableProviderFilterPushDown::Exact);
- }
Review Comment:
If that's the case, why don't we fully remove `supports_filters_pushdown()` API at all
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -589,4 +559,49 @@ impl FileSource for ParquetSource {
}
}
}
+
+ fn try_pushdown_filters(
+ &self,
+ fd: FilterDescription,
+ config: &datafusion_common::config::ConfigOptions,
+ ) -> datafusion_common::Result<FilterPushdownResult<Arc<dyn FileSource>>> {
+ let Some(file_schema) = self.file_schema.clone() else {
Review Comment:
I'm asking to learn: in which cases ParquetSource doesn't have the schema?
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -589,4 +559,49 @@ impl FileSource for ParquetSource {
}
}
}
+
+ fn try_pushdown_filters(
+ &self,
+ fd: FilterDescription,
+ config: &datafusion_common::config::ConfigOptions,
+ ) -> datafusion_common::Result<FilterPushdownResult<Arc<dyn FileSource>>> {
+ let Some(file_schema) = self.file_schema.clone() else {
+ return Ok(filter_pushdown_not_supported(fd));
+ };
+ let config_pushdown_enabled = config.execution.parquet.pushdown_filters;
+ let table_pushdown_enabled = self.pushdown_filters();
+ if table_pushdown_enabled || config_pushdown_enabled {
+ let mut conf = self.clone();
Review Comment:
`conf` name is not very clear
##########
datafusion/core/src/datasource/listing/table.rs:
##########
@@ -982,18 +980,6 @@ impl TableProvider for ListingTable {
return Ok(TableProviderFilterPushDown::Exact);
}
- // if we can't push it down completely with only the filename-based/path-based
- // column names, then we should check if we can do parquet predicate pushdown
- let supports_pushdown = self.options.format.supports_filters_pushdown(
- &self.file_schema,
- &self.table_schema,
- &[filter],
- )?;
-
- if supports_pushdown == FilePushdownSupport::Supported {
- return Ok(TableProviderFilterPushDown::Exact);
- }
Review Comment:
I have one question: aren't we expecting/preparing for, people to use ListingTable if they read Parquet files? Are we eventually planning to remove all format-specific handlings? Or this is a case only for filter pushdown?
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -81,11 +81,15 @@ EXPLAIN select a from t_pushdown where b > 2 ORDER BY a;
----
logical_plan
01)Sort: t_pushdown.a ASC NULLS LAST
-02)--TableScan: t_pushdown projection=[a], full_filters=[t_pushdown.b > Int32(2)]
+02)--Projection: t_pushdown.a
+03)----Filter: t_pushdown.b > Int32(2)
+04)------TableScan: t_pushdown projection=[a, b], partial_filters=[t_pushdown.b > Int32(2)]
physical_plan
01)SortPreservingMergeExec: [a@0 ASC NULLS LAST]
02)--SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]
-03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 2, required_guarantees=[]
+03)----CoalesceBatchesExec: target_batch_size=8192
+04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=2
+05)--------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2 AND b@1 > 2
Review Comment:
I've a guess but not proved: CoalesceBatchesExec comes because of RepartitionExec, and RepartitionExec is inserted to satisfy partition count, which is 4. That's required by FilterExec now (which was pushed down at the logical level before), but that FilterExec is pushed down later after EnforceDistribution.
So, this makes me think about the correct order of physical rules. PushdownFilter should probably work before distribution&order satisfiers. But that could also bring some issues, I'm not sure.
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -589,4 +559,49 @@ impl FileSource for ParquetSource {
}
}
}
+
+ fn try_pushdown_filters(
+ &self,
+ fd: FilterDescription,
+ config: &datafusion_common::config::ConfigOptions,
+ ) -> datafusion_common::Result<FilterPushdownResult<Arc<dyn FileSource>>> {
+ let Some(file_schema) = self.file_schema.clone() else {
+ return Ok(filter_pushdown_not_supported(fd));
+ };
+ let config_pushdown_enabled = config.execution.parquet.pushdown_filters;
+ let table_pushdown_enabled = self.pushdown_filters();
+ if table_pushdown_enabled || config_pushdown_enabled {
+ let mut conf = self.clone();
+ let mut allowed_filters = vec![];
+ let mut remaining_filters = vec![];
+ for filter in &fd.filters {
Review Comment:
`fd.take_filters()` to avoid clone's below
##########
datafusion/sqllogictest/test_files/aggregate.slt:
##########
@@ -5006,7 +5006,7 @@ SELECT column5, avg(column1) FROM d GROUP BY column5;
query I??
SELECT column5, column1, avg(column1) OVER (PARTITION BY column5 ORDER BY column1 ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) as window_avg
-FROM d WHERE column1 IS NOT NULL;
+FROM d WHERE column1 IS NOT NULL ORDER BY column5 DESC;
Review Comment:
Plan doesn't change, but why results changed 🤔
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2070868393
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -81,11 +81,15 @@ EXPLAIN select a from t_pushdown where b > 2 ORDER BY a;
----
logical_plan
01)Sort: t_pushdown.a ASC NULLS LAST
-02)--TableScan: t_pushdown projection=[a], full_filters=[t_pushdown.b > Int32(2)]
+02)--Projection: t_pushdown.a
+03)----Filter: t_pushdown.b > Int32(2)
+04)------TableScan: t_pushdown projection=[a, b], partial_filters=[t_pushdown.b > Int32(2)]
physical_plan
01)SortPreservingMergeExec: [a@0 ASC NULLS LAST]
02)--SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]
-03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 2, required_guarantees=[]
+03)----CoalesceBatchesExec: target_batch_size=8192
+04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=2
+05)--------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2 AND b@1 > 2
Review Comment:
This would also be fixed by https://github.com/apache/datafusion/pull/15801
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2075188445
##########
datafusion/physical-plan/src/filter_pushdown.rs:
##########
@@ -98,6 +116,13 @@ impl PredicateSupports {
.collect()
}
+ /// Are all filters marked as [`PredicateSupport::Supported`]?
+ pub fn is_exact(&self) -> bool {
Review Comment:
Already removed in 15443d5
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#issuecomment-2855539680
WOOHOO!
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2163862690
##########
datafusion/datasource/src/file_format.rs:
##########
@@ -94,7 +93,6 @@ pub trait FileFormat: Send + Sync + fmt::Debug {
&self,
state: &dyn Session,
conf: FileScanConfig,
- filters: Option<&Arc<dyn PhysicalExpr>>,
Review Comment:
I've found a way to work around the change, not a big deal, thanks for your quick response
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2058172660
##########
datafusion/core/src/datasource/listing/table.rs:
##########
@@ -982,18 +980,6 @@ impl TableProvider for ListingTable {
return Ok(TableProviderFilterPushDown::Exact);
}
- // if we can't push it down completely with only the filename-based/path-based
- // column names, then we should check if we can do parquet predicate pushdown
- let supports_pushdown = self.options.format.supports_filters_pushdown(
- &self.file_schema,
- &self.table_schema,
- &[filter],
- )?;
-
- if supports_pushdown == FilePushdownSupport::Supported {
- return Ok(TableProviderFilterPushDown::Exact);
- }
Review Comment:
We can justify implementing other TableProviders for Parquet, but still I cannot understand why we need to degrade the capabilities of our ListingTable. Is't it always better pruning/simplifying things at the higher levels as possible?
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2074207291
##########
datafusion/physical-optimizer/src/optimizer.rs:
##########
@@ -95,6 +95,10 @@ impl PhysicalOptimizer {
// as that rule may inject other operations in between the different AggregateExecs.
// Applying the rule early means only directly-connected AggregateExecs must be examined.
Arc::new(LimitedDistinctAggregation::new()),
+ // The FilterPushdown rule tries to push down filters as far as it can.
+ // For example, it will push down filtering from a `FilterExec` to
+ // a `DataSourceExec`, or from a `TopK`'s current state to a `DataSourceExec`.
+ Arc::new(FilterPushdown::new()),
Review Comment:
@alamb this is actually an issue, moving this does not work alongside https://github.com/apache/datafusion/pull/15770:
- `SortExec`s filter has to be pushed down after any manipulation of the `SortExec`s in the plan (otherwise it will be referencing a `SortExec` that gets removed / replaced and not function)
- `EnforceSorting` creates new `SortExec`s, so `FilterPushdown` would have to come after that
- The extra `RepartitionExec`s get introduced in `EnforceDistribution`
- Since `EnforceSorting` must come after `EnforceDistribution` there is no correct place to put `FilterPushdown`
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2052753333
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -589,4 +559,49 @@ impl FileSource for ParquetSource {
}
}
}
+
+ fn try_pushdown_filters(
+ &self,
+ fd: FilterDescription,
+ config: &datafusion_common::config::ConfigOptions,
+ ) -> datafusion_common::Result<FilterPushdownResult<Arc<dyn FileSource>>> {
+ let Some(file_schema) = self.file_schema.clone() else {
+ return Ok(filter_pushdown_not_supported(fd));
+ };
+ let config_pushdown_enabled = config.execution.parquet.pushdown_filters;
+ let table_pushdown_enabled = self.pushdown_filters();
+ if table_pushdown_enabled || config_pushdown_enabled {
Review Comment:
The current behavior is not documented anywhere, I tried to match the existing tests: https://github.com/apache/datafusion/blob/9730404028a91a7fe875ea3f88bafdbcb305ae6c/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt#L44-L88
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2053975043
##########
datafusion/sqllogictest/test_files/aggregate.slt:
##########
@@ -5006,7 +5006,7 @@ SELECT column5, avg(column1) FROM d GROUP BY column5;
query I??
SELECT column5, column1, avg(column1) OVER (PARTITION BY column5 ORDER BY column1 ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) as window_avg
-FROM d WHERE column1 IS NOT NULL;
+FROM d WHERE column1 IS NOT NULL ORDER BY column5 DESC;
Review Comment:
You probably need to rebase, I find the result in main branch is the same as your
##########
datafusion/sqllogictest/test_files/aggregate.slt:
##########
@@ -5006,7 +5006,7 @@ SELECT column5, avg(column1) FROM d GROUP BY column5;
query I??
SELECT column5, column1, avg(column1) OVER (PARTITION BY column5 ORDER BY column1 ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) as window_avg
-FROM d WHERE column1 IS NOT NULL;
+FROM d WHERE column1 IS NOT NULL ORDER BY column5 DESC;
Review Comment:
You probably need to rebase, I find the result in main branch is the same as yours
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2052746932
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -81,11 +81,15 @@ EXPLAIN select a from t_pushdown where b > 2 ORDER BY a;
----
logical_plan
01)Sort: t_pushdown.a ASC NULLS LAST
-02)--TableScan: t_pushdown projection=[a], full_filters=[t_pushdown.b > Int32(2)]
+02)--Projection: t_pushdown.a
+03)----Filter: t_pushdown.b > Int32(2)
+04)------TableScan: t_pushdown projection=[a, b], partial_filters=[t_pushdown.b > Int32(2)]
physical_plan
01)SortPreservingMergeExec: [a@0 ASC NULLS LAST]
02)--SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]
-03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 2, required_guarantees=[]
+03)----CoalesceBatchesExec: target_batch_size=8192
+04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=2
+05)--------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2 AND b@1 > 2
Review Comment:
> PushdownFilter should probably work before distribution&order satisfiers
That makes sense to me. It does more "invasive" re-arranging of plans than those do.
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2052747672
##########
datafusion/sqllogictest/test_files/aggregate.slt:
##########
@@ -5006,7 +5006,7 @@ SELECT column5, avg(column1) FROM d GROUP BY column5;
query I??
SELECT column5, column1, avg(column1) OVER (PARTITION BY column5 ORDER BY column1 ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) as window_avg
-FROM d WHERE column1 IS NOT NULL;
+FROM d WHERE column1 IS NOT NULL ORDER BY column5 DESC;
Review Comment:
Yeah I'm a bit confused as well. It's just the order of the results that changed. I feel like there must be something trivial causing this that we can revert.
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2051614507
##########
datafusion/sqllogictest/test_files/aggregate.slt:
##########
@@ -5006,7 +5006,7 @@ SELECT column5, avg(column1) FROM d GROUP BY column5;
query I??
SELECT column5, column1, avg(column1) OVER (PARTITION BY column5 ORDER BY column1 ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) as window_avg
-FROM d WHERE column1 IS NOT NULL;
+FROM d WHERE column1 IS NOT NULL ORDER BY column5 DESC;
Review Comment:
Without this in place the order of the rows changed. I'm not sure why, but I also don't think anything guarantees it without an `ORDER BY` clause so it seemed sensible to add one. A second opinion would be valuable.
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#issuecomment-2816943989
> Thanks @adriangb -- I am about to be offline for a week so I will review this when I return
Enjoy your vacation! I think you'll like this diff:
<img width="144" alt="image" src="https://github.com/user-attachments/assets/d9692934-910b-4974-95dd-fb9ed7beaca3" />
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#issuecomment-2828241765
> I'm not really sure how this degrades anything. The end result is the same, users won't see any difference.
Logical planning results are changing. We are also using DF end-to-end, but there could always be people only relying on logical plans of DF.
> We might be able to leave the stuff in TableProvider in place but we'll be dealing with duplication and confusing methods on DataSource, which is already a complex bit of code. When I first tried to implement it this way I ran into cases with duplicate pushdown and other confusing scenarios. Probably it could have been resolved but I felt like why make one of the most complex bits in DataFusion even more complex instead of simplifying it where possible.
I'm also challenging to decide to be which side because of that complexity :D
I will take a look to other parts in this PR, and try to find a solution for the points like https://github.com/apache/datafusion/pull/15769/files#r2051612926. Maybe there isn't something wrong at all as you said, there is no harm to double check
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2052745403
##########
datafusion/core/src/datasource/listing/table.rs:
##########
@@ -982,18 +980,6 @@ impl TableProvider for ListingTable {
return Ok(TableProviderFilterPushDown::Exact);
}
- // if we can't push it down completely with only the filename-based/path-based
- // column names, then we should check if we can do parquet predicate pushdown
- let supports_pushdown = self.options.format.supports_filters_pushdown(
- &self.file_schema,
- &self.table_schema,
- &[filter],
- )?;
-
- if supports_pushdown == FilePushdownSupport::Supported {
- return Ok(TableProviderFilterPushDown::Exact);
- }
Review Comment:
I think many users of DataFusion (based on our usage, talks I've seen and examples we have) use custom `TableProvider` implementations.
I would keep `supports_filters_pushdown` so that `TableProviders` can do `Exact` pruning of filters, e.g. using partition columns.
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2075160616
##########
datafusion/physical-optimizer/src/optimizer.rs:
##########
@@ -95,6 +95,10 @@ impl PhysicalOptimizer {
// as that rule may inject other operations in between the different AggregateExecs.
// Applying the rule early means only directly-connected AggregateExecs must be examined.
Arc::new(LimitedDistinctAggregation::new()),
+ // The FilterPushdown rule tries to push down filters as far as it can.
+ // For example, it will push down filtering from a `FilterExec` to
+ // a `DataSourceExec`, or from a `TopK`'s current state to a `DataSourceExec`.
+ Arc::new(FilterPushdown::new()),
Review Comment:
If this version is okay in this PR, I can take a look at that in #15770. I've some ideas, like making FilterPushdown smarter to remove any unnecessary patterns behind
##########
datafusion/physical-plan/src/filter_pushdown.rs:
##########
@@ -98,6 +116,13 @@ impl PredicateSupports {
.collect()
}
+ /// Are all filters marked as [`PredicateSupport::Supported`]?
+ pub fn is_exact(&self) -> bool {
Review Comment:
can we find a better name for this? and it's not used, maybe removed until it's needed?
##########
datafusion/core/tests/sql/path_partition.rs:
##########
@@ -57,55 +53,6 @@ use object_store::{
use object_store::{Attributes, MultipartUpload, PutMultipartOpts, PutPayload};
use url::Url;
-#[tokio::test]
-async fn parquet_partition_pruning_filter() -> Result<()> {
Review Comment:
can we also apply the changes similar to the test in `datafusion/core/tests/parquet/file_statistics.rs ` to keep this test for now?
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#issuecomment-2848665962
> @adriangb do you have time to address the last suggestions? I understand the mistake here, and I think we should take this in asap
I am going to try to address the last round of review later today on a flight. In particular:
- https://github.com/apache/datafusion/pull/15769#discussion_r2070806936: change the relevant tests to show the physical plan
- https://github.com/apache/datafusion/pull/15769#discussion_r2070258342: try to show _a_ predicate in explain plans (it will not exactly match the final predicate though since the latter will be generated based on the physical file schema)
Is there anything I'm missing? Is this what you meant by `the mistake`?
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2070868393
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -81,11 +81,15 @@ EXPLAIN select a from t_pushdown where b > 2 ORDER BY a;
----
logical_plan
01)Sort: t_pushdown.a ASC NULLS LAST
-02)--TableScan: t_pushdown projection=[a], full_filters=[t_pushdown.b > Int32(2)]
+02)--Projection: t_pushdown.a
+03)----Filter: t_pushdown.b > Int32(2)
+04)------TableScan: t_pushdown projection=[a, b], partial_filters=[t_pushdown.b > Int32(2)]
physical_plan
01)SortPreservingMergeExec: [a@0 ASC NULLS LAST]
02)--SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]
-03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 2, required_guarantees=[]
+03)----CoalesceBatchesExec: target_batch_size=8192
+04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=2
+05)--------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2 AND b@1 > 2
Review Comment:
This would also be fixed by https://github.com/apache/datafusion/pull/15801
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2163133150
##########
datafusion/datasource/src/file_format.rs:
##########
@@ -94,7 +93,6 @@ pub trait FileFormat: Send + Sync + fmt::Debug {
&self,
state: &dyn Session,
conf: FileScanConfig,
- filters: Option<&Arc<dyn PhysicalExpr>>,
Review Comment:
This is definitely a breaking API change, we depend on the interface intensively 😢
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#issuecomment-2852276025
It looks like there are a few doc errors and doc exaple errors
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2074216845
##########
datafusion/physical-optimizer/src/optimizer.rs:
##########
@@ -95,6 +95,10 @@ impl PhysicalOptimizer {
// as that rule may inject other operations in between the different AggregateExecs.
// Applying the rule early means only directly-connected AggregateExecs must be examined.
Arc::new(LimitedDistinctAggregation::new()),
+ // The FilterPushdown rule tries to push down filters as far as it can.
+ // For example, it will push down filtering from a `FilterExec` to
+ // a `DataSourceExec`, or from a `TopK`'s current state to a `DataSourceExec`.
+ Arc::new(FilterPushdown::new()),
Review Comment:
I guess the only logical solution is to place `FilterPushdown` at the end and remove any unnecessary repartitions?
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#issuecomment-2849130181
I updated the order of the pushdown rules in this PR, it worked to get rid of the extra nodes.
I've also added the upgrade guide and the pushdown preview is being shown in the physical plans.
@alamb I think the only point missing is https://github.com/apache/datafusion/pull/15769#discussion_r2070806936 which I need a big of guidance on
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2052755948
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -589,4 +559,49 @@ impl FileSource for ParquetSource {
}
}
}
+
+ fn try_pushdown_filters(
+ &self,
+ fd: FilterDescription,
+ config: &datafusion_common::config::ConfigOptions,
+ ) -> datafusion_common::Result<FilterPushdownResult<Arc<dyn FileSource>>> {
+ let Some(file_schema) = self.file_schema.clone() else {
+ return Ok(filter_pushdown_not_supported(fd));
+ };
+ let config_pushdown_enabled = config.execution.parquet.pushdown_filters;
+ let table_pushdown_enabled = self.pushdown_filters();
+ if table_pushdown_enabled || config_pushdown_enabled {
+ let mut conf = self.clone();
Review Comment:
It's the pattern used everywhere else but I'll change 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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2053975043
##########
datafusion/sqllogictest/test_files/aggregate.slt:
##########
@@ -5006,7 +5006,7 @@ SELECT column5, avg(column1) FROM d GROUP BY column5;
query I??
SELECT column5, column1, avg(column1) OVER (PARTITION BY column5 ORDER BY column1 ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) as window_avg
-FROM d WHERE column1 IS NOT NULL;
+FROM d WHERE column1 IS NOT NULL ORDER BY column5 DESC;
Review Comment:
You probably need to rebase, I find the result in main branch is the same as yours
https://github.com/apache/datafusion/blob/449c744c8369ecc3436d39010ceb70b271389c5f/datafusion/sqllogictest/test_files/aggregate.slt#L5008-L5015
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#issuecomment-2828267340
I do think you make a good point of "can we keep the current thing and add the new one". It's worth a shot, at least to split the PR into two. And if that's too complicated or if we just want to simplify we can evaluate from there.
--
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] re-implement filter pushdown for parquet [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#issuecomment-2816667139
Thanks @adriangb -- I am about to be offline for a week so I will review this when I return
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2053975043
##########
datafusion/sqllogictest/test_files/aggregate.slt:
##########
@@ -5006,7 +5006,7 @@ SELECT column5, avg(column1) FROM d GROUP BY column5;
query I??
SELECT column5, column1, avg(column1) OVER (PARTITION BY column5 ORDER BY column1 ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) as window_avg
-FROM d WHERE column1 IS NOT NULL;
+FROM d WHERE column1 IS NOT NULL ORDER BY column5 DESC;
Review Comment:
You probably need to rebase
https://github.com/apache/datafusion/blob/449c744c8369ecc3436d39010ceb70b271389c5f/datafusion/sqllogictest/test_files/aggregate.slt#L5008-L5015
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2055066768
##########
datafusion/sqllogictest/test_files/aggregate.slt:
##########
@@ -5006,7 +5006,7 @@ SELECT column5, avg(column1) FROM d GROUP BY column5;
query I??
SELECT column5, column1, avg(column1) OVER (PARTITION BY column5 ORDER BY column1 ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) as window_avg
-FROM d WHERE column1 IS NOT NULL;
+FROM d WHERE column1 IS NOT NULL ORDER BY column5 DESC;
Review Comment:
Thank you indeed that fixed it 😄
##########
datafusion/sqllogictest/test_files/aggregate.slt:
##########
@@ -5006,7 +5006,7 @@ SELECT column5, avg(column1) FROM d GROUP BY column5;
query I??
SELECT column5, column1, avg(column1) OVER (PARTITION BY column5 ORDER BY column1 ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) as window_avg
-FROM d WHERE column1 IS NOT NULL;
+FROM d WHERE column1 IS NOT NULL ORDER BY column5 DESC;
Review Comment:
Thank you indeed rebase fixed 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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2074213248
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -589,4 +559,49 @@ impl FileSource for ParquetSource {
}
}
}
+
+ fn try_pushdown_filters(
+ &self,
+ fd: FilterDescription,
+ config: &datafusion_common::config::ConfigOptions,
+ ) -> datafusion_common::Result<FilterPushdownResult<Arc<dyn FileSource>>> {
+ let Some(file_schema) = self.file_schema.clone() else {
Review Comment:
https://github.com/apache/datafusion/issues/15952
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2070243952
##########
datafusion/core/src/datasource/listing/table.rs:
##########
@@ -982,18 +980,6 @@ impl TableProvider for ListingTable {
return Ok(TableProviderFilterPushDown::Exact);
}
- // if we can't push it down completely with only the filename-based/path-based
Review Comment:
This change makes sense to me -- when @itsjunetime originally implemented this code, there was some complexity because there was no way to do filter pushdown in ExecutionPlans so in my mind this approach was a (clever) workaround
The comments even hint that this is a parquet specific special case
I think the new pattern of handling predicates more generally in this PR is cleaner and will support more cases. Since this code is only currently executed
Perhaps @cisaacson has some other thoughts
##########
datafusion/datasource-parquet/src/mod.rs:
##########
@@ -244,7 +242,7 @@ impl ParquetExecBuilder {
inner: DataSourceExec::new(Arc::new(base_config.clone())),
base_config,
predicate,
- pruning_predicate: parquet.pruning_predicate,
+ pruning_predicate: None, // for backwards compat since `ParquetExec` is only for backwards compat anyway
Review Comment:
yeah this is fine too in my opinion. It is almost time to remove ParquetExec anyways -- maybe we should just do it in this release 🤔
##########
datafusion/core/src/datasource/listing/table.rs:
##########
@@ -982,18 +980,6 @@ impl TableProvider for ListingTable {
return Ok(TableProviderFilterPushDown::Exact);
}
- // if we can't push it down completely with only the filename-based/path-based
- // column names, then we should check if we can do parquet predicate pushdown
- let supports_pushdown = self.options.format.supports_filters_pushdown(
- &self.file_schema,
- &self.table_schema,
- &[filter],
- )?;
-
- if supports_pushdown == FilePushdownSupport::Supported {
- return Ok(TableProviderFilterPushDown::Exact);
- }
Review Comment:
> We can justify implementing other TableProviders for Parquet, but still I cannot understand why we need to degrade the capabilities of our ListingTable. Is't it always better pruning/simplifying things at the higher levels as possible?
I don't think this degrades the capabilities of the current listing table. I think the only implications are for anyone who used a custom `FileFormat` and impleented `supports_filters_pushdown` -- I suspect this is not very common and we can likely avoid consternation by mentioning it in the upgrade guide (see comment below)
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -559,25 +549,8 @@ impl FileSource for ParquetSource {
.predicate()
.map(|p| format!(", predicate={p}"))
.unwrap_or_default();
- let pruning_predicate_string = self
- .pruning_predicate
- .as_ref()
- .map(|pre| {
- let mut guarantees = pre
- .literal_guarantees()
- .iter()
- .map(|item| format!("{}", item))
- .collect_vec();
- guarantees.sort();
- format!(
- ", pruning_predicate={}, required_guarantees=[{}]",
- pre.predicate_expr(),
- guarantees.join(", ")
- )
- })
- .unwrap_or_default();
Review Comment:
I think it is important to keep these in the physical plans -- in particular what I think is important is to be able to check via the explain plan if pruning is happening by looking at the explain plan
##########
datafusion/datasource/src/file_format.rs:
##########
@@ -109,37 +108,10 @@ pub trait FileFormat: Send + Sync + fmt::Debug {
not_impl_err!("Writer not implemented for this format")
}
- /// Check if the specified file format has support for pushing down the provided filters within
- /// the given schemas. Added initially to support the Parquet file format's ability to do this.
- fn supports_filters_pushdown(
Review Comment:
yes I agree
Since [`FileFormat`](https://docs.rs/datafusion/latest/datafusion/datasource/file_format/trait.FileFormat.html) is a pub trait, this is technically a breaking API change, but I do think it was a parquet specific optimization
I recommend we mark this PR as an API change and add a note to the upgrade guide https://github.com/apache/datafusion/blob/main/docs/source/library-user-guide/upgrading.md
I think it should basically say if you implemented `FileFormat` (which probably no one did) and ListingTable you will have to implement the newly added `ExecutionPlan::try_pushdown_filter` method into your execution plan directly if you want the filters to be pushed down
##########
datafusion/core/src/datasource/listing/table.rs:
##########
@@ -982,18 +980,6 @@ impl TableProvider for ListingTable {
return Ok(TableProviderFilterPushDown::Exact);
}
- // if we can't push it down completely with only the filename-based/path-based
- // column names, then we should check if we can do parquet predicate pushdown
- let supports_pushdown = self.options.format.supports_filters_pushdown(
- &self.file_schema,
- &self.table_schema,
- &[filter],
- )?;
-
- if supports_pushdown == FilePushdownSupport::Supported {
- return Ok(TableProviderFilterPushDown::Exact);
- }
Review Comment:
> I have one question: aren't we expecting/preparing for, people to use ListingTable if they read Parquet files? Are we eventually planning to remove all format-specific handlings? Or this is a case only for filter pushdown?
For what it is worth, we (InfluxData) doesn't use ListingTable to read parquet files, instead we provide our own equivalent and create the DataSourceExec's directly
> I would keep supports_filters_pushdown so that TableProviders can do Exact pruning of filters, e.g. using partition columns.
Yes I think that is important too -- I don't think we should be removing any APIs from ListingTable
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2070806936
##########
datafusion/sqllogictest/test_files/push_down_filter.slt:
##########
@@ -218,43 +219,57 @@ LOCATION 'test_files/scratch/push_down_filter/t.parquet';
query TT
explain select a from t where a = '100';
----
-logical_plan TableScan: t projection=[a], full_filters=[t.a = Int32(100)]
+logical_plan
+01)Filter: t.a = Int32(100)
+02)--TableScan: t projection=[a], partial_filters=[t.a = Int32(100)]
# The predicate should not have a column cast when the value is a valid i32
query TT
explain select a from t where a != '100';
----
-logical_plan TableScan: t projection=[a], full_filters=[t.a != Int32(100)]
+logical_plan
+01)Filter: t.a != Int32(100)
+02)--TableScan: t projection=[a], partial_filters=[t.a != Int32(100)]
Review Comment:
I recommend we change these tests to show the physical plans (not the logical plans) as that would more accurately show the pushdown happening. Maybe also something we could do as a separate PR
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -81,11 +81,15 @@ EXPLAIN select a from t_pushdown where b > 2 ORDER BY a;
----
logical_plan
01)Sort: t_pushdown.a ASC NULLS LAST
-02)--TableScan: t_pushdown projection=[a], full_filters=[t_pushdown.b > Int32(2)]
+02)--Projection: t_pushdown.a
+03)----Filter: t_pushdown.b > Int32(2)
+04)------TableScan: t_pushdown projection=[a, b], partial_filters=[t_pushdown.b > Int32(2)]
physical_plan
01)SortPreservingMergeExec: [a@0 ASC NULLS LAST]
02)--SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]
-03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 2, required_guarantees=[]
+03)----CoalesceBatchesExec: target_batch_size=8192
+04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=2
+05)--------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2 AND b@1 > 2
Review Comment:
I agree it is important to remove the coalesce / repartition
Maybe we can make a separate PR to move the filter pushdown code earlier in the physical planning
An alternate could be to update the filter pushdown optimizer pass somehow to remove these -- but I think it would be cleaner / easier to understand if they were never added in the first place
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -81,11 +81,15 @@ EXPLAIN select a from t_pushdown where b > 2 ORDER BY a;
----
logical_plan
01)Sort: t_pushdown.a ASC NULLS LAST
-02)--TableScan: t_pushdown projection=[a], full_filters=[t_pushdown.b > Int32(2)]
+02)--Projection: t_pushdown.a
+03)----Filter: t_pushdown.b > Int32(2)
+04)------TableScan: t_pushdown projection=[a, b], partial_filters=[t_pushdown.b > Int32(2)]
Review Comment:
I think it makes sense and is ok that the logical plans show the filter not pushed down
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -589,4 +559,49 @@ impl FileSource for ParquetSource {
}
}
}
+
+ fn try_pushdown_filters(
+ &self,
+ fd: FilterDescription,
+ config: &datafusion_common::config::ConfigOptions,
+ ) -> datafusion_common::Result<FilterPushdownResult<Arc<dyn FileSource>>> {
+ let Some(file_schema) = self.file_schema.clone() else {
Review Comment:
maybe we file a follow on ticket
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2070335984
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -559,25 +549,8 @@ impl FileSource for ParquetSource {
.predicate()
.map(|p| format!(", predicate={p}"))
.unwrap_or_default();
- let pruning_predicate_string = self
- .pruning_predicate
- .as_ref()
- .map(|pre| {
- let mut guarantees = pre
- .literal_guarantees()
- .iter()
- .map(|item| format!("{}", item))
- .collect_vec();
- guarantees.sort();
- format!(
- ", pruning_predicate={}, required_guarantees=[{}]",
- pre.predicate_expr(),
- guarantees.join(", ")
- )
- })
- .unwrap_or_default();
Review Comment:
Hmm okay. I'll see if I can make it happen...
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#issuecomment-2822076779
https://github.com/apache/datafusion/issues/15812 surfaced another reason why building the predicates from the files schemas is necessary
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2052746932
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -81,11 +81,15 @@ EXPLAIN select a from t_pushdown where b > 2 ORDER BY a;
----
logical_plan
01)Sort: t_pushdown.a ASC NULLS LAST
-02)--TableScan: t_pushdown projection=[a], full_filters=[t_pushdown.b > Int32(2)]
+02)--Projection: t_pushdown.a
+03)----Filter: t_pushdown.b > Int32(2)
+04)------TableScan: t_pushdown projection=[a, b], partial_filters=[t_pushdown.b > Int32(2)]
physical_plan
01)SortPreservingMergeExec: [a@0 ASC NULLS LAST]
02)--SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]
-03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 2, required_guarantees=[]
+03)----CoalesceBatchesExec: target_batch_size=8192
+04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=2
+05)--------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2 AND b@1 > 2
Review Comment:
> PushdownFilter should probably work before distribution&order satisfiers
That makes sense to me. It does more "invasive" re-arranging of plans than those do.
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#issuecomment-2848681802
If you can make a PR to change the order of the rules and open that ticket I would appreciate it I won't be able to for several hours 🙏
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2072538338
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -81,11 +81,15 @@ EXPLAIN select a from t_pushdown where b > 2 ORDER BY a;
----
logical_plan
01)Sort: t_pushdown.a ASC NULLS LAST
-02)--TableScan: t_pushdown projection=[a], full_filters=[t_pushdown.b > Int32(2)]
+02)--Projection: t_pushdown.a
+03)----Filter: t_pushdown.b > Int32(2)
+04)------TableScan: t_pushdown projection=[a, b], partial_filters=[t_pushdown.b > Int32(2)]
physical_plan
01)SortPreservingMergeExec: [a@0 ASC NULLS LAST]
02)--SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]
-03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 2, required_guarantees=[]
+03)----CoalesceBatchesExec: target_batch_size=8192
+04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=2
+05)--------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2 AND b@1 > 2
Review Comment:
I opened https://github.com/apache/datafusion/issues/15938
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2074193501
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -221,5 +226,23 @@ physical_plan DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion
query TT
EXPLAIN select * from t_pushdown where val != 'c';
----
-logical_plan TableScan: t_pushdown projection=[val, part], full_filters=[t_pushdown.val != Utf8("c")]
+logical_plan
+01)Filter: t_pushdown.val != Utf8("c")
+02)--TableScan: t_pushdown projection=[val, part], partial_filters=[t_pushdown.val != Utf8("c")]
physical_plan DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_part_test/part=a/file.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_part_test/part=b/file.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_part_test/part=c/file.parquet]]}, projection=[val, part], file_type=parquet, predicate=val@0 != c, pruning_predicate=val_null_count@2 != row_count@3 AND (val_min@0 != c OR c != val_max@1), required_guarantees=[val not in (c)]
+
+# If we have a mix of filters:
+# - The partition filters get evaluated during planning
+# - The mixed filters end up in a FilterExec
+# - The file filters get pushed down into the scan
+query TT
+EXPLAIN select * from t_pushdown where val != 'd' AND val != 'c' AND part = 'a' AND part != val;
+----
+logical_plan
+01)Filter: t_pushdown.val != Utf8("d") AND t_pushdown.val != Utf8("c") AND t_pushdown.val != t_pushdown.part
+02)--TableScan: t_pushdown projection=[val, part], full_filters=[t_pushdown.part = Utf8("a")], partial_filters=[t_pushdown.val != Utf8("d"), t_pushdown.val != Utf8("c"), t_pushdown.val != t_pushdown.part]
+physical_plan
+01)CoalesceBatchesExec: target_batch_size=8192
+02)--FilterExec: val@0 != part@1
+03)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_part_test/part=a/file.parquet]]}, projection=[val, part], file_type=parquet, predicate=val@0 != d AND val@0 != c, pruning_predicate=val_null_count@2 != row_count@3 AND (val_min@0 != d OR d != val_max@1) AND val_null_count@2 != row_count@3 AND (val_min@0 != c OR c != val_max@1), required_guarantees=[val not in (c, d)]
Review Comment:
Replaces deleted test
##########
datafusion/core/tests/sql/path_partition.rs:
##########
@@ -57,55 +53,6 @@ use object_store::{
use object_store::{Attributes, MultipartUpload, PutMultipartOpts, PutPayload};
use url::Url;
-#[tokio::test]
-async fn parquet_partition_pruning_filter() -> Result<()> {
Review Comment:
It was relying on `ListingTable` to do the pushdown without the optimizer pass.
I felt that it was better suited for an `.slt` test that is fully wired up, so I essentially moved it: https://github.com/apache/datafusion/pull/15769/files#r2074193501
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2075028834
##########
datafusion/core/tests/sql/path_partition.rs:
##########
@@ -57,55 +53,6 @@ use object_store::{
use object_store::{Attributes, MultipartUpload, PutMultipartOpts, PutPayload};
use url::Url;
-#[tokio::test]
-async fn parquet_partition_pruning_filter() -> Result<()> {
Review Comment:
~can we also apply the changes similar to the test in `datafusion/core/tests/parquet/file_statistics.rs ` to keep this test for now?~
I saw the new slt now 👍🏻
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#issuecomment-2854063717
Will be merged after resolving conflicts 🚀
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada merged PR #15769:
URL: https://github.com/apache/datafusion/pull/15769
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2051611694
##########
datafusion/core/src/datasource/physical_plan/parquet.rs:
##########
@@ -148,7 +149,7 @@ mod tests {
let mut source = ParquetSource::default();
if let Some(predicate) = predicate {
- source = source.with_predicate(Arc::clone(&file_schema), predicate);
+ source = source.with_predicate(predicate);
Review Comment:
This seemed like an easy win since I was able to just change this so that the schema is _always_ passed in by the `FileSourceConfigBuilder` instead of only when `with_predicate` is called.
This was necessary becasue `with_predicate` is no longer called to attach a predicate, instaed it happens during an optimization pass so `ParquetSource` neesd to have it available at that point.
I left `with_predicate` in there to avoid churn and in case there is a use case for attaching a predicate directly through the scan instad of a as a FilterExec that later gets pushed into the scan.
##########
datafusion/core/src/datasource/listing/table.rs:
##########
@@ -982,18 +980,6 @@ impl TableProvider for ListingTable {
return Ok(TableProviderFilterPushDown::Exact);
}
- // if we can't push it down completely with only the filename-based/path-based
- // column names, then we should check if we can do parquet predicate pushdown
- let supports_pushdown = self.options.format.supports_filters_pushdown(
- &self.file_schema,
- &self.table_schema,
- &[filter],
- )?;
-
- if supports_pushdown == FilePushdownSupport::Supported {
- return Ok(TableProviderFilterPushDown::Exact);
- }
Review Comment:
The point of this PR is that this moves from being something specialized that `ListingTable` does to anything that works for any TableProvider / they don't need to do anything special! The checks for compatibility also happen all within the parquet data source machinery, instead of leaking implementations via `supports_filters_pushdown`.
##########
datafusion/datasource-parquet/src/row_filter.rs:
##########
@@ -649,73 +649,45 @@ mod test {
#[test]
fn nested_data_structures_prevent_pushdown() {
- let table_schema = get_basic_table_schema();
-
- let file_schema = Schema::new(vec![Field::new(
- "list_col",
- DataType::Struct(Fields::empty()),
- true,
- )]);
Review Comment:
This test was wrong! It wanted to test that `list_col` prevents pushdown because it's a nested type. Instead it was prevented because `list_col` is not in the table / schema!
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -587,4 +560,49 @@ impl FileSource for ParquetSource {
}
}
}
+
+ fn try_pushdown_filters(
Review Comment:
cc @berkaysynnada for this implementation
##########
datafusion/datasource-parquet/src/mod.rs:
##########
@@ -244,7 +242,7 @@ impl ParquetExecBuilder {
inner: DataSourceExec::new(Arc::new(base_config.clone())),
base_config,
predicate,
- pruning_predicate: parquet.pruning_predicate,
+ pruning_predicate: None, // for backwards compat since `ParquetExec` is only for backwards compat anyway
Review Comment:
Open to other suggestions (i.e. removing it). I felt like this minimizes breakage for folks still using `ParquetExec`, who are likely the same folks that want to do the least amount of work possible to upgrade.
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -81,11 +81,15 @@ EXPLAIN select a from t_pushdown where b > 2 ORDER BY a;
----
logical_plan
01)Sort: t_pushdown.a ASC NULLS LAST
-02)--TableScan: t_pushdown projection=[a], full_filters=[t_pushdown.b > Int32(2)]
+02)--Projection: t_pushdown.a
+03)----Filter: t_pushdown.b > Int32(2)
+04)------TableScan: t_pushdown projection=[a, b], partial_filters=[t_pushdown.b > Int32(2)]
Review Comment:
This is because the pushdown no longer happens at the logical level - it happens at the physical level. This makes sense, in part because the checks for suitability of pushdown are better at the physical level (there may be reasons to reject a pushdown at the physical level that are not present at the logical level, e.g. partition columns or encodings).
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -559,25 +549,8 @@ impl FileSource for ParquetSource {
.predicate()
.map(|p| format!(", predicate={p}"))
.unwrap_or_default();
- let pruning_predicate_string = self
- .pruning_predicate
- .as_ref()
- .map(|pre| {
- let mut guarantees = pre
- .literal_guarantees()
- .iter()
- .map(|item| format!("{}", item))
- .collect_vec();
- guarantees.sort();
- format!(
- ", pruning_predicate={}, required_guarantees=[{}]",
- pre.predicate_expr(),
- guarantees.join(", ")
- )
- })
- .unwrap_or_default();
Review Comment:
In https://github.com/apache/datafusion/pull/15561#pullrequestreview-2741008179 Andrew asked me to keep this, but now since the schema isn't even being passed in to `with_predicate` it's going to be hard to keep these. I suggest we just accept that they won't be present in the physical plans. If that's not okay what I could do is generate them on the fly in `fmt_extra` or generate them if `with_predicate` is called with a schema or `with_schema` is called with a predicate. But I'd like to avoid that unless someone thinks is worth it or has another suggestion.
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -81,11 +81,15 @@ EXPLAIN select a from t_pushdown where b > 2 ORDER BY a;
----
logical_plan
01)Sort: t_pushdown.a ASC NULLS LAST
-02)--TableScan: t_pushdown projection=[a], full_filters=[t_pushdown.b > Int32(2)]
+02)--Projection: t_pushdown.a
+03)----Filter: t_pushdown.b > Int32(2)
+04)------TableScan: t_pushdown projection=[a, b], partial_filters=[t_pushdown.b > Int32(2)]
physical_plan
01)SortPreservingMergeExec: [a@0 ASC NULLS LAST]
02)--SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]
-03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 2, required_guarantees=[]
+03)----CoalesceBatchesExec: target_batch_size=8192
+04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=2
+05)--------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2 AND b@1 > 2
Review Comment:
@berkaysynnada any idea why we have extra `CoalesceBatchesExec` and `RepartitionExec` now?
##########
datafusion/datasource/src/file_format.rs:
##########
@@ -109,37 +108,10 @@ pub trait FileFormat: Send + Sync + fmt::Debug {
not_impl_err!("Writer not implemented for this format")
}
- /// Check if the specified file format has support for pushing down the provided filters within
- /// the given schemas. Added initially to support the Parquet file format's ability to do this.
- fn supports_filters_pushdown(
Review Comment:
Binning specialized code that was also leaking parquet stuff through DataSource and into TableProvider 😄
##########
datafusion/datasource/src/file_format.rs:
##########
@@ -109,37 +108,10 @@ pub trait FileFormat: Send + Sync + fmt::Debug {
not_impl_err!("Writer not implemented for this format")
}
- /// Check if the specified file format has support for pushing down the provided filters within
- /// the given schemas. Added initially to support the Parquet file format's ability to do this.
- fn supports_filters_pushdown(
- &self,
- _file_schema: &Schema,
- _table_schema: &Schema,
- _filters: &[&Expr],
- ) -> Result<FilePushdownSupport> {
- Ok(FilePushdownSupport::NoSupport)
- }
-
/// Return the related FileSource such as `CsvSource`, `JsonSource`, etc.
fn file_source(&self) -> Arc<dyn FileSource>;
}
-/// An enum to distinguish between different states when determining if certain filters can be
-/// pushed down to file scanning
-#[derive(Debug, PartialEq)]
-pub enum FilePushdownSupport {
Review Comment:
Another one of these enums!
--
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] Implement Parquet filter pushdown via new filter pushdown APIs [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15769:
URL: https://github.com/apache/datafusion/pull/15769#discussion_r2163855317
##########
datafusion/datasource/src/file_format.rs:
##########
@@ -94,7 +93,6 @@ pub trait FileFormat: Send + Sync + fmt::Debug {
&self,
state: &dyn Session,
conf: FileScanConfig,
- filters: Option<&Arc<dyn PhysicalExpr>>,
Review Comment:
Agreed. Sorry for the breakage. Do the new APIs get the job done for you, or is there something now 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