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 20:00:19 UTC
[PR] TopK dynamic filter pushdown attempt 2 [datafusion]
adriangb opened a new pull request, #15770:
URL: https://github.com/apache/datafusion/pull/15770
(no comment)
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2051499380
##########
datafusion/core/tests/physical_optimizer/push_down_filter/mod.rs:
##########
@@ -0,0 +1,401 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::{Arc, LazyLock};
+
+use arrow::{
+ array::record_batch,
+ datatypes::{DataType, Field, Schema, SchemaRef},
+};
+use arrow_schema::SortOptions;
+use datafusion::{
+ logical_expr::Operator,
+ physical_plan::{
+ expressions::{BinaryExpr, Column, Literal},
+ PhysicalExpr,
+ },
+ prelude::{SessionConfig, SessionContext},
+ scalar::ScalarValue,
+};
+use datafusion_common::config::ConfigOptions;
+use datafusion_execution::object_store::ObjectStoreUrl;
+use datafusion_expr::test::function_stub::count_udaf;
+use datafusion_physical_expr::expressions::col;
+use datafusion_physical_expr::{aggregate::AggregateExprBuilder, Partitioning};
+use datafusion_physical_expr::{LexOrdering, PhysicalSortExpr};
+use datafusion_physical_optimizer::{
+ push_down_filter::PushdownFilter, PhysicalOptimizerRule,
+};
+use datafusion_physical_plan::{
+ aggregates::{AggregateExec, AggregateMode, PhysicalGroupBy},
+ coalesce_batches::CoalesceBatchesExec,
+ filter::FilterExec,
+ repartition::RepartitionExec,
+};
+use datafusion_physical_plan::{sorts::sort::SortExec, ExecutionPlan};
+
+use futures::StreamExt;
+use object_store::memory::InMemory;
+use util::{format_plan_for_test, OptimizationTest, TestScanBuilder};
+
+mod util;
+
+#[test]
+fn test_pushdown_into_scan() {
+ let scan = TestScanBuilder::new(schema()).with_support(true).build();
+ let predicate = col_lit_predicate("a", "foo", &schema());
+ let plan = Arc::new(FilterExec::try_new(predicate, scan).unwrap());
+
+ // expect the predicate to be pushed down into the DataSource
+ insta::assert_snapshot!(
+ OptimizationTest::new(plan, PushdownFilter{}, true),
+ @r"
+ OptimizationTest:
+ input:
+ - FilterExec: a@0 = foo
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
+ output:
+ Ok:
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true, predicate=a@0 = foo
+ "
+ );
+}
+
+/// Show that we can use config options to determine how to do pushdown.
+#[test]
+fn test_pushdown_into_scan_with_config_options() {
+ let scan = TestScanBuilder::new(schema()).with_support(true).build();
+ let predicate = col_lit_predicate("a", "foo", &schema());
+ let plan = Arc::new(FilterExec::try_new(predicate, scan).unwrap()) as _;
+
+ let mut cfg = ConfigOptions::default();
+ insta::assert_snapshot!(
+ OptimizationTest::new(
+ Arc::clone(&plan),
+ PushdownFilter {},
+ false
+ ),
+ @r"
+ OptimizationTest:
+ input:
+ - FilterExec: a@0 = foo
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
+ output:
+ Ok:
+ - FilterExec: a@0 = foo
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
+ "
+ );
+
+ cfg.execution.parquet.pushdown_filters = true;
+ insta::assert_snapshot!(
+ OptimizationTest::new(
+ plan,
+ PushdownFilter {},
+ true
+ ),
+ @r"
+ OptimizationTest:
+ input:
+ - FilterExec: a@0 = foo
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
+ output:
+ Ok:
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true, predicate=a@0 = foo
+ "
+ );
+}
+
+#[test]
+fn test_filter_collapse() {
+ // filter should be pushed down into the parquet scan with two filters
+ let scan = TestScanBuilder::new(schema()).with_support(true).build();
+ let predicate1 = col_lit_predicate("a", "foo", &schema());
+ let filter1 = Arc::new(FilterExec::try_new(predicate1, scan).unwrap());
+ let predicate2 = col_lit_predicate("b", "bar", &schema());
+ let plan = Arc::new(FilterExec::try_new(predicate2, filter1).unwrap());
+
+ insta::assert_snapshot!(
+ OptimizationTest::new(plan, PushdownFilter{}, true),
+ @r"
+ OptimizationTest:
+ input:
+ - FilterExec: b@1 = bar
+ - FilterExec: a@0 = foo
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
+ output:
+ Ok:
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true, predicate=b@1 = bar AND a@0 = foo
+ "
+ );
+}
+
+#[test]
+fn test_filter_with_projection() {
+ let scan = TestScanBuilder::new(schema()).with_support(true).build();
+ let projection = vec![1, 0];
+ let predicate = col_lit_predicate("a", "foo", &schema());
+ let plan = Arc::new(
+ FilterExec::try_new(predicate, Arc::clone(&scan))
+ .unwrap()
+ .with_projection(Some(projection))
+ .unwrap(),
+ );
+
+ // expect the predicate to be pushed down into the DataSource but the FilterExec to be converted to ProjectionExec
+ insta::assert_snapshot!(
+ OptimizationTest::new(plan, PushdownFilter{}, true),
+ @r"
+ OptimizationTest:
+ input:
+ - FilterExec: a@0 = foo, projection=[b@1, a@0]
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
+ output:
+ Ok:
+ - ProjectionExec: expr=[b@1 as b, a@0 as a]
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true, predicate=a@0 = foo
+ ",
+ );
+
+ // add a test where the filter is on a column that isn't included in the output
+ let projection = vec![1];
+ let predicate = col_lit_predicate("a", "foo", &schema());
+ let plan = Arc::new(
+ FilterExec::try_new(predicate, scan)
+ .unwrap()
+ .with_projection(Some(projection))
+ .unwrap(),
+ );
+ insta::assert_snapshot!(
+ OptimizationTest::new(plan, PushdownFilter{},true),
+ @r"
+ OptimizationTest:
+ input:
+ - FilterExec: a@0 = foo, projection=[b@1]
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
+ output:
+ Ok:
+ - ProjectionExec: expr=[b@1 as b]
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true, predicate=a@0 = foo
+ "
+ );
+}
+
+#[test]
+fn test_push_down_through_transparent_nodes() {
+ // expect the predicate to be pushed down into the DataSource
+ let scan = TestScanBuilder::new(schema()).with_support(true).build();
+ let coalesce = Arc::new(CoalesceBatchesExec::new(scan, 1));
+ let predicate = col_lit_predicate("a", "foo", &schema());
+ let filter = Arc::new(FilterExec::try_new(predicate, coalesce).unwrap());
+ let repartition = Arc::new(
+ RepartitionExec::try_new(filter, Partitioning::RoundRobinBatch(1)).unwrap(),
+ );
+ let predicate = col_lit_predicate("b", "bar", &schema());
+ let plan = Arc::new(FilterExec::try_new(predicate, repartition).unwrap());
+
+ // expect the predicate to be pushed down into the DataSource
+ insta::assert_snapshot!(
+ OptimizationTest::new(plan, PushdownFilter{},true),
+ @r"
+ OptimizationTest:
+ input:
+ - FilterExec: b@1 = bar
+ - RepartitionExec: partitioning=RoundRobinBatch(1), input_partitions=1
+ - FilterExec: a@0 = foo
+ - CoalesceBatchesExec: target_batch_size=1
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
+ output:
+ Ok:
+ - RepartitionExec: partitioning=RoundRobinBatch(1), input_partitions=1
+ - CoalesceBatchesExec: target_batch_size=1
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true, predicate=b@1 = bar AND a@0 = foo
+ "
+ );
+}
+
+#[test]
+fn test_no_pushdown_through_aggregates() {
+ // There are 2 important points here:
+ // 1. The outer filter **is not** pushed down at all because we haven't implemented pushdown support
+ // yet for AggregateExec.
+ // 2. The inner filter **is** pushed down into the DataSource.
+ let scan = TestScanBuilder::new(schema()).with_support(true).build();
+
+ let coalesce = Arc::new(CoalesceBatchesExec::new(scan, 10));
+
+ let filter = Arc::new(
+ FilterExec::try_new(col_lit_predicate("a", "foo", &schema()), coalesce).unwrap(),
+ );
+
+ let aggregate_expr =
+ vec![
+ AggregateExprBuilder::new(count_udaf(), vec![col("a", &schema()).unwrap()])
+ .schema(schema())
+ .alias("cnt")
+ .build()
+ .map(Arc::new)
+ .unwrap(),
+ ];
+ let group_by = PhysicalGroupBy::new_single(vec![
+ (col("a", &schema()).unwrap(), "a".to_string()),
+ (col("b", &schema()).unwrap(), "b".to_string()),
+ ]);
+ let aggregate = Arc::new(
+ AggregateExec::try_new(
+ AggregateMode::Final,
+ group_by,
+ aggregate_expr.clone(),
+ vec![None],
+ filter,
+ schema(),
+ )
+ .unwrap(),
+ );
+
+ let coalesce = Arc::new(CoalesceBatchesExec::new(aggregate, 100));
+
+ let predicate = col_lit_predicate("b", "bar", &schema());
+ let plan = Arc::new(FilterExec::try_new(predicate, coalesce).unwrap());
+
+ // expect the predicate to be pushed down into the DataSource
+ insta::assert_snapshot!(
+ OptimizationTest::new(plan, PushdownFilter{}, true),
+ @r"
+ OptimizationTest:
+ input:
+ - FilterExec: b@1 = bar
+ - CoalesceBatchesExec: target_batch_size=100
+ - AggregateExec: mode=Final, gby=[a@0 as a, b@1 as b], aggr=[cnt], ordering_mode=PartiallySorted([0])
+ - FilterExec: a@0 = foo
+ - CoalesceBatchesExec: target_batch_size=10
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
+ output:
+ Ok:
+ - FilterExec: b@1 = bar
+ - CoalesceBatchesExec: target_batch_size=100
+ - AggregateExec: mode=Final, gby=[a@0 as a, b@1 as b], aggr=[cnt]
+ - CoalesceBatchesExec: target_batch_size=10
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true, predicate=a@0 = foo
+ "
+ );
+}
+
+#[tokio::test]
+async fn test_topk_dynamic_filter_pushdown() {
+ // This test is a bit of a hack, but it shows that we can push down dynamic filters
+ // into the DataSourceExec. The test is not perfect because we don't have a real
+ // implementation of the dynamic filter yet, so we just use a static filter.
+ let batches = vec![
+ record_batch!(
+ ("a", Utf8, ["aa", "ab"]),
+ ("b", Utf8, ["bd", "bc"]),
+ ("c", Float64, [1.0, 2.0])
+ )
+ .unwrap(),
+ record_batch!(
+ ("a", Utf8, ["ac", "ad"]),
+ ("b", Utf8, ["bb", "ba"]),
+ ("c", Float64, [2.0, 1.0])
+ )
+ .unwrap(),
+ ];
+ let scan = TestScanBuilder::new(schema())
+ .with_support(true)
+ .with_batches(batches)
+ .build();
+ let plan = Arc::new(
+ SortExec::new(
+ LexOrdering::new(vec![PhysicalSortExpr::new(
+ col("b", &schema()).unwrap(),
+ SortOptions::new(true, false), // descending, nulls_first
+ )]),
+ Arc::clone(&scan),
+ )
+ .with_fetch(Some(1)),
+ ) as Arc<dyn ExecutionPlan>;
+
+ // expect the predicate to be pushed down into the DataSource
+ insta::assert_snapshot!(
+ OptimizationTest::new(Arc::clone(&plan), PushdownFilter{}, true),
+ @r"
+ OptimizationTest:
+ input:
+ - SortExec: TopK(fetch=1), expr=[b@1 DESC NULLS LAST], preserve_partitioning=[false]
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
+ output:
+ Ok:
+ - SortExec: TopK(fetch=1), expr=[b@1 DESC NULLS LAST], preserve_partitioning=[false]
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true, predicate=DynamicFilterPhysicalExpr [ true ]
+ "
+ );
+
+ // Actually apply the optimization to the plan
Review Comment:
I recognize these diverge a bit from other tests, happy to move them somewhere better....
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2051461834
##########
datafusion/physical-expr/src/expressions/mod.rs:
##########
@@ -22,7 +22,7 @@ mod binary;
mod case;
mod cast;
mod column;
-mod dynamic_filters;
+pub mod dynamic_filters;
Review Comment:
This bit has me tripped up. I'm not sure where the right place to put `dynamic_filters` is such that it's public for our internal use in operators but private from the outside world π€
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -1205,9 +1221,10 @@ impl ExecutionPlan for SortExec {
self: Arc<Self>,
children: Vec<Arc<dyn ExecutionPlan>>,
) -> Result<Arc<dyn ExecutionPlan>> {
- let new_sort = SortExec::new(self.expr.clone(), Arc::clone(&children[0]))
+ let mut new_sort = SortExec::new(self.expr.clone(), Arc::clone(&children[0]))
.with_fetch(self.fetch)
.with_preserve_partitioning(self.preserve_partitioning);
+ new_sort.filter = Arc::clone(&self.filter);
Review Comment:
I missed this for a while and spent an hour trying to figure out why my test was failing. IMO we should have a test that enforces the invariant that `ExecutionPlan::with_new_children(Arc::clone(&node), node.children()) == node`
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2977736578
> > Sounds reasonable. Perhaps just keep it this way and see if we somehow can separate it to 2 different passes later.
> > It seems like the branch [pydantic#30](https://github.com/pydantic/datafusion/pull/30) passes?
>
> Then since you think that's reasonable I'd like to merge it and go with that instead for now even if we're not 100% sold on either direction it will let us keep iterating.
>
> I'll wait for your approval before merging this PR into main.
I think it's great! I added one suggestion to minimize the changes to the explain output and maybe phrase the "pre" and "post" more as "normal" filter pushdown and "dynamic" filter pushdown? I think that might explain a bit more of what might be happening.
Maybe @alamb can verify the (increased) performance improvements once more?
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2971441638
> We might be able to split the filter pushdown into two steps: static (cannot make assumptions about reference links but can modify the plan tree, e.g. for FitlerExec) and dynamic (can make reference links but cannot modify the plan tree, e.g. SortExec and HashJoin).
@alamb I had not thought of this before but it does seem interesting. we'd basically add a new (simpler) rule called `PushDownDynamicFilters` or something like that. we'd also have to add a new method to `ExecutionPlan::gather_dynamic_filters_for_pushdown` and `SortExec` / `HashJoinExec` / any operator that wants to push down a filter that depends on the plan's structure / has references back to itself uses this method instead of the existing pushdown methods. then this can be run after `EnforceSorting`. I think the main con to this would be if an operator wanted to push down a dyanmic filter _and_ modify the plan structure. I don't think it will want to remove itself from the plan structure like `FilterExec` does, but I could see it wanting to add a node (e.g. if we implemented https://github.com/apache/datafusion/pull/15697 by having `SortExec` insert a `FilterExec` underneath itself). the big pro here is that I think it makes a clear distinction of capabilities and resolves th
e fragility of rewriting a plan while keeping track of references between nodes.
what do you think of that approach vs. the approach we have here of teaching `EnforceSorting` how to pass around dynamic filters?
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2145924185
##########
datafusion/physical-optimizer/src/enforce_sorting/sort_pushdown.rs:
##########
@@ -114,6 +118,18 @@ fn pushdown_sorts_helper(
sort_push_down.data.fetch = fetch;
sort_push_down.data.ordering_requirement =
Some(OrderingRequirements::from(sort_ordering));
+ let filter = plan
+ .as_any()
+ .downcast_ref::<SortExec>()
+ .and_then(|s| s.filter().clone());
Review Comment:
I see what you mean now. It would be `ExecutionPlan::dynamic_filters(&self) -> Arc<DynamicFilterPhysicalExpr>` which is fully typed. That sounds reasonable to me π.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2971173365
> > QQuery 25 β 380.03 ms β 279.23 ms β +1.36x faster
>
> ```sql
> SELECT SearchPhrase FROM hits WHERE SearchPhrase <> '' ORDER BY SearchPhrase LIMIT 10;
> ```
>
> Looks like it might be real!
>
> And obviously we expect Q23 to get a huge bump once we enable filter pushdown...
>
> I think the only thing left here is for someone to review in detail π
I can reproduce this improvement locally.
I tested using
```shell
$ cat q25.sql
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10;
$ ./datafusion-cli-topk-dynamic-filters -f q25.sql
...
```
main (merge-base)
```
Elapsed 0.209 seconds.
Elapsed 0.214 seconds.
Elapsed 0.210 seconds.
```
This branch
```
Elapsed 0.173 seconds.
Elapsed 0.182 seconds.
Elapsed 0.178 seconds.
```
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2145895957
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -246,38 +246,3 @@ physical_plan
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)]
-
-# The order of filters should not matter
Review Comment:
Might be accidental / from merging. Will 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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2053476865
##########
datafusion/physical-optimizer/src/push_down_filter.rs:
##########
@@ -382,7 +383,7 @@ impl PhysicalOptimizerRule for PushdownFilter {
context
.transform_up(|node| {
- if node.plan.as_any().downcast_ref::<FilterExec>().is_some() {
+ if node.plan.as_any().downcast_ref::<FilterExec>().is_some() || node.plan.as_any().downcast_ref::<SortExec>().is_some() {
Review Comment:
Needs fixing of some failing tests, cleanup of the plethora of helper methods I added and a lot of docs but here's the idea: 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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2852350646
I think some tweaks will be needed based on https://github.com/apache/datafusion/pull/15769/files#r2074207291
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2083518706
##########
datafusion/sqllogictest/test_files/prepare.slt:
##########
@@ -264,16 +264,19 @@ WHERE run_id = 'foo'
ORDER BY random()
LIMIT $1
-query I
+query error
EXECUTE get_N_rand_ints_from_last_run(1);
----
-1
+DataFusion error: Internal error: PhysicalOptimizer rule 'LimitPushdown' failed. Schema mismatch. Expected original schema: Schema { fields: [Field { name: "id", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }], metadata: {} }, got new schema: Schema { fields: [Field { name: "id", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "run_id", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }], metadata: {} }.
Review Comment:
Perhaps @jonahgao can make a quick diagnose here, as I cannot debug these EXECUTE's via "explain" or "explain verbose". I'm also open if someone has any idea how these changes break the EXECUTE feature
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2971531318
> I don't understand why FilterPushdown can't also push down DynamicFlters if it was run after EnforceSorting but I vaguely remember it being discussed and rejected before
It could (and does currently) handle push down of dynamic filters, the issue is that *it cannot be run after EnforceSorting* because EnforceSorting and EnforceDistribution need to be run after e.g. a `FilterExec` is removed from the plan.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2972769559
Combined with changes in my PR (https://github.com/apache/datafusion/pull/16408), this is looking sweet for the TopK benchmarks:
```
Benchmark run_topk_tpch.json
--------------------
ββββββββββββββββ³ββββββββββββ³ββββββββββββββββββββ³ββββββββββββββββ
β Query β main β topk-apply-filter β Change β
β‘βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Q1 β 28.59 ms β 16.69 ms β +1.71x faster β
β Q2 β 30.36 ms β 20.65 ms β +1.47x faster β
β Q3 β 101.92 ms β 54.89 ms β +1.86x faster β
β Q4 β 27.61 ms β 22.37 ms β +1.23x faster β
β Q5 β 19.31 ms β 12.73 ms β +1.52x faster β
β Q6 β 42.45 ms β 26.39 ms β +1.61x faster β
β Q7 β 85.18 ms β 70.72 ms β +1.20x faster β
β Q8 β 103.80 ms β 46.25 ms β +2.24x faster β
β Q9 β 99.54 ms β 63.53 ms β +1.57x faster β
β Q10 β 141.00 ms β 100.62 ms β +1.40x faster β
β Q11 β 78.28 ms β 56.94 ms β +1.37x faster β
ββββββββββββββββ΄ββββββββββββ΄ββββββββββββββββββββ΄ββββββββββββββββ
```
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2818261990
@adriangb I'll complete reviewing this after merging other open PR's.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2978663984
@alamb I think we're ready to merge this and keep chipping away in https://github.com/apache/datafusion/pull/16424 and other spots right?
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2150648173
##########
datafusion/physical-optimizer/src/filter_pushdown.rs:
##########
@@ -362,17 +363,25 @@ use itertools::izip;
/// [`ProjectionExec`]: datafusion_physical_plan::projection::ProjectionExec
/// [`AggregateExec`]: datafusion_physical_plan::aggregates::AggregateExec
#[derive(Debug)]
-pub struct FilterPushdown {}
+pub struct FilterPushdown {
+ phase: FilterPushdownPhase,
+ name: String,
+}
impl FilterPushdown {
- pub fn new() -> Self {
- Self {}
+ fn new(phase: FilterPushdownPhase) -> Self {
+ Self {
+ phase,
+ name: format!("FilterPushdown({phase})"),
Review Comment:
Perhaps we could keep formatting the original rule as `FilterPushdown` and `FilterPushdown(Dynamic)`?
##########
datafusion/physical-optimizer/src/filter_pushdown.rs:
##########
@@ -362,17 +363,25 @@ use itertools::izip;
/// [`ProjectionExec`]: datafusion_physical_plan::projection::ProjectionExec
/// [`AggregateExec`]: datafusion_physical_plan::aggregates::AggregateExec
#[derive(Debug)]
-pub struct FilterPushdown {}
+pub struct FilterPushdown {
+ phase: FilterPushdownPhase,
+ name: String,
+}
impl FilterPushdown {
- pub fn new() -> Self {
- Self {}
+ fn new(phase: FilterPushdownPhase) -> Self {
+ Self {
+ phase,
+ name: format!("FilterPushdown({phase})"),
Review Comment:
Perhaps we could keep formatting the original rule as `FilterPushdown` and the new as`FilterPushdown(Dynamic)`?
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2981406974
<img width="1234" alt="image" src="https://github.com/user-attachments/assets/b96882b9-f3da-4d9a-8635-ba77bdf8fbf3" />
Well, it looks like for these benchmarks 95% is now spent on just scanning the data and only 5% elsewhere, so I guess we need to focus there :D
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2913206532
hi @berkaysynnada any luck getting this last little bit through? anything I can do to help get this across the line?
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2915043264
> hi @berkaysynnada any luck getting this last little bit through? anything I can do to help get this across the line?
Hi @adriangb. I had to shift my focus to another area and havenβt been able to find the time to finish this yet. If you or anyone else has time to investigate the issue, Iβd really appreciate it. Otherwise, Iβll try to wrap it up this weekend. Thanks for your patience and understanding :(
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2965256342
I couldn't find much time on upstream issues lately @adriangb, sorry for that. But I'll certainly take a look at this 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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2972849206
Let's merge your PR into here :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2052761831
##########
datafusion/physical-optimizer/src/push_down_filter.rs:
##########
@@ -382,7 +383,7 @@ impl PhysicalOptimizerRule for PushdownFilter {
context
.transform_up(|node| {
- if node.plan.as_any().downcast_ref::<FilterExec>().is_some() {
+ if node.plan.as_any().downcast_ref::<FilterExec>().is_some() || node.plan.as_any().downcast_ref::<SortExec>().is_some() {
Review Comment:
I propose an API something like this:
```rust
trait ExecutionPlan {
fn gather_filters_for_pushdown(
&self,
parent_filters: &[Arc<dyn ExecutionPlan>],
) -> Result<FilterPushdownPlan> {
let unsupported = vec![FilterPushdownSupport::Unsupported; parent_filters.len()];
Ok(
FilterPushdownPlan {
parent_filters_for_children: vec![unsupported; self.children().len()],
self_filters_for_children: vec![vec![]; self.children().len()],
},
)
}
fn propagate_filter_pushdown(
&self,
parent_pushdown_result: Vec<FilterPushdowChildResult>,
_self_filter_pushdown_result: Vec<FilterPushdowChildResult>,
) -> Result<FilterPushdownPropagation> {
Ok(
FilterPushdownPropagation {
parent_filter_result: parent_pushdown_result,
new_node: None,
},
)
}
}
pub struct FilterPushdownPropagation {
parent_filter_result: Vec<FilterPushdowChildResult>,
new_node: Option<Arc<dyn ExecutionPlan>>,
}
#[derive(Debug, Clone, Copy)]
pub enum FilterPushdowChildResult {
Supported,
Unsupported,
}
impl FilterPushdowChildResult {
}
#[derive(Debug, Clone)]
pub enum FilterPushdownSupport {
Supported(Arc<dyn PhysicalExpr>),
Unsupported,
}
#[derive(Debug, Clone)]
pub struct FilterPushdownPlan {
parent_filters_for_children: Vec<Vec<FilterPushdownSupport>>,
self_filters_for_children: Vec<Vec<FilterPushdownSupport>>,
}
```
The optimizer rule will have to do a bit of bookeeping and slicing correctly but this should avoid the need for any downcast matching or `retry` and minimize clones of plans.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2816673274
> Pausing this until #15769 is done
I was able to unblock by wiring up to TestDataSource
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2053476865
##########
datafusion/physical-optimizer/src/push_down_filter.rs:
##########
@@ -382,7 +383,7 @@ impl PhysicalOptimizerRule for PushdownFilter {
context
.transform_up(|node| {
- if node.plan.as_any().downcast_ref::<FilterExec>().is_some() {
+ if node.plan.as_any().downcast_ref::<FilterExec>().is_some() || node.plan.as_any().downcast_ref::<SortExec>().is_some() {
Review Comment:
Needs fixing of some failing tests, cleanup of the plethora of helper methods I added and a lot of docs but here's the idea: https://github.com/apache/datafusion/pull/15801. The points are:
- No downcast matching / hardcoding of implementations
- Only recurses once / no retrying
- Does no cloning / copying for branches that have no changes
- Doesn't insert new operators
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2150649026
##########
datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs:
##########
@@ -118,7 +131,7 @@ fn test_filter_collapse() {
let plan = Arc::new(FilterExec::try_new(predicate2, filter1).unwrap());
insta::assert_snapshot!(
- OptimizationTest::new(plan, FilterPushdown{}, true),
+ OptimizationTest::new(plan, FilterPushdown::new_pre_optimization(), true),
Review Comment:
Maybe call it `FilterPushdown::new` and `FilterPushdown::new_dynamic` ?
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2977884089
I'd be curious to see perf impact after we merge https://github.com/apache/datafusion/pull/16424 as well
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2145258417
##########
datafusion/physical-plan/src/topk/mod.rs:
##########
@@ -207,6 +221,7 @@ impl TopK {
// Idea: filter out rows >= self.heap.max() early (before passing to `RowConverter`)
// this avoids some work and also might be better vectorizable.
let mut batch_entry = self.heap.register_batch(batch.clone());
+ let mut updated = false;
Review Comment:
This doesn't do the filtering here, right (https://github.com/apache/datafusion/pull/15697 ). Is there any benefit doing that as well?
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2970721901
π€: Benchmark completed
<details><summary>Details</summary>
<p>
```
Comparing HEAD and topk-dynamic-filters
--------------------
Benchmark clickbench_extended.json
--------------------
ββββββββββββββββ³ββββββββββββββ³βββββββββββββββββββββββ³ββββββββββββ
β Query β HEAD β topk-dynamic-filters β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β QQuery 0 β 1982.56 ms β 1955.26 ms β no change β
β QQuery 1 β 711.82 ms β 678.72 ms β no change β
β QQuery 2 β 1513.91 ms β 1450.23 ms β no change β
β QQuery 3 β 708.20 ms β 693.14 ms β no change β
β QQuery 4 β 1448.54 ms β 1465.17 ms β no change β
β QQuery 5 β 16141.33 ms β 16207.13 ms β no change β
β QQuery 6 β 2038.97 ms β 2054.36 ms β no change β
β QQuery 7 β 2148.16 ms β 2078.78 ms β no change β
β QQuery 8 β 875.11 ms β 850.36 ms β no change β
ββββββββββββββββ΄ββββββββββββββ΄βββββββββββββββββββββββ΄ββββββββββββ
βββββββββββββββββββββββββββββββββββββββ³βββββββββββββ
β Benchmark Summary β β
β‘βββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 27568.61ms β
β Total Time (topk-dynamic-filters) β 27433.15ms β
β Average Time (HEAD) β 3063.18ms β
β Average Time (topk-dynamic-filters) β 3048.13ms β
β Queries Faster β 0 β
β Queries Slower β 0 β
β Queries with No Change β 9 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββββββ΄βββββββββββββ
--------------------
Benchmark clickbench_partitioned.json
--------------------
ββββββββββββββββ³ββββββββββββββ³βββββββββββββββββββββββ³ββββββββββββββββ
β Query β HEAD β topk-dynamic-filters β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β QQuery 0 β 14.73 ms β 16.09 ms β 1.09x slower β
β QQuery 1 β 32.46 ms β 34.15 ms β 1.05x slower β
β QQuery 2 β 82.16 ms β 82.93 ms β no change β
β QQuery 3 β 99.44 ms β 93.57 ms β +1.06x faster β
β QQuery 4 β 586.83 ms β 640.21 ms β 1.09x slower β
β QQuery 5 β 869.83 ms β 895.81 ms β no change β
β QQuery 6 β 24.73 ms β 23.35 ms β +1.06x faster β
β QQuery 7 β 38.17 ms β 39.22 ms β no change β
β QQuery 8 β 927.63 ms β 926.34 ms β no change β
β QQuery 9 β 1189.27 ms β 1241.04 ms β no change β
β QQuery 10 β 268.91 ms β 265.63 ms β no change β
β QQuery 11 β 296.18 ms β 297.87 ms β no change β
β QQuery 12 β 919.60 ms β 916.36 ms β no change β
β QQuery 13 β 1367.05 ms β 1375.43 ms β no change β
β QQuery 14 β 854.12 ms β 866.26 ms β no change β
β QQuery 15 β 827.71 ms β 829.16 ms β no change β
β QQuery 16 β 1760.76 ms β 1734.60 ms β no change β
β QQuery 17 β 1640.40 ms β 1595.59 ms β no change β
β QQuery 18 β 3152.52 ms β 3106.21 ms β no change β
β QQuery 19 β 85.80 ms β 89.23 ms β no change β
β QQuery 20 β 1139.03 ms β 1165.68 ms β no change β
β QQuery 21 β 1327.33 ms β 1383.27 ms β no change β
β QQuery 22 β 2174.39 ms β 2303.49 ms β 1.06x slower β
β QQuery 23 β 7881.25 ms β 7861.79 ms β no change β
β QQuery 24 β 469.80 ms β 482.91 ms β no change β
β QQuery 25 β 380.03 ms β 279.23 ms β +1.36x faster β
β QQuery 26 β 533.18 ms β 546.21 ms β no change β
β QQuery 27 β 1595.23 ms β 1646.22 ms β no change β
β QQuery 28 β 12629.48 ms β 12542.27 ms β no change β
β QQuery 29 β 537.79 ms β 542.54 ms β no change β
β QQuery 30 β 791.58 ms β 814.30 ms β no change β
β QQuery 31 β 856.79 ms β 864.66 ms β no change β
β QQuery 32 β 2646.01 ms β 2693.32 ms β no change β
β QQuery 33 β 3339.13 ms β 3374.26 ms β no change β
β QQuery 34 β 3352.60 ms β 3397.57 ms β no change β
β QQuery 35 β 1269.19 ms β 1270.10 ms β no change β
β QQuery 36 β 126.35 ms β 128.61 ms β no change β
β QQuery 37 β 56.05 ms β 57.52 ms β no change β
β QQuery 38 β 125.95 ms β 128.50 ms β no change β
β QQuery 39 β 199.01 ms β 205.63 ms β no change β
β QQuery 40 β 47.73 ms β 50.43 ms β 1.06x slower β
β QQuery 41 β 44.76 ms β 47.79 ms β 1.07x slower β
β QQuery 42 β 40.09 ms β 38.73 ms β no change β
ββββββββββββββββ΄ββββββββββββββ΄βββββββββββββββββββββββ΄ββββββββββββββββ
βββββββββββββββββββββββββββββββββββββββ³βββββββββββββ
β Benchmark Summary β β
β‘βββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 56601.04ms β
β Total Time (topk-dynamic-filters) β 56894.09ms β
β Average Time (HEAD) β 1316.30ms β
β Average Time (topk-dynamic-filters) β 1323.12ms β
β Queries Faster β 3 β
β Queries Slower β 6 β
β Queries with No Change β 34 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββββββ΄βββββββββββββ
--------------------
Benchmark tpch_mem_sf1.json
--------------------
ββββββββββββββββ³ββββββββββββ³βββββββββββββββββββββββ³ββββββββββββββββ
β Query β HEAD β topk-dynamic-filters β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β QQuery 1 β 118.07 ms β 122.57 ms β no change β
β QQuery 2 β 22.99 ms β 21.84 ms β no change β
β QQuery 3 β 34.65 ms β 34.20 ms β no change β
β QQuery 4 β 19.95 ms β 20.17 ms β no change β
β QQuery 5 β 55.13 ms β 53.98 ms β no change β
β QQuery 6 β 12.41 ms β 12.12 ms β no change β
β QQuery 7 β 104.45 ms β 97.63 ms β +1.07x faster β
β QQuery 8 β 27.20 ms β 26.36 ms β no change β
β QQuery 9 β 62.15 ms β 60.70 ms β no change β
β QQuery 10 β 51.21 ms β 51.67 ms β no change β
β QQuery 11 β 11.88 ms β 11.66 ms β no change β
β QQuery 12 β 41.25 ms β 43.00 ms β no change β
β QQuery 13 β 28.02 ms β 27.51 ms β no change β
β QQuery 14 β 10.41 ms β 10.17 ms β no change β
β QQuery 15 β 23.97 ms β 23.13 ms β no change β
β QQuery 16 β 21.79 ms β 22.31 ms β no change β
β QQuery 17 β 100.36 ms β 99.64 ms β no change β
β QQuery 18 β 207.06 ms β 222.22 ms β 1.07x slower β
β QQuery 19 β 26.04 ms β 26.58 ms β no change β
β QQuery 20 β 35.86 ms β 35.89 ms β no change β
β QQuery 21 β 164.05 ms β 161.99 ms β no change β
β QQuery 22 β 17.29 ms β 17.02 ms β no change β
ββββββββββββββββ΄ββββββββββββ΄βββββββββββββββββββββββ΄ββββββββββββββββ
βββββββββββββββββββββββββββββββββββββββ³ββββββββββββ
β Benchmark Summary β β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 1196.16ms β
β Total Time (topk-dynamic-filters) β 1202.36ms β
β Average Time (HEAD) β 54.37ms β
β Average Time (topk-dynamic-filters) β 54.65ms β
β Queries Faster β 1 β
β Queries Slower β 1 β
β Queries with No Change β 20 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββββββ΄ββββββββββββ
```
</p>
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2145547651
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -843,6 +846,8 @@ pub struct SortExec {
common_sort_prefix: Vec<PhysicalSortExpr>,
/// Cache holding plan properties like equivalences, output partitioning etc.
cache: PlanProperties,
+ /// Filter matching the state of the sort for dynamic filter pushdown
+ filter: Option<Arc<DynamicFilterPhysicalExpr>>,
Review Comment:
It will be created regardless atm. Possibly could be optimized, e.g. by keeping track of refs but not too sure.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2145548000
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -843,6 +846,8 @@ pub struct SortExec {
common_sort_prefix: Vec<PhysicalSortExpr>,
/// Cache holding plan properties like equivalences, output partitioning etc.
cache: PlanProperties,
+ /// Filter matching the state of the sort for dynamic filter pushdown
+ filter: Option<Arc<DynamicFilterPhysicalExpr>>,
Review Comment:
It _won't_ be created it there's no `limit` applied is the idea.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2150951172
##########
datafusion/physical-optimizer/src/optimizer.rs:
##########
@@ -131,6 +131,8 @@ impl PhysicalOptimizer {
// replacing operators with fetching variants, or adding limits
// past operators that support limit pushdown.
Arc::new(LimitPushdown::new()),
+ // This FilterPushdown handles dynamic filters that may have references to the source ExecutionPlan
+ Arc::new(FilterPushdown::new_post_optimization()),
Review Comment:
I will move it lower and add a comment, with a reference to the enum with larger docs.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2981552270
> Perhaps we can compare against the current filter and only update the expression if it is greater / more selective?
Yeah I think that would be good.
For context (I had to remember for a sec): the filter itself is shared across partitions.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2974704013
> I think two different phases sounds good.
>
> I don't really like complicating FilterPushdown with post / pre "phases". Why not creating the `PushDownDynamicFilters` one you suggested earlier?
My reasoning for using a parameters was that there's a lot of shared implementation both on the optimizer side and ExectionPlan side. E.g. the transparent nodes like CoalesceBatchesExec don't really care which phase it is, but FilterExec needs to ignore the dynamic phase and SortExec needs to ignore the static filter pushdown phase.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2977763190
> I think it's great! I added one suggestion to minimize the changes to the explain output and maybe phrase the "pre" and "post" more as "normal" filter pushdown and "dynamic" filter pushdown? I think that might explain a bit more of what might be happening.
>
> Maybe @alamb can verify the (increased) performance improvements once more?
Thanks! I have no strong opinions about the naming so I'll wait for Andrew to review to see if he has any thoughts otherwise I'll go with your suggestion π
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2052311257
##########
datafusion/physical-optimizer/src/push_down_filter.rs:
##########
@@ -382,7 +383,7 @@ impl PhysicalOptimizerRule for PushdownFilter {
context
.transform_up(|node| {
- if node.plan.as_any().downcast_ref::<FilterExec>().is_some() {
+ if node.plan.as_any().downcast_ref::<FilterExec>().is_some() || node.plan.as_any().downcast_ref::<SortExec>().is_some() {
Review Comment:
Yes, you're right. We can run this pushdown logic on every operator actually, but then it will work in worst-time complexity always. I've shared the solution of removing revisit parameter, and let me open an issue for that. I strongly believe it will be taken and implemented in short time by some people.
To remove these downcasts, I think we can either introduce a new method to the API just returning a boolean saying that "this operator might introduce a filter or not", or try to understand that by the existing API's, maybe with some refactor. Do you have an idea for the latter?
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2869830929
The PR looks really nice BTW, and every line is clearly understandable. Of course there are some possible optimizations (as you've also noticed some of them like the todo during batch insertion), but this initial version is more than enough IMO.
The rule issue is not very trivial because we cannot just track and eliminate some hardcoded patterns, since we also need to be aware of upper parts of the plan, and new patterns may emerge as new tests are added, it's not limited to some specific cases.
as I mentioned, I might have found another solution, and if it works I'll ping you and we can merge this
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2857381137
Probably, we cannot solve this planning issue just by changing the order of the rules. My instinct is that we need a smarter FilterPushdown logic, but I need to investigate it further. I don't think it will require a lot of effort, so I can send it today, along with the other review suggestions.
Thanks again @adriangb, we are very close to the end, and I can't wait to start implementing the join information pass
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2981154277
π€ `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running
Linux aal-dev 6.11.0-1015-gcp #15~24.04.1-Ubuntu SMP Thu Apr 24 20:41:05 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing topk-dynamic-filters (8e88bd9913f0e8b3d04c1a61e8e928d09f271dba) to dd936cb1b25cb685e0e146f297950eb00048c64c [diff](https://github.com/apache/datafusion/compare/dd936cb1b25cb685e0e146f297950eb00048c64c..8e88bd9913f0e8b3d04c1a61e8e928d09f271dba)
Benchmarks: topk_tpch
Results will be posted here when complete
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2957661232
I actually seem to have been able to get the tests passing, it wasn't as complex of a bug as I thought. Still would appreciate π since this is complex bits of logic that are new to me.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2970575427
π€ `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running
Linux aal-dev 6.11.0-1015-gcp #15~24.04.1-Ubuntu SMP Thu Apr 24 20:41:05 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing topk-dynamic-filters (61a2dfdc71508bd0b09abaefdc882d33a9e0d91e) to 1d73c59b998a0b76ee2ebaacd22df682b31a540c [diff](https://github.com/apache/datafusion/compare/1d73c59b998a0b76ee2ebaacd22df682b31a540c..61a2dfdc71508bd0b09abaefdc882d33a9e0d91e)
Benchmarks: tpch_mem clickbench_partitioned clickbench_extended
Results will be posted here when complete
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2145790872
##########
datafusion/core/tests/fuzz_cases/topk_filter_pushdown.rs:
##########
@@ -0,0 +1,387 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::HashMap;
+use std::sync::{Arc, LazyLock};
+
+use arrow::array::{Int32Array, StringArray, StringDictionaryBuilder};
+use arrow::datatypes::Int32Type;
+use arrow::record_batch::RecordBatch;
+use arrow::util::pretty::pretty_format_batches;
+use arrow_schema::{DataType, Field, Schema};
+use datafusion::datasource::listing::{ListingOptions, ListingTable, ListingTableConfig};
+use datafusion::prelude::{SessionConfig, SessionContext};
+use datafusion_datasource::ListingTableUrl;
+use datafusion_datasource_parquet::ParquetFormat;
+use datafusion_execution::object_store::ObjectStoreUrl;
+use itertools::Itertools;
+use object_store::memory::InMemory;
+use object_store::path::Path;
+use object_store::{ObjectStore, PutPayload};
+use parquet::arrow::ArrowWriter;
+use rand::rngs::StdRng;
+use rand::{Rng, SeedableRng};
+use tokio::sync::Mutex;
+use tokio::task::JoinSet;
+
+#[derive(Clone)]
+struct TestDataSet {
+ store: Arc<dyn ObjectStore>,
+ schema: Arc<Schema>,
+}
+
+/// List of in memory parquet files with UTF8 data
+// Use a mutex rather than LazyLock to allow for async initialization
+static TESTFILES: LazyLock<Mutex<Vec<TestDataSet>>> =
+ LazyLock::new(|| Mutex::new(vec![]));
+
+async fn test_files() -> Vec<TestDataSet> {
+ let files_mutex = &TESTFILES;
+ let mut files = files_mutex.lock().await;
+ if !files.is_empty() {
+ return (*files).clone();
+ }
+
+ let mut rng = StdRng::seed_from_u64(0);
+
+ for nulls_in_ids in [false, true] {
+ for nulls_in_names in [false, true] {
+ for nulls_in_departments in [false, true] {
+ let store = Arc::new(InMemory::new());
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, nulls_in_ids),
+ Field::new("name", DataType::Utf8, nulls_in_names),
+ Field::new(
+ "department",
+ DataType::Dictionary(
+ Box::new(DataType::Int32),
+ Box::new(DataType::Utf8),
+ ),
+ nulls_in_departments,
+ ),
+ ]));
+
+ let name_choices = if nulls_in_names {
+ [Some("Alice"), Some("Bob"), None, Some("David"), None]
+ } else {
+ [
+ Some("Alice"),
+ Some("Bob"),
+ Some("Charlie"),
+ Some("David"),
+ Some("Eve"),
+ ]
+ };
+
+ let department_choices = if nulls_in_departments {
+ [
+ Some("Theater"),
+ Some("Engineering"),
+ None,
+ Some("Arts"),
+ None,
+ ]
+ } else {
+ [
+ Some("Theater"),
+ Some("Engineering"),
+ Some("Healthcare"),
+ Some("Arts"),
+ Some("Music"),
+ ]
+ };
+
+ // Generate 5 files, some with overlapping or repeated ids some without
+ for i in 0..5 {
+ let num_batches = rng.random_range(1..3);
+ let mut batches = Vec::with_capacity(num_batches);
+ for _ in 0..num_batches {
+ let num_rows = 25;
+ let ids = Int32Array::from_iter((0..num_rows).map(|file| {
+ if nulls_in_ids {
+ if rng.random_bool(1.0 / 10.0) {
+ None
+ } else {
+ Some(rng.random_range(file..file + 5))
+ }
+ } else {
+ Some(rng.random_range(file..file + 5))
+ }
+ }));
+ let names = StringArray::from_iter((0..num_rows).map(|_| {
+ // randomly select a name
+ let idx = rng.random_range(0..name_choices.len());
+ name_choices[idx].map(|s| s.to_string())
+ }));
+ let mut departments = StringDictionaryBuilder::<Int32Type>::new();
+ for _ in 0..num_rows {
+ // randomly select a department
+ let idx = rng.random_range(0..department_choices.len());
+ departments.append_option(department_choices[idx].as_ref());
+ }
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(ids),
+ Arc::new(names),
+ Arc::new(departments.finish()),
+ ],
+ )
+ .unwrap();
+ batches.push(batch);
+ }
+ let mut buf = vec![];
+ {
+ let mut writer =
+ ArrowWriter::try_new(&mut buf, schema.clone(), None).unwrap();
+ for batch in batches {
+ writer.write(&batch).unwrap();
+ writer.flush().unwrap();
+ }
+ writer.flush().unwrap();
+ writer.finish().unwrap();
+ }
+ let payload = PutPayload::from(buf);
+ let path = Path::from(format!("file_{i}.parquet"));
+ store.put(&path, payload).await.unwrap();
+ }
+ files.push(TestDataSet { store, schema });
+ }
+ }
+ }
+ (*files).clone()
+}
+
+struct RunResult {
+ results: Vec<RecordBatch>,
+ explain_plan: String,
+}
+
+async fn run_query_with_config(
+ query: &str,
+ config: SessionConfig,
+ dataset: TestDataSet,
+) -> RunResult {
+ let store = dataset.store;
+ let schema = dataset.schema;
+ let ctx = SessionContext::new_with_config(config);
+ let url = ObjectStoreUrl::parse("memory://").unwrap();
+ ctx.register_object_store(url.as_ref(), store.clone());
+
+ let format = Arc::new(
+ ParquetFormat::default()
+ .with_options(ctx.state().table_options().parquet.clone()),
+ );
+ let options = ListingOptions::new(format);
+ let table_path = ListingTableUrl::parse("memory:///").unwrap();
+ let config = ListingTableConfig::new(table_path)
+ .with_listing_options(options)
+ .with_schema(schema);
+ let table = Arc::new(ListingTable::try_new(config).unwrap());
+
+ ctx.register_table("test_table", table).unwrap();
+
+ let results = ctx.sql(query).await.unwrap().collect().await.unwrap();
+ let explain_batches = ctx
+ .sql(&format!("EXPLAIN ANALYZE {query}"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let explain_plan = pretty_format_batches(&explain_batches).unwrap().to_string();
+ RunResult {
+ results,
+ explain_plan,
+ }
+}
+
+#[derive(Debug)]
+struct RunQueryResult {
+ query: String,
+ result: Vec<RecordBatch>,
+ expected: Vec<RecordBatch>,
+}
+
+impl RunQueryResult {
+ fn expected_formated(&self) -> String {
+ format!("{}", pretty_format_batches(&self.expected).unwrap())
+ }
+
+ fn result_formated(&self) -> String {
+ format!("{}", pretty_format_batches(&self.result).unwrap())
+ }
+
+ fn is_ok(&self) -> bool {
+ self.expected_formated() == self.result_formated()
+ }
+}
+
+/// Iterate over each line in the plan and check that one of them has `DataSourceExec` and `DynamicFilterPhysicalExpr` in the same line.
+fn has_dynamic_filter_expr_pushdown(plan: &str) -> bool {
+ for line in plan.lines() {
+ if line.contains("DataSourceExec") && line.contains("DynamicFilterPhysicalExpr") {
+ return true;
+ }
+ }
+ false
+}
+
+async fn run_query(
+ query: String,
+ cfg: SessionConfig,
+ dataset: TestDataSet,
+) -> RunQueryResult {
+ let cfg_with_dynamic_filters = cfg
+ .clone()
+ .set_bool("datafusion.optimizer.enable_dynamic_filter_pushdown", true);
+ let cfg_without_dynamic_filters = cfg
+ .clone()
+ .set_bool("datafusion.optimizer.enable_dynamic_filter_pushdown", false);
+
+ let expected_result =
+ run_query_with_config(&query, cfg_without_dynamic_filters, dataset.clone()).await;
+ let result =
+ run_query_with_config(&query, cfg_with_dynamic_filters, dataset.clone()).await;
+ // Check that dynamic filters were actually pushed down
+ if !has_dynamic_filter_expr_pushdown(&result.explain_plan) {
Review Comment:
nice
##########
datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs:
##########
@@ -346,6 +359,137 @@ fn test_node_handles_child_pushdown_result() {
);
}
+#[tokio::test]
+async fn test_topk_dynamic_filter_pushdown() {
+ // This test is a bit of a hack, but it shows that we can push down dynamic filters
+ // into the DataSourceExec. The test is not perfect because we don't have a real
+ // implementation of the dynamic filter yet, so we just use a static filter.
Review Comment:
this comment seems maybe out of date as we now have a real implementation of the filter
##########
datafusion/physical-optimizer/src/enforce_sorting/sort_pushdown.rs:
##########
@@ -70,6 +71,8 @@ pub fn assign_initial_requirements(sort_push_down: &mut SortPushDown) {
// If the parent has a fetch value, assign it to the children
// Or use the fetch value of the child.
fetch: child.plan.fetch(),
+ // If the parent has a filter, assign it to the children
Review Comment:
I am surprised to see that filter pushdown is happening in "EnforceSorting" -- I would have expected it to happen in the `FilterPushdown` physical optimizer rule (which was added I thought for precisely this usecase)
##########
datafusion/physical-plan/src/topk/mod.rs:
##########
@@ -232,6 +248,105 @@ impl TopK {
// which means the top K won't change and the computation can be finished early.
self.attempt_early_completion(&batch)?;
+ if updated {
+ // update the filter representation of our TopK heap
+ self.update_filter()?;
+ }
+
+ Ok(())
+ }
+
+ /// Update the filter representation of our TopK heap
Review Comment:
FWIW I recommend adding an example to this doc string showing what the filter expression would be for a multi-column sort order
##########
datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs:
##########
@@ -346,6 +359,137 @@ fn test_node_handles_child_pushdown_result() {
);
}
+#[tokio::test]
+async fn test_topk_dynamic_filter_pushdown() {
+ // This test is a bit of a hack, but it shows that we can push down dynamic filters
+ // into the DataSourceExec. The test is not perfect because we don't have a real
+ // implementation of the dynamic filter yet, so we just use a static filter.
+ let batches = vec![
+ record_batch!(
+ ("a", Utf8, ["aa", "ab"]),
+ ("b", Utf8, ["bd", "bc"]),
+ ("c", Float64, [1.0, 2.0])
+ )
+ .unwrap(),
+ record_batch!(
+ ("a", Utf8, ["ac", "ad"]),
+ ("b", Utf8, ["bb", "ba"]),
+ ("c", Float64, [2.0, 1.0])
+ )
+ .unwrap(),
+ ];
+ let scan = TestScanBuilder::new(schema())
+ .with_support(true)
+ .with_batches(batches)
+ .build();
+ let plan = Arc::new(
+ SortExec::new(
+ LexOrdering::new(vec![PhysicalSortExpr::new(
+ col("b", &schema()).unwrap(),
+ SortOptions::new(true, false), // descending, nulls_first
+ )])
+ .unwrap(),
+ Arc::clone(&scan),
+ )
+ .with_fetch(Some(1)),
+ ) as Arc<dyn ExecutionPlan>;
+
+ // expect the predicate to be pushed down into the DataSource
+ insta::assert_snapshot!(
+ OptimizationTest::new(Arc::clone(&plan), FilterPushdown{}, true),
+ @r"
+ OptimizationTest:
+ input:
+ - SortExec: TopK(fetch=1), expr=[b@1 DESC NULLS LAST], preserve_partitioning=[false]
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
Review Comment:
I don't know if we care but parquet is misspelled here: `test.paqruet`
##########
datafusion/physical-plan/src/topk/mod.rs:
##########
@@ -570,6 +680,47 @@ impl TopKHeap {
+ self.store.size()
+ self.owned_bytes
}
+
+ fn get_threshold_values(
+ &self,
+ sort_exprs: &[PhysicalSortExpr],
+ ) -> Result<Option<Vec<ScalarValue>>> {
+ // If the heap doesn't have k elements yet, we can't create thresholds
+ let max_row = match self.max() {
+ Some(row) => row,
+ None => return Ok(None),
+ };
+
+ // Get the batch that contains the max row
+ let batch_entry = match self.store.get(max_row.batch_id) {
+ Some(entry) => entry,
+ None => return internal_err!("Invalid batch ID in TopKRow"),
+ };
+
+ // Extract threshold values for each sort expression
+ let mut scalar_values = Vec::with_capacity(sort_exprs.len());
+ for sort_expr in sort_exprs {
Review Comment:
100ms is a LONG time -- I wouldn't expect this to ever take that long. I am not sure exactly what we would parallelize π€
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -843,6 +846,8 @@ pub struct SortExec {
common_sort_prefix: Vec<PhysicalSortExpr>,
/// Cache holding plan properties like equivalences, output partitioning etc.
cache: PlanProperties,
+ /// Filter matching the state of the sort for dynamic filter pushdown
+ filter: Option<Arc<DynamicFilterPhysicalExpr>>,
Review Comment:
it would also be nice to add a note about when this is field is set or not
##########
datafusion/physical-plan/src/topk/mod.rs:
##########
@@ -232,6 +248,105 @@ impl TopK {
// which means the top K won't change and the computation can be finished early.
self.attempt_early_completion(&batch)?;
+ if updated {
+ // update the filter representation of our TopK heap
+ self.update_filter()?;
+ }
+
+ Ok(())
+ }
+
+ /// Update the filter representation of our TopK heap
+ fn update_filter(&mut self) -> Result<()> {
+ let Some(filter) = &self.filter else {
+ return Ok(());
+ };
+ if let Some(thresholds) = self.heap.get_threshold_values(&self.expr)? {
Review Comment:
(minor) You could probably remove another level of indent here with the same pattern.
```suggestion
let Some(thresholds) = self.heap.get_threshold_values(&self.expr)? else {
return Ok(());
}
```
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -246,38 +246,3 @@ physical_plan
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)]
-
-# The order of filters should not matter
Review Comment:
why is this removed?
##########
datafusion/physical-optimizer/src/enforce_sorting/sort_pushdown.rs:
##########
@@ -114,6 +118,18 @@ fn pushdown_sorts_helper(
sort_push_down.data.fetch = fetch;
sort_push_down.data.ordering_requirement =
Some(OrderingRequirements::from(sort_ordering));
+ let filter = plan
+ .as_any()
+ .downcast_ref::<SortExec>()
+ .and_then(|s| s.filter().clone());
Review Comment:
I think all we need to do is add a method to the `ExecutionPlan` like `fn dynamic_filters` and it will be general and well documented
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2145917211
##########
datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs:
##########
@@ -346,6 +359,137 @@ fn test_node_handles_child_pushdown_result() {
);
}
+#[tokio::test]
+async fn test_topk_dynamic_filter_pushdown() {
+ // This test is a bit of a hack, but it shows that we can push down dynamic filters
+ // into the DataSourceExec. The test is not perfect because we don't have a real
+ // implementation of the dynamic filter yet, so we just use a static filter.
+ let batches = vec![
+ record_batch!(
+ ("a", Utf8, ["aa", "ab"]),
+ ("b", Utf8, ["bd", "bc"]),
+ ("c", Float64, [1.0, 2.0])
+ )
+ .unwrap(),
+ record_batch!(
+ ("a", Utf8, ["ac", "ad"]),
+ ("b", Utf8, ["bb", "ba"]),
+ ("c", Float64, [2.0, 1.0])
+ )
+ .unwrap(),
+ ];
+ let scan = TestScanBuilder::new(schema())
+ .with_support(true)
+ .with_batches(batches)
+ .build();
+ let plan = Arc::new(
+ SortExec::new(
+ LexOrdering::new(vec![PhysicalSortExpr::new(
+ col("b", &schema()).unwrap(),
+ SortOptions::new(true, false), // descending, nulls_first
+ )])
+ .unwrap(),
+ Arc::clone(&scan),
+ )
+ .with_fetch(Some(1)),
+ ) as Arc<dyn ExecutionPlan>;
+
+ // expect the predicate to be pushed down into the DataSource
+ insta::assert_snapshot!(
+ OptimizationTest::new(Arc::clone(&plan), FilterPushdown{}, true),
+ @r"
+ OptimizationTest:
+ input:
+ - SortExec: TopK(fetch=1), expr=[b@1 DESC NULLS LAST], preserve_partitioning=[false]
+ - DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
Review Comment:
https://github.com/apache/datafusion/pull/16403
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2145521489
##########
datafusion/physical-plan/src/topk/mod.rs:
##########
@@ -207,6 +221,7 @@ impl TopK {
// Idea: filter out rows >= self.heap.max() early (before passing to `RowConverter`)
// this avoids some work and also might be better vectorizable.
let mut batch_entry = self.heap.register_batch(batch.clone());
+ let mut updated = false;
Review Comment:
In a followup would be fine!
From my earlier results here https://github.com/apache/datafusion/pull/15697 I see some queries that have a larger speedup for some of the benchmarks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2145572928
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -843,6 +846,8 @@ pub struct SortExec {
common_sort_prefix: Vec<PhysicalSortExpr>,
/// Cache holding plan properties like equivalences, output partitioning etc.
cache: PlanProperties,
+ /// Filter matching the state of the sort for dynamic filter pushdown
+ filter: Option<Arc<DynamicFilterPhysicalExpr>>,
Review Comment:
That might make sense for now π€
I think once we do https://github.com/apache/datafusion/pull/15697 probably it's not needed to disable computing 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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2971539472
> It could (and does currently) handle push down of dynamic filters, the issue is that it cannot be run after EnforceSorting because EnforceSorting and EnforceDistribution need to be run after e.g. a FilterExec is removed from the plan.
I see. I will need to think if there is any better way, but I think someone would just have to try to make it work / detangle EnforceSorting more
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2865349477
Update: I tried to re-order the rules like:
1) EnforceSorting
2) FilterPushdown
3) EnforceDistribution
4) CombinePartialFinalAggregate
...
5) CoalesceBatches
The issue here is fixed, but some others came up (related with EnforceSorting and EnforceDistribution swap). I will further debug that issue. EnforceSorting and EnforceDistribution are initially designed to operate orthogonally, and there are many tests for that. But it seems somehow we broke it.
Unless I can figure out easily to recover that orthogonality feature, I'll manually implement FilterPushdown rule to eliminate
```
CoalesceBatches
--Repartition
```
patterns
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2857400638
Curious to see what you come up with!
A couple of ideas:
1. Split filter pushdown into two phases. One that can modify the plan structure but not push down filters with references to nodes (so just pushing static FilterExec filters down) and another that pushes down dynamic filters but cannot modify the shape of the plan.
2. Have an extra pass that specifically deals with removing redundant CoalesceBatchesExec & RepartitionExec.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2083516280
##########
datafusion/sqllogictest/test_files/prepare.slt:
##########
@@ -264,16 +264,19 @@ WHERE run_id = 'foo'
ORDER BY random()
LIMIT $1
-query I
+query error
EXECUTE get_N_rand_ints_from_last_run(1);
----
-1
+DataFusion error: Internal error: PhysicalOptimizer rule 'LimitPushdown' failed. Schema mismatch. Expected original schema: Schema { fields: [Field { name: "id", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }], metadata: {} }, got new schema: Schema { fields: [Field { name: "id", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "run_id", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }], metadata: {} }.
Review Comment:
I'm dealing with these 2 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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2816116031
Pausing this until https://github.com/apache/datafusion/pull/15769 is 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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2865382987
> Maybe worth confirming that it is broken independently of this PR (on main)?
Yes, it was broken before.
One can move EnforceSorting up on EnforceDistribution and see the 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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2981852178
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2978979555
> π€: Benchmark completed
>
> Details
> π€: Benchmark completed
>
> Details
could you maybe confirm the topk benchmark results @alamb ? `topk_tp ch`?
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2138485861
##########
datafusion/core/tests/fuzz_cases/topk_filter_pushdown.rs:
##########
@@ -0,0 +1,354 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::HashMap;
+use std::sync::{Arc, LazyLock};
+
+use arrow::array::{Int32Array, StringArray, StringDictionaryBuilder};
+use arrow::datatypes::Int32Type;
+use arrow::record_batch::RecordBatch;
+use arrow::util::pretty::pretty_format_batches;
+use arrow_schema::{DataType, Field, Schema};
+use datafusion::datasource::listing::{ListingOptions, ListingTable, ListingTableConfig};
+use datafusion::prelude::{SessionConfig, SessionContext};
+use datafusion_datasource::ListingTableUrl;
+use datafusion_datasource_parquet::ParquetFormat;
+use datafusion_execution::object_store::ObjectStoreUrl;
+use itertools::Itertools;
+use object_store::memory::InMemory;
+use object_store::path::Path;
+use object_store::{ObjectStore, PutPayload};
+use parquet::arrow::ArrowWriter;
+use rand::rngs::StdRng;
+use rand::{Rng, SeedableRng};
+use tokio::sync::Mutex;
+use tokio::task::JoinSet;
+
+#[derive(Clone)]
+struct TestDataSet {
+ store: Arc<dyn ObjectStore>,
+ schema: Arc<Schema>,
+}
+
+/// List of in memory parquet files with UTF8 data
+// Use a mutex rather than LazyLock to allow for async initialization
+static TESTFILES: LazyLock<Mutex<Vec<TestDataSet>>> =
+ LazyLock::new(|| Mutex::new(vec![]));
+
+async fn test_files() -> Vec<TestDataSet> {
+ let files_mutex = &TESTFILES;
+ let mut files = files_mutex.lock().await;
+ if !files.is_empty() {
+ return (*files).clone();
+ }
+
+ let mut rng = StdRng::seed_from_u64(0);
+
+ for nulls_in_ids in [false, true] {
+ for nulls_in_names in [false, true] {
+ for nulls_in_departments in [false, true] {
+ let store = Arc::new(InMemory::new());
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, nulls_in_ids),
+ Field::new("name", DataType::Utf8, nulls_in_names),
+ Field::new(
+ "department",
+ DataType::Dictionary(
+ Box::new(DataType::Int32),
+ Box::new(DataType::Utf8),
+ ),
+ nulls_in_departments,
+ ),
+ ]));
+
+ let name_choices = if nulls_in_names {
+ [Some("Alice"), Some("Bob"), None, Some("David"), None]
+ } else {
+ [
+ Some("Alice"),
+ Some("Bob"),
+ Some("Charlie"),
+ Some("David"),
+ Some("Eve"),
+ ]
+ };
+
+ let department_choices = if nulls_in_departments {
+ [
+ Some("Theater"),
+ Some("Engineering"),
+ None,
+ Some("Arts"),
+ None,
+ ]
+ } else {
+ [
+ Some("Theater"),
+ Some("Engineering"),
+ Some("Healthcare"),
+ Some("Arts"),
+ Some("Music"),
+ ]
+ };
+
+ // Generate 5 files, some with overlapping or repeated ids some without
+ for i in 0..5 {
+ let num_batches = rng.gen_range(1..3);
+ let mut batches = Vec::with_capacity(num_batches);
+ for _ in 0..num_batches {
+ let num_rows = 25;
+ let ids = Int32Array::from_iter((0..num_rows).map(|file| {
+ if nulls_in_ids {
+ if rng.gen_bool(1.0 / 10.0) {
+ None
+ } else {
+ Some(rng.gen_range(file..file + 5))
+ }
+ } else {
+ Some(rng.gen_range(file..file + 5))
+ }
+ }));
+ let names = StringArray::from_iter((0..num_rows).map(|_| {
+ // randomly select a name
+ let idx = rng.gen_range(0..name_choices.len());
+ name_choices[idx].map(|s| s.to_string())
+ }));
+ let mut departments = StringDictionaryBuilder::<Int32Type>::new();
+ for _ in 0..num_rows {
+ // randomly select a department
+ let idx = rng.gen_range(0..department_choices.len());
+ departments.append_option(department_choices[idx].as_ref());
+ }
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(ids),
+ Arc::new(names),
+ Arc::new(departments.finish()),
+ ],
+ )
+ .unwrap();
+ batches.push(batch);
+ }
+ let mut buf = vec![];
+ {
+ let mut writer =
+ ArrowWriter::try_new(&mut buf, schema.clone(), None).unwrap();
+ for batch in batches {
+ writer.write(&batch).unwrap();
+ writer.flush().unwrap();
+ }
+ writer.flush().unwrap();
+ writer.finish().unwrap();
+ }
+ let payload = PutPayload::from(buf);
+ let path = Path::from(format!("file_{i}.parquet"));
+ store.put(&path, payload).await.unwrap();
+ }
+ files.push(TestDataSet { store, schema });
+ }
+ }
+ }
+ (*files).clone()
+}
+
+async fn run_query_with_config(
+ query: &str,
+ config: SessionConfig,
+ dataset: TestDataSet,
+) -> Vec<RecordBatch> {
+ let store = dataset.store;
+ let schema = dataset.schema;
+ let ctx = SessionContext::new_with_config(config);
+ let url = ObjectStoreUrl::parse("memory://").unwrap();
+ ctx.register_object_store(url.as_ref(), store.clone());
+
+ let format = Arc::new(
+ ParquetFormat::default()
+ .with_options(ctx.state().table_options().parquet.clone()),
+ );
+ let options = ListingOptions::new(format);
+ let table_path = ListingTableUrl::parse("memory:///").unwrap();
+ let config = ListingTableConfig::new(table_path)
+ .with_listing_options(options)
+ .with_schema(schema);
+ let table = Arc::new(ListingTable::try_new(config).unwrap());
+
+ ctx.register_table("test_table", table).unwrap();
+
+ ctx.sql(query).await.unwrap().collect().await.unwrap()
+}
+
+#[derive(Debug)]
+struct RunQueryResult {
+ query: String,
+ result: Vec<RecordBatch>,
+ expected: Vec<RecordBatch>,
+}
+
+impl RunQueryResult {
+ fn expected_formated(&self) -> String {
+ format!("{}", pretty_format_batches(&self.expected).unwrap())
+ }
+
+ fn result_formated(&self) -> String {
+ format!("{}", pretty_format_batches(&self.result).unwrap())
+ }
+
+ fn is_ok(&self) -> bool {
+ self.expected_formated() == self.result_formated()
+ }
+}
+
+async fn run_query(
+ query: String,
+ cfg: SessionConfig,
+ dataset: TestDataSet,
+) -> RunQueryResult {
+ let cfg_with_dynamic_filters = cfg
+ .clone()
+ .set_bool("datafusion.optimizer.enable_dynamic_filter_pushdown", true);
+ let cfg_without_dynamic_filters = cfg
+ .clone()
+ .set_bool("datafusion.optimizer.enable_dynamic_filter_pushdown", false);
+
+ let expected_result =
+ run_query_with_config(&query, cfg_without_dynamic_filters, dataset.clone()).await;
+ let result =
+ run_query_with_config(&query, cfg_with_dynamic_filters, dataset.clone()).await;
+
+ RunQueryResult {
+ query: query.to_string(),
+ result,
+ expected: expected_result,
+ }
+}
+
+struct TestCase {
+ query: String,
+ cfg: SessionConfig,
+ dataset: TestDataSet,
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_fuzz_topk_filter_pushdown() {
+ let order_columns = ["id", "name", "department"];
+ let order_directions = ["ASC", "DESC"];
+ let null_orders = ["NULLS FIRST", "NULLS LAST"];
+
+ let start = datafusion_common::instant::Instant::now();
+ let mut orders: HashMap<String, Vec<String>> = HashMap::new();
+ for order_column in &order_columns {
+ for order_direction in &order_directions {
+ for null_order in &null_orders {
+ // if there is a vec for this column insert the order, otherwise create a new vec
+ let ordering =
+ format!("{} {} {}", order_column, order_direction, null_order);
+ match orders.get_mut(*order_column) {
+ Some(order_vec) => {
+ order_vec.push(ordering);
+ }
+ None => {
+ orders.insert(order_column.to_string(), vec![ordering]);
+ }
+ }
+ }
+ }
+ }
+
+ let mut queries = vec![];
+
+ for limit in [1, 10] {
+ for num_order_by_columns in [1, 2, 3] {
+ for order_columns in ["id", "name", "department"]
+ .iter()
+ .combinations(num_order_by_columns)
+ {
+ for orderings in order_columns
+ .iter()
+ .map(|col| orders.get(**col).unwrap())
+ .multi_cartesian_product()
+ {
+ let query = format!(
+ "SELECT * FROM test_table ORDER BY {} LIMIT {}",
+ orderings.into_iter().join(", "),
+ limit
+ );
+ queries.push(query);
+ }
+ }
+ }
+ }
+
+ queries.sort_unstable();
+ println!(
+ "Generated {} queries in {:?}",
+ queries.len(),
+ start.elapsed()
+ );
+
+ let start = datafusion_common::instant::Instant::now();
+ let datasets = test_files().await;
+ println!("Generated test files in {:?}", start.elapsed());
+
+ let mut test_cases = vec![];
+ for enable_filter_pushdown in [true, false] {
+ for query in &queries {
+ for dataset in &datasets {
+ let mut cfg = SessionConfig::new();
+ cfg = cfg.set_bool(
+ "datafusion.optimizer.enable_dynamic_filter_pushdown",
+ enable_filter_pushdown,
+ );
+ test_cases.push(TestCase {
+ query: query.to_string(),
+ cfg,
+ dataset: dataset.clone(),
+ });
+ }
+ }
+ }
+
+ let start = datafusion_common::instant::Instant::now();
+ let mut join_set = JoinSet::new();
+ for tc in test_cases {
+ join_set.spawn(run_query(tc.query, tc.cfg, tc.dataset));
Review Comment:
db8cf0060
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2960041995
@alamb could you kick off your benchmarks run on this branch?
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2971305067
π€: Benchmark completed
<details><summary>Details</summary>
<p>
```
Comparing HEAD and topk-dynamic-filters
--------------------
Benchmark sort_tpch.json
--------------------
ββββββββββββββββ³βββββββββββββ³βββββββββββββββββββββββ³βββββββββββββββ
β Query β HEAD β topk-dynamic-filters β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Q1 β 350.14 ms β 352.17 ms β no change β
β Q2 β 266.37 ms β 302.58 ms β 1.14x slower β
β Q3 β 1146.06 ms β 1164.19 ms β no change β
β Q4 β 425.71 ms β 417.77 ms β no change β
β Q5 β 425.76 ms β 430.96 ms β no change β
β Q6 β 467.68 ms β 467.15 ms β no change β
β Q7 β 926.46 ms β 921.51 ms β no change β
β Q8 β 781.20 ms β 789.52 ms β no change β
β Q9 β 829.24 ms β 825.37 ms β no change β
β Q10 β 1223.05 ms β 1208.09 ms β no change β
β Q11 β 719.15 ms β 790.11 ms β 1.10x slower β
ββββββββββββββββ΄βββββββββββββ΄βββββββββββββββββββββββ΄βββββββββββββββ
βββββββββββββββββββββββββββββββββββββββ³ββββββββββββ
β Benchmark Summary β β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 7560.84ms β
β Total Time (topk-dynamic-filters) β 7669.42ms β
β Average Time (HEAD) β 687.35ms β
β Average Time (topk-dynamic-filters) β 697.22ms β
β Queries Faster β 0 β
β Queries Slower β 2 β
β Queries with No Change β 9 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββββββ΄ββββββββββββ
```
</p>
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2971297789
π€ `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running
Linux aal-dev 6.11.0-1015-gcp #15~24.04.1-Ubuntu SMP Thu Apr 24 20:41:05 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing topk-dynamic-filters (61a2dfdc71508bd0b09abaefdc882d33a9e0d91e) to 1d73c59b998a0b76ee2ebaacd22df682b31a540c [diff](https://github.com/apache/datafusion/compare/1d73c59b998a0b76ee2ebaacd22df682b31a540c..61a2dfdc71508bd0b09abaefdc882d33a9e0d91e)
Benchmarks: sort_tpch
Results will be posted here when complete
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2145900061
##########
datafusion/physical-optimizer/src/enforce_sorting/sort_pushdown.rs:
##########
@@ -114,6 +118,18 @@ fn pushdown_sorts_helper(
sort_push_down.data.fetch = fetch;
sort_push_down.data.ordering_requirement =
Some(OrderingRequirements::from(sort_ordering));
+ let filter = plan
+ .as_any()
+ .downcast_ref::<SortExec>()
+ .and_then(|s| s.filter().clone());
Review Comment:
We kind of already have that for the filter pushdown. The issue is that so that `SortExec` / `TopK` can call `DynamicFilterPhysicalExpr::update` they have to know it's a concrete `DynamicFilterPhysicalExpr` and not an arbitrary `Arc<dyn PhysicalExpr>`, but the existing methods all deal with `Arc<dyn PhysicalExpr>`.
Fundamentally what makes all of this code weird / scary to me is that we are taking data from a typed thing (`SortExec`) moving into an untyped world (a tree of `ExecutionPlan`) and then trying to re-create the typed thing restoring all data (`filter/limit`).
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2971038192
I wonder if we see any improvements on the "sort tpch with limit" benchmark?
```cargo run --release --bin dfbench -- sort-tpch --iterations 5 --path "${TPCH_DIR}" -o "${RESULTS_FILE}" --limit 100```
https://github.com/apache/datafusion/pull/15697/files#diff-1769f5787dc11c8b1f1b48288cdf3c89d25a5b5cbc6be4740bfcc70a6313ba99R930
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2972642194
Results of applying this in TopK itself: https://github.com/apache/datafusion/pull/16408#issue-3145903557
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2083506643
##########
datafusion/physical-plan/src/topk/mod.rs:
##########
@@ -570,6 +680,47 @@ impl TopKHeap {
+ self.store.size()
+ self.owned_bytes
}
+
+ fn get_threshold_values(
+ &self,
+ sort_exprs: &[PhysicalSortExpr],
+ ) -> Result<Option<Vec<ScalarValue>>> {
+ // If the heap doesn't have k elements yet, we can't create thresholds
+ let max_row = match self.max() {
+ Some(row) => row,
+ None => return Ok(None),
+ };
+
+ // Get the batch that contains the max row
+ let batch_entry = match self.store.get(max_row.batch_id) {
+ Some(entry) => entry,
+ None => return internal_err!("Invalid batch ID in TopKRow"),
+ };
+
+ // Extract threshold values for each sort expression
+ let mut scalar_values = Vec::with_capacity(sort_exprs.len());
+ for sort_expr in sort_exprs {
Review Comment:
I wonder if it is worth to make this evaluates in a parallelized way π€
##########
datafusion/core/tests/fuzz_cases/topk_filter_pushdown.rs:
##########
@@ -0,0 +1,354 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::HashMap;
+use std::sync::{Arc, LazyLock};
+
+use arrow::array::{Int32Array, StringArray, StringDictionaryBuilder};
+use arrow::datatypes::Int32Type;
+use arrow::record_batch::RecordBatch;
+use arrow::util::pretty::pretty_format_batches;
+use arrow_schema::{DataType, Field, Schema};
+use datafusion::datasource::listing::{ListingOptions, ListingTable, ListingTableConfig};
+use datafusion::prelude::{SessionConfig, SessionContext};
+use datafusion_datasource::ListingTableUrl;
+use datafusion_datasource_parquet::ParquetFormat;
+use datafusion_execution::object_store::ObjectStoreUrl;
+use itertools::Itertools;
+use object_store::memory::InMemory;
+use object_store::path::Path;
+use object_store::{ObjectStore, PutPayload};
+use parquet::arrow::ArrowWriter;
+use rand::rngs::StdRng;
+use rand::{Rng, SeedableRng};
+use tokio::sync::Mutex;
+use tokio::task::JoinSet;
+
+#[derive(Clone)]
+struct TestDataSet {
+ store: Arc<dyn ObjectStore>,
+ schema: Arc<Schema>,
+}
+
+/// List of in memory parquet files with UTF8 data
+// Use a mutex rather than LazyLock to allow for async initialization
+static TESTFILES: LazyLock<Mutex<Vec<TestDataSet>>> =
+ LazyLock::new(|| Mutex::new(vec![]));
+
+async fn test_files() -> Vec<TestDataSet> {
+ let files_mutex = &TESTFILES;
+ let mut files = files_mutex.lock().await;
+ if !files.is_empty() {
+ return (*files).clone();
+ }
+
+ let mut rng = StdRng::seed_from_u64(0);
+
+ for nulls_in_ids in [false, true] {
+ for nulls_in_names in [false, true] {
+ for nulls_in_departments in [false, true] {
+ let store = Arc::new(InMemory::new());
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, nulls_in_ids),
+ Field::new("name", DataType::Utf8, nulls_in_names),
+ Field::new(
+ "department",
+ DataType::Dictionary(
+ Box::new(DataType::Int32),
+ Box::new(DataType::Utf8),
+ ),
+ nulls_in_departments,
+ ),
+ ]));
+
+ let name_choices = if nulls_in_names {
+ [Some("Alice"), Some("Bob"), None, Some("David"), None]
+ } else {
+ [
+ Some("Alice"),
+ Some("Bob"),
+ Some("Charlie"),
+ Some("David"),
+ Some("Eve"),
+ ]
+ };
+
+ let department_choices = if nulls_in_departments {
+ [
+ Some("Theater"),
+ Some("Engineering"),
+ None,
+ Some("Arts"),
+ None,
+ ]
+ } else {
+ [
+ Some("Theater"),
+ Some("Engineering"),
+ Some("Healthcare"),
+ Some("Arts"),
+ Some("Music"),
+ ]
+ };
+
+ // Generate 5 files, some with overlapping or repeated ids some without
+ for i in 0..5 {
+ let num_batches = rng.gen_range(1..3);
+ let mut batches = Vec::with_capacity(num_batches);
+ for _ in 0..num_batches {
+ let num_rows = 25;
+ let ids = Int32Array::from_iter((0..num_rows).map(|file| {
+ if nulls_in_ids {
+ if rng.gen_bool(1.0 / 10.0) {
+ None
+ } else {
+ Some(rng.gen_range(file..file + 5))
+ }
+ } else {
+ Some(rng.gen_range(file..file + 5))
+ }
+ }));
+ let names = StringArray::from_iter((0..num_rows).map(|_| {
+ // randomly select a name
+ let idx = rng.gen_range(0..name_choices.len());
+ name_choices[idx].map(|s| s.to_string())
+ }));
+ let mut departments = StringDictionaryBuilder::<Int32Type>::new();
+ for _ in 0..num_rows {
+ // randomly select a department
+ let idx = rng.gen_range(0..department_choices.len());
+ departments.append_option(department_choices[idx].as_ref());
+ }
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(ids),
+ Arc::new(names),
+ Arc::new(departments.finish()),
+ ],
+ )
+ .unwrap();
+ batches.push(batch);
+ }
+ let mut buf = vec![];
+ {
+ let mut writer =
+ ArrowWriter::try_new(&mut buf, schema.clone(), None).unwrap();
+ for batch in batches {
+ writer.write(&batch).unwrap();
+ writer.flush().unwrap();
+ }
+ writer.flush().unwrap();
+ writer.finish().unwrap();
+ }
+ let payload = PutPayload::from(buf);
+ let path = Path::from(format!("file_{i}.parquet"));
+ store.put(&path, payload).await.unwrap();
+ }
+ files.push(TestDataSet { store, schema });
+ }
+ }
+ }
+ (*files).clone()
+}
+
+async fn run_query_with_config(
+ query: &str,
+ config: SessionConfig,
+ dataset: TestDataSet,
+) -> Vec<RecordBatch> {
+ let store = dataset.store;
+ let schema = dataset.schema;
+ let ctx = SessionContext::new_with_config(config);
+ let url = ObjectStoreUrl::parse("memory://").unwrap();
+ ctx.register_object_store(url.as_ref(), store.clone());
+
+ let format = Arc::new(
+ ParquetFormat::default()
+ .with_options(ctx.state().table_options().parquet.clone()),
+ );
+ let options = ListingOptions::new(format);
+ let table_path = ListingTableUrl::parse("memory:///").unwrap();
+ let config = ListingTableConfig::new(table_path)
+ .with_listing_options(options)
+ .with_schema(schema);
+ let table = Arc::new(ListingTable::try_new(config).unwrap());
+
+ ctx.register_table("test_table", table).unwrap();
+
+ ctx.sql(query).await.unwrap().collect().await.unwrap()
+}
+
+#[derive(Debug)]
+struct RunQueryResult {
+ query: String,
+ result: Vec<RecordBatch>,
+ expected: Vec<RecordBatch>,
+}
+
+impl RunQueryResult {
+ fn expected_formated(&self) -> String {
+ format!("{}", pretty_format_batches(&self.expected).unwrap())
+ }
+
+ fn result_formated(&self) -> String {
+ format!("{}", pretty_format_batches(&self.result).unwrap())
+ }
+
+ fn is_ok(&self) -> bool {
+ self.expected_formated() == self.result_formated()
+ }
+}
+
+async fn run_query(
+ query: String,
+ cfg: SessionConfig,
+ dataset: TestDataSet,
+) -> RunQueryResult {
+ let cfg_with_dynamic_filters = cfg
+ .clone()
+ .set_bool("datafusion.optimizer.enable_dynamic_filter_pushdown", true);
+ let cfg_without_dynamic_filters = cfg
+ .clone()
+ .set_bool("datafusion.optimizer.enable_dynamic_filter_pushdown", false);
+
+ let expected_result =
+ run_query_with_config(&query, cfg_without_dynamic_filters, dataset.clone()).await;
+ let result =
+ run_query_with_config(&query, cfg_with_dynamic_filters, dataset.clone()).await;
+
+ RunQueryResult {
+ query: query.to_string(),
+ result,
+ expected: expected_result,
+ }
+}
+
+struct TestCase {
+ query: String,
+ cfg: SessionConfig,
+ dataset: TestDataSet,
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_fuzz_topk_filter_pushdown() {
+ let order_columns = ["id", "name", "department"];
+ let order_directions = ["ASC", "DESC"];
+ let null_orders = ["NULLS FIRST", "NULLS LAST"];
+
+ let start = datafusion_common::instant::Instant::now();
+ let mut orders: HashMap<String, Vec<String>> = HashMap::new();
+ for order_column in &order_columns {
+ for order_direction in &order_directions {
+ for null_order in &null_orders {
+ // if there is a vec for this column insert the order, otherwise create a new vec
+ let ordering =
+ format!("{} {} {}", order_column, order_direction, null_order);
+ match orders.get_mut(*order_column) {
+ Some(order_vec) => {
+ order_vec.push(ordering);
+ }
+ None => {
+ orders.insert(order_column.to_string(), vec![ordering]);
+ }
+ }
+ }
+ }
+ }
+
+ let mut queries = vec![];
+
+ for limit in [1, 10] {
+ for num_order_by_columns in [1, 2, 3] {
+ for order_columns in ["id", "name", "department"]
+ .iter()
+ .combinations(num_order_by_columns)
+ {
+ for orderings in order_columns
+ .iter()
+ .map(|col| orders.get(**col).unwrap())
+ .multi_cartesian_product()
+ {
+ let query = format!(
+ "SELECT * FROM test_table ORDER BY {} LIMIT {}",
+ orderings.into_iter().join(", "),
+ limit
+ );
+ queries.push(query);
+ }
+ }
+ }
+ }
+
+ queries.sort_unstable();
+ println!(
+ "Generated {} queries in {:?}",
+ queries.len(),
+ start.elapsed()
+ );
+
+ let start = datafusion_common::instant::Instant::now();
+ let datasets = test_files().await;
+ println!("Generated test files in {:?}", start.elapsed());
+
+ let mut test_cases = vec![];
+ for enable_filter_pushdown in [true, false] {
+ for query in &queries {
+ for dataset in &datasets {
+ let mut cfg = SessionConfig::new();
+ cfg = cfg.set_bool(
+ "datafusion.optimizer.enable_dynamic_filter_pushdown",
+ enable_filter_pushdown,
+ );
+ test_cases.push(TestCase {
+ query: query.to_string(),
+ cfg,
+ dataset: dataset.clone(),
+ });
+ }
+ }
+ }
+
+ let start = datafusion_common::instant::Instant::now();
+ let mut join_set = JoinSet::new();
+ for tc in test_cases {
+ join_set.spawn(run_query(tc.query, tc.cfg, tc.dataset));
Review Comment:
can we somehow assert that the filter is pushed down to the source dynamically in these tests?
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2869838330
> The rule issue is not very trivial because we cannot just track and eliminate some hardcoded patterns, since we also need to be aware of upper parts of the plan, and new patterns may emerge as new tests are added, it's not limited to some specific cases.
Yes this is unfortunate. Sadly I donβt think thatβs even specific to this change: just the way that rules work and are applied iteratively means thereβs bound to be conflicts. IMO each rule should have an apply and check step where eg it inserts repartitioning then on the way back up checks if itβs still needed and removes it if not. That way at least the logic is self contained with a rule.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2869838533
@berkaysynnada you might need to resolve the conflicts for CI to run
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2083517075
##########
datafusion/physical-optimizer/src/optimizer.rs:
##########
@@ -126,6 +119,13 @@ impl PhysicalOptimizer {
// into an `order by max(x) limit y`. In this case it will copy the limit value down
// to the aggregation, allowing it to use only y number of accumulators.
Arc::new(TopKAggregation::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:
Should we exclude this step if the pushdown config given as false?
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2083516712
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -229,7 +232,9 @@ EXPLAIN select * from t_pushdown where val != '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)]
+physical_plan
+01)RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=3
Review Comment:
The last issue is this. I think we need to eventually be able to run FilterPushdown rule before EnforceSorting.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2978414721
π€: Benchmark completed
<details><summary>Details</summary>
<p>
```
Comparing HEAD and topk-dynamic-filters
--------------------
Benchmark clickbench_extended.json
--------------------
ββββββββββββββββ³ββββββββββββββ³βββββββββββββββββββββββ³ββββββββββββ
β Query β HEAD β topk-dynamic-filters β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β QQuery 0 β 1932.37 ms β 1879.32 ms β no change β
β QQuery 1 β 700.00 ms β 708.26 ms β no change β
β QQuery 2 β 1327.98 ms β 1357.73 ms β no change β
β QQuery 3 β 675.15 ms β 663.31 ms β no change β
β QQuery 4 β 1358.70 ms β 1371.20 ms β no change β
β QQuery 5 β 14941.43 ms β 15114.86 ms β no change β
β QQuery 6 β 2003.62 ms β 1988.73 ms β no change β
β QQuery 7 β 1974.89 ms β 1889.14 ms β no change β
β QQuery 8 β 804.64 ms β 796.40 ms β no change β
ββββββββββββββββ΄ββββββββββββββ΄βββββββββββββββββββββββ΄ββββββββββββ
βββββββββββββββββββββββββββββββββββββββ³βββββββββββββ
β Benchmark Summary β β
β‘βββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 25718.78ms β
β Total Time (topk-dynamic-filters) β 25768.95ms β
β Average Time (HEAD) β 2857.64ms β
β Average Time (topk-dynamic-filters) β 2863.22ms β
β Queries Faster β 0 β
β Queries Slower β 0 β
β Queries with No Change β 9 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββββββ΄βββββββββββββ
--------------------
Benchmark clickbench_partitioned.json
--------------------
ββββββββββββββββ³ββββββββββββββ³βββββββββββββββββββββββ³ββββββββββββββββ
β Query β HEAD β topk-dynamic-filters β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β QQuery 0 β 15.60 ms β 15.34 ms β no change β
β QQuery 1 β 33.50 ms β 32.81 ms β no change β
β QQuery 2 β 80.76 ms β 80.51 ms β no change β
β QQuery 3 β 99.67 ms β 95.83 ms β no change β
β QQuery 4 β 628.76 ms β 577.08 ms β +1.09x faster β
β QQuery 5 β 871.37 ms β 852.85 ms β no change β
β QQuery 6 β 24.15 ms β 23.11 ms β no change β
β QQuery 7 β 35.78 ms β 35.47 ms β no change β
β QQuery 8 β 868.37 ms β 880.50 ms β no change β
β QQuery 9 β 1149.06 ms β 1167.61 ms β no change β
β QQuery 10 β 254.37 ms β 252.96 ms β no change β
β QQuery 11 β 285.04 ms β 280.14 ms β no change β
β QQuery 12 β 867.64 ms β 856.84 ms β no change β
β QQuery 13 β 1276.06 ms β 1264.78 ms β no change β
β QQuery 14 β 812.54 ms β 791.25 ms β no change β
β QQuery 15 β 767.98 ms β 778.42 ms β no change β
β QQuery 16 β 1628.17 ms β 1592.90 ms β no change β
β QQuery 17 β 1597.69 ms β 1583.41 ms β no change β
β QQuery 18 β 2883.08 ms β 2848.60 ms β no change β
β QQuery 19 β 83.29 ms β 85.79 ms β no change β
β QQuery 20 β 1102.75 ms β 1125.53 ms β no change β
β QQuery 21 β 1243.71 ms β 1264.80 ms β no change β
β QQuery 22 β 2065.35 ms β 2062.86 ms β no change β
β QQuery 23 β 7575.06 ms β 7234.54 ms β no change β
β QQuery 24 β 445.22 ms β 429.24 ms β no change β
β QQuery 25 β 374.18 ms β 300.39 ms β +1.25x faster β
β QQuery 26 β 510.11 ms β 424.15 ms β +1.20x faster β
β QQuery 27 β 1514.36 ms β 1528.90 ms β no change β
β QQuery 28 β 11858.75 ms β 12005.25 ms β no change β
β QQuery 29 β 532.06 ms β 537.76 ms β no change β
β QQuery 30 β 760.11 ms β 750.64 ms β no change β
β QQuery 31 β 793.67 ms β 786.40 ms β no change β
β QQuery 32 β 2473.13 ms β 2403.92 ms β no change β
β QQuery 33 β 3116.67 ms β 3104.59 ms β no change β
β QQuery 34 β 3141.07 ms β 3148.95 ms β no change β
β QQuery 35 β 1238.65 ms β 1195.36 ms β no change β
β QQuery 36 β 124.11 ms β 120.55 ms β no change β
β QQuery 37 β 57.10 ms β 55.05 ms β no change β
β QQuery 38 β 118.93 ms β 120.48 ms β no change β
β QQuery 39 β 189.95 ms β 195.15 ms β no change β
β QQuery 40 β 46.63 ms β 47.38 ms β no change β
β QQuery 41 β 44.56 ms β 43.14 ms β no change β
β QQuery 42 β 39.37 ms β 37.98 ms β no change β
ββββββββββββββββ΄ββββββββββββββ΄βββββββββββββββββββββββ΄ββββββββββββββββ
βββββββββββββββββββββββββββββββββββββββ³βββββββββββββ
β Benchmark Summary β β
β‘βββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 53628.39ms β
β Total Time (topk-dynamic-filters) β 53019.21ms β
β Average Time (HEAD) β 1247.17ms β
β Average Time (topk-dynamic-filters) β 1233.00ms β
β Queries Faster β 3 β
β Queries Slower β 0 β
β Queries with No Change β 40 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββββββ΄βββββββββββββ
--------------------
Benchmark tpch_mem_sf1.json
--------------------
ββββββββββββββββ³ββββββββββββ³βββββββββββββββββββββββ³ββββββββββββ
β Query β HEAD β topk-dynamic-filters β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β QQuery 1 β 101.86 ms β 98.87 ms β no change β
β QQuery 2 β 20.65 ms β 21.20 ms β no change β
β QQuery 3 β 32.27 ms β 32.46 ms β no change β
β QQuery 4 β 18.87 ms β 18.37 ms β no change β
β QQuery 5 β 50.65 ms β 49.11 ms β no change β
β QQuery 6 β 11.98 ms β 11.87 ms β no change β
β QQuery 7 β 85.68 ms β 86.77 ms β no change β
β QQuery 8 β 25.13 ms β 24.62 ms β no change β
β QQuery 9 β 54.08 ms β 53.86 ms β no change β
β QQuery 10 β 43.18 ms β 43.63 ms β no change β
β QQuery 11 β 11.47 ms β 11.92 ms β no change β
β QQuery 12 β 34.72 ms β 34.90 ms β no change β
β QQuery 13 β 25.32 ms β 25.76 ms β no change β
β QQuery 14 β 9.91 ms β 9.83 ms β no change β
β QQuery 15 β 19.76 ms β 19.29 ms β no change β
β QQuery 16 β 18.61 ms β 19.02 ms β no change β
β QQuery 17 β 95.21 ms β 95.70 ms β no change β
β QQuery 18 β 188.48 ms β 196.82 ms β no change β
β QQuery 19 β 25.87 ms β 25.89 ms β no change β
β QQuery 20 β 32.11 ms β 32.11 ms β no change β
β QQuery 21 β 148.35 ms β 149.18 ms β no change β
β QQuery 22 β 14.86 ms β 15.06 ms β no change β
ββββββββββββββββ΄ββββββββββββ΄βββββββββββββββββββββββ΄ββββββββββββ
βββββββββββββββββββββββββββββββββββββββ³ββββββββββββ
β Benchmark Summary β β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 1069.03ms β
β Total Time (topk-dynamic-filters) β 1076.26ms β
β Average Time (HEAD) β 48.59ms β
β Average Time (topk-dynamic-filters) β 48.92ms β
β Queries Faster β 0 β
β Queries Slower β 0 β
β Queries with No Change β 22 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββββββ΄ββββββββββββ
```
</p>
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2978414775
π€ `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running
Linux aal-dev 6.11.0-1015-gcp #15~24.04.1-Ubuntu SMP Thu Apr 24 20:41:05 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing topk-dynamic-filters (cd56084f922cccd27a7c847fc18eb9cd213578c1) to dd936cb1b25cb685e0e146f297950eb00048c64c [diff](https://github.com/apache/datafusion/compare/dd936cb1b25cb685e0e146f297950eb00048c64c..cd56084f922cccd27a7c847fc18eb9cd213578c1)
Benchmarks: sort_tpch
Results will be posted here when complete
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2981524048
Hm @adriangb another thing I wondered is `update_filter` does seem to take only the heap of the current partition into account, as in TopK (currently at least) each partition has it's own partition.
Perhaps we can compare against the current filter and only update the expression if it is greater?
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2150957763
##########
datafusion/physical-optimizer/src/filter_pushdown.rs:
##########
@@ -362,17 +363,25 @@ use itertools::izip;
/// [`ProjectionExec`]: datafusion_physical_plan::projection::ProjectionExec
/// [`AggregateExec`]: datafusion_physical_plan::aggregates::AggregateExec
#[derive(Debug)]
-pub struct FilterPushdown {}
+pub struct FilterPushdown {
+ phase: FilterPushdownPhase,
+ name: String,
+}
impl FilterPushdown {
- pub fn new() -> Self {
- Self {}
+ fn new(phase: FilterPushdownPhase) -> Self {
+ Self {
+ phase,
+ name: format!("FilterPushdown({phase})"),
Review Comment:
I feel that if we do that we should rename the enum to `Static/Dynamic`. I think it's confusing to mix `Static/Dynamic` and `Pre/Post`.
I like `Pre/Post` because that's the actual reason we're splitting it into two phases: one runs before other optimizations, one runs after.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Omega359 (via GitHub)" <gi...@apache.org>.
Omega359 commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-3025141986
Can someone (@adriangb maybe?) give me a high level summary of what was improved in this PR for the [DF 49 blog post](https://github.com/apache/datafusion/issues/16347)? The performance improvement highlighted in https://github.com/apache/datafusion/pull/15770#issuecomment-2981184419 seem to make it a worthy candidate for highlighting in the performance improvement section :)
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2146805677
##########
datafusion/common/src/config.rs:
##########
@@ -614,6 +614,13 @@ config_namespace! {
/// during aggregations, if possible
pub enable_topk_aggregation: bool, default = true
+ /// When set to true attempts to push down dynamic filters generated by operators into the file scan phase.
Review Comment:
I think we could also extend non-scan operators with the dynamic filter pushdown?
It would be interesting to extend filters / aggregates / joins / etc. with the support of this as well...
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2970903681
> And obviously we expect Q23 to get a huge bump once we enable filter pushdown...
>
> I think the only thing left here is for someone to review in detail π
I will do so later this afternoon
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2971653072
@alamb I opened https://github.com/pydantic/datafusion/pull/30 to explore the idea of having two pushdown phases. It's not complete (some failing tests, some TODOs) but I think it captures the idea.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2145545013
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -843,6 +846,8 @@ pub struct SortExec {
common_sort_prefix: Vec<PhysicalSortExpr>,
/// Cache holding plan properties like equivalences, output partitioning etc.
cache: PlanProperties,
+ /// Filter matching the state of the sort for dynamic filter pushdown
+ filter: Option<Arc<DynamicFilterPhysicalExpr>>,
Review Comment:
What isn't really clear to me from this PR - will this create a filter here regardless whether it can be pushed down or do we keep this as `None` if it can't? As there might be a small overhead of evaluating this filter.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2959969124
Here's the current Q23 benchmarks on my local machine on this branch using the following test script:
```sql
-- Current status quo before this PR: filter pushdown on, no dynamic filter pushdown, default target partitions.
SET datafusion.execution.parquet.pushdown_filters = true;
SET datafusion.optimizer.enable_dynamic_filter_pushdown = false;
SET datafusion.execution.target_partitions = 0;
explain analyze SELECT "EventTime" FROM 'benchmarks/data/hits_partitioned/' ORDER BY "EventTime" LIMIT 10;
-- With dynamic filters turned on
SET datafusion.execution.parquet.pushdown_filters = true;
SET datafusion.optimizer.enable_dynamic_filter_pushdown = true;
SET datafusion.execution.target_partitions = 0;
explain analyze SELECT "EventTime" FROM 'benchmarks/data/hits_partitioned/' ORDER BY "EventTime" LIMIT 10;
-- With filter pushdown off, dynamic filters on
SET datafusion.execution.parquet.pushdown_filters = false;
SET datafusion.optimizer.enable_dynamic_filter_pushdown = true;
SET datafusion.execution.target_partitions = 0;
SELECT "EventTime" FROM 'benchmarks/data/hits_partitioned/' ORDER BY "EventTime" LIMIT 10;
-- With dynamic filters off and target partitions set to 1
SET datafusion.execution.parquet.pushdown_filters = true;
SET datafusion.optimizer.enable_dynamic_filter_pushdown = false;
SET datafusion.execution.target_partitions = 1;
SELECT "EventTime" FROM 'benchmarks/data/hits_partitioned/' ORDER BY "EventTime" LIMIT 10;
-- With dynamic filters on and target partitions set to 1
SET datafusion.execution.parquet.pushdown_filters = true;
SET datafusion.optimizer.enable_dynamic_filter_pushdown = true;
SET datafusion.execution.target_partitions = 1;
SELECT "EventTime" FROM 'benchmarks/data/hits_partitioned/' ORDER BY "EventTime" LIMIT 10;
```
```
DataFusion CLI v48.0.0
> -- Current status quo before this PR: filter pushdown on, no dynamic filter pushdown, default target partitions.
SET datafusion.execution.parquet.pushdown_filters = true;
SET datafusion.optimizer.enable_dynamic_filter_pushdown = false;
SET datafusion.execution.target_partitions = 0;
SELECT "EventTime" FROM 'benchmarks/data/hits_partitioned/' ORDER BY "EventTime" LIMIT 10;
0 row(s) fetched.
Elapsed 0.003 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
+------------+
| EventTime |
+------------+
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
+------------+
10 row(s) fetched.
Elapsed 0.214 seconds.
> -- With dynamic filters turned on
SET datafusion.execution.parquet.pushdown_filters = true;
SET datafusion.optimizer.enable_dynamic_filter_pushdown = true;
SET datafusion.execution.target_partitions = 0;
SELECT "EventTime" FROM 'benchmarks/data/hits_partitioned/' ORDER BY "EventTime" LIMIT 10;
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
+------------+
| EventTime |
+------------+
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
+------------+
10 row(s) fetched.
Elapsed 0.091 seconds.
> -- With filter pushdown off, dynamic filters on
SET datafusion.execution.parquet.pushdown_filters = false;
SET datafusion.optimizer.enable_dynamic_filter_pushdown = true;
SET datafusion.execution.target_partitions = 0;
SELECT "EventTime" FROM 'benchmarks/data/hits_partitioned/' ORDER BY "EventTime" LIMIT 10;
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
+------------+
| EventTime |
+------------+
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
+------------+
10 row(s) fetched.
Elapsed 0.104 seconds.
> -- With dynamic filters off and target partitions set to 1
SET datafusion.execution.parquet.pushdown_filters = true;
SET datafusion.optimizer.enable_dynamic_filter_pushdown = false;
SET datafusion.execution.target_partitions = 1;
SELECT "EventTime" FROM 'benchmarks/data/hits_partitioned/' ORDER BY "EventTime" LIMIT 10;
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
+------------+
| EventTime |
+------------+
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
+------------+
10 row(s) fetched.
Elapsed 0.803 seconds.
> -- With dynamic filters on and target partitions set to 1
SET datafusion.execution.parquet.pushdown_filters = true;
SET datafusion.optimizer.enable_dynamic_filter_pushdown = true;
SET datafusion.execution.target_partitions = 1;
SELECT "EventTime" FROM 'benchmarks/data/hits_partitioned/' ORDER BY "EventTime" LIMIT 10;
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
+------------+
| EventTime |
+------------+
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
+------------+
10 row(s) fetched.
Elapsed 0.095 seconds.
```
<details>
<summary>And with explain plans</summary>
```
> -- Current status quo before this PR: filter pushdown on, no dynamic filter pushdown, default target partitions.
SET datafusion.execution.parquet.pushdown_filters = true;
SET datafusion.optimizer.enable_dynamic_filter_pushdown = false;
SET datafusion.execution.target_partitions = 0;
explain analyze SELECT "EventTime" FROM 'benchmarks/data/hits_partitioned/' ORDER BY "EventTime" LIMIT 10;
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan
|
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Plan with Metrics | SortPreservingMergeExec: [EventTime@0 ASC NULLS LAST], fetch=10, metrics=[output_rows=10, elapsed_compute=2.208Β΅s]
|
| | SortExec: TopK(fetch=10), expr=[EventTime@0 ASC NULLS LAST], preserve_partitioning=[true], filter=[EventTime@0 < 1372708802], metrics=[output_rows=120, elapsed_compute=454.554682ms, row_replacements=4239]
|
| | DataSourceExec: file_groups={12 groups: [[Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_0.parquet:0..122446530, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_1.parquet:0..174965044, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_10.parquet:0..101513258, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_11.parquet:0..118419888, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_12.parquet:0..149514164, ...], [Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_17.parquet:59860177..116867853, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_18.parquet:0..133119589, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_19.parquet:0..103692598, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_2.parquet:0..230595491, Users/adriangb/GitHub/datafusion/benchmarks/data/hit
s_partitioned/hits_20.parquet:0..85766533, ...], [Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_26.parquet:122668027..156510916, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_27.parquet:0..166286210, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_28.parquet:0..162772407, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_29.parquet:0..79213288, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_3.parquet:0..192507052, ...], [Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_35.parquet:54087341..153632381, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_36.parquet:0..92487304, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_37.parquet:0..108247781, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_38.parquet:0..132005180, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitione
d/hits_39.parquet:0..103522954, ...], [Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_42.parquet:118278445..288524057, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_43.parquet:0..299692947, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_44.parquet:0..242404750, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_45.parquet:0..148061387, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_46.parquet:0..92407680, ...], ...]}, projection=[EventTime], file_type=parquet, metrics=[output_rows=99997497, elapsed_compute=12ns, bytes_scanned=413239770, file_open_errors=0, file_scan_errors=0, num_predicate_creation_errors=0, page_index_rows_matched=0, page_index_rows_pruned=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, row_groups_matched_bloom_filter=0, row_groups_matched_statistics=0, row_groups_pruned_bloom_filter=0, row_groups_pruned_statistics=
0, bloom_filter_eval_time=222ns, metadata_load_time=131.111478ms, page_index_eval_time=222ns, row_pushdown_eval_time=222ns, statistics_eval_time=222ns, time_elapsed_opening=17.80283ms, time_elapsed_processing=607.505781ms, time_elapsed_scanning_total=1.202561798s, time_elapsed_scanning_until_data=185.184704ms] |
| |
|
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
1 row(s) fetched.
Elapsed 0.160 seconds.
> -- With dynamic filters turned on
SET datafusion.execution.parquet.pushdown_filters = true;
SET datafusion.optimizer.enable_dynamic_filter_pushdown = true;
SET datafusion.execution.target_partitions = 0;
explain analyze SELECT "EventTime" FROM 'benchmarks/data/hits_partitioned/' ORDER BY "EventTime" LIMIT 10;
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan
|
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Plan with Metrics | SortPreservingMergeExec: [EventTime@0 ASC NULLS LAST], fetch=10, metrics=[output_rows=10, elapsed_compute=2.208Β΅s]
|
| | SortExec: TopK(fetch=10), expr=[EventTime@0 ASC NULLS LAST], preserve_partitioning=[true], filter=[EventTime@0 < 1372708802], metrics=[output_rows=120, elapsed_compute=28.725461ms, row_replacements=2513]
|
| | DataSourceExec: file_groups={12 groups: [[Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_0.parquet:0..122446530, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_1.parquet:0..174965044, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_10.parquet:0..101513258, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_11.parquet:0..118419888, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_12.parquet:0..149514164, ...], [Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_17.parquet:59860177..116867853, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_18.parquet:0..133119589, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_19.parquet:0..103692598, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_2.parquet:0..230595491, Users/adriangb/GitHub/datafusion/benchmarks/data/hit
s_partitioned/hits_20.parquet:0..85766533, ...], [Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_26.parquet:122668027..156510916, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_27.parquet:0..166286210, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_28.parquet:0..162772407, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_29.parquet:0..79213288, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_3.parquet:0..192507052, ...], [Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_35.parquet:54087341..153632381, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_36.parquet:0..92487304, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_37.parquet:0..108247781, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_38.parquet:0..132005180, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitione
d/hits_39.parquet:0..103522954, ...], [Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_42.parquet:118278445..288524057, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_43.parquet:0..299692947, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_44.parquet:0..242404750, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_45.parquet:0..148061387, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_46.parquet:0..92407680, ...], ...]}, projection=[EventTime], file_type=parquet, predicate=DynamicFilterPhysicalExpr [ EventTime@0 < 1372708802 ], pruning_predicate=EventTime_null_count@1 != row_count@2 AND EventTime_min@0 < 1372708802, required_guarantees=[] |
| | , metrics=[output_rows=6238470, elapsed_compute=12ns, bytes_scanned=195749199, file_open_errors=0, file_scan_errors=0, num_predicate_creation_errors=0, page_index_rows_matched=0, page_index_rows_pruned=0, predicate_evaluation_errors=0, pushdown_rows_matched=6238470, pushdown_rows_pruned=40208246, row_groups_matched_bloom_filter=0, row_groups_matched_statistics=87, row_groups_pruned_bloom_filter=0, row_groups_pruned_statistics=180, bloom_filter_eval_time=329.981Β΅s, metadata_load_time=109.567194ms, page_index_eval_time=5.192Β΅s, row_pushdown_eval_time=10.217872ms, statistics_eval_time=610.473Β΅s, time_elapsed_opening=35.371746ms, time_elapsed_processing=413.242048ms, time_elapsed_scanning_total=506.896542ms, time_elapsed_scanning_until_data=326.002671ms]
|
| |
|
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
1 row(s) fetched.
Elapsed 0.090 seconds.
> -- With filter pushdown off, dynamic filters on
SET datafusion.execution.parquet.pushdown_filters = false;
SET datafusion.optimizer.enable_dynamic_filter_pushdown = true;
SET datafusion.execution.target_partitions = 0;
explain analyze SELECT "EventTime" FROM 'benchmarks/data/hits_partitioned/' ORDER BY "EventTime" LIMIT 10;
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan
|
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Plan with Metrics | SortPreservingMergeExec: [EventTime@0 ASC NULLS LAST], fetch=10, metrics=[output_rows=10, elapsed_compute=1.833Β΅s]
|
| | SortExec: TopK(fetch=10), expr=[EventTime@0 ASC NULLS LAST], preserve_partitioning=[true], filter=[EventTime@0 < 1372708804], metrics=[output_rows=120, elapsed_compute=198.546347ms, row_replacements=4021]
|
| | DataSourceExec: file_groups={12 groups: [[Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_0.parquet:0..122446530, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_1.parquet:0..174965044, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_10.parquet:0..101513258, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_11.parquet:0..118419888, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_12.parquet:0..149514164, ...], [Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_17.parquet:59860177..116867853, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_18.parquet:0..133119589, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_19.parquet:0..103692598, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_2.parquet:0..230595491, Users/adriangb/GitHub/datafusion/benchmarks/data/hit
s_partitioned/hits_20.parquet:0..85766533, ...], [Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_26.parquet:122668027..156510916, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_27.parquet:0..166286210, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_28.parquet:0..162772407, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_29.parquet:0..79213288, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_3.parquet:0..192507052, ...], [Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_35.parquet:54087341..153632381, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_36.parquet:0..92487304, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_37.parquet:0..108247781, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_38.parquet:0..132005180, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitione
d/hits_39.parquet:0..103522954, ...], [Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_42.parquet:118278445..288524057, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_43.parquet:0..299692947, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_44.parquet:0..242404750, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_45.parquet:0..148061387, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_46.parquet:0..92407680, ...], ...]}, projection=[EventTime], file_type=parquet, predicate=DynamicFilterPhysicalExpr [ EventTime@0 < 1372708804 ], pruning_predicate=EventTime_null_count@1 != row_count@2 AND EventTime_min@0 < 1372708804, required_guarantees=[] |
| | , metrics=[output_rows=43758473, elapsed_compute=12ns, bytes_scanned=184844240, file_open_errors=0, file_scan_errors=0, num_predicate_creation_errors=0, page_index_rows_matched=0, page_index_rows_pruned=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, row_groups_matched_bloom_filter=0, row_groups_matched_statistics=80, row_groups_pruned_bloom_filter=0, row_groups_pruned_statistics=186, bloom_filter_eval_time=297.062Β΅s, metadata_load_time=91.240148ms, page_index_eval_time=4.319Β΅s, row_pushdown_eval_time=222ns, statistics_eval_time=524.059Β΅s, time_elapsed_opening=36.330791ms, time_elapsed_processing=295.282703ms, time_elapsed_scanning_total=533.27498ms, time_elapsed_scanning_until_data=105.782465ms]
|
| |
|
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
1 row(s) fetched.
Elapsed 0.106 seconds.
> -- With dynamic filters off and target partitions set to 1
SET datafusion.execution.parquet.pushdown_filters = true;
SET datafusion.optimizer.enable_dynamic_filter_pushdown = false;
SET datafusion.execution.target_partitions = 1;
explain analyze SELECT "EventTime" FROM 'benchmarks/data/hits_partitioned/' ORDER BY "EventTime" LIMIT 10;
0 row(s) fetched.
Elapsed 0.001 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan
|
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Plan with Metrics | SortExec: TopK(fetch=10), expr=[EventTime@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[EventTime@0 < 1372708800], metrics=[output_rows=10, elapsed_compute=349.617653ms, row_replacements=367]
|
| | DataSourceExec: file_groups={1 group: [[Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_0.parquet:0..122446530, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_1.parquet:0..174965044, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_10.parquet:0..101513258, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_11.parquet:0..118419888, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_12.parquet:0..149514164, ...]]}, projection=[EventTime], file_type=parquet, metrics=[output_rows=99997497, elapsed_compute=1ns, bytes_scanned=413239770, file_open_errors=0, file_scan_errors=0, num_predicate_creation_errors=0, page_index_rows_matched=0, page_index_rows_pruned=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, row_groups_matched_bloom_filter=0, row_groups_matched_statistics=0, row_groups_pruned_bloom_filter=0, row_groups_pruned
_statistics=0, bloom_filter_eval_time=200ns, metadata_load_time=24.393942ms, page_index_eval_time=200ns, row_pushdown_eval_time=200ns, statistics_eval_time=200ns, time_elapsed_opening=1.642709ms, time_elapsed_processing=374.266438ms, time_elapsed_scanning_total=778.7721ms, time_elapsed_scanning_until_data=72.534498ms] |
| |
|
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
1 row(s) fetched.
Elapsed 0.823 seconds.
> -- With dynamic filters on and target partitions set to 1
SET datafusion.execution.parquet.pushdown_filters = true;
SET datafusion.optimizer.enable_dynamic_filter_pushdown = true;
SET datafusion.execution.target_partitions = 1;
explain analyze SELECT "EventTime" FROM 'benchmarks/data/hits_partitioned/' ORDER BY "EventTime" LIMIT 10;
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
+-------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+-------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Plan with Metrics | SortExec: TopK(fetch=10), expr=[EventTime@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[EventTime@0 < 1372708800], metrics=[output_rows=10, elapsed_compute=3.53979ms, row_replacements=367] |
| | DataSourceExec: file_groups={1 group: [[Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_0.parquet:0..122446530, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_1.parquet:0..174965044, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_10.parquet:0..101513258, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_11.parquet:0..118419888, Users/adriangb/GitHub/datafusion/benchmarks/data/hits_partitioned/hits_12.parquet:0..149514164, ...]]}, projection=[EventTime], file_type=parquet, predicate=DynamicFilterPhysicalExpr [ EventTime@0 < 1372708800 ], pruning_predicate=EventTime_null_count@1 != row_count@2 AND EventTime_min@0 < 1372708800, required_guarantees=[] |
| | , metrics=[output_rows=705744, elapsed_compute=1ns, bytes_scanned=15876270, file_open_errors=0, file_scan_errors=0, num_predicate_creation_errors=0, page_index_rows_matched=0, page_index_rows_pruned=0, predicate_evaluation_errors=0, pushdown_rows_matched=705744, pushdown_rows_pruned=3294256, row_groups_matched_bloom_filter=0, row_groups_matched_statistics=7, row_groups_pruned_bloom_filter=0, row_groups_pruned_statistics=313, bloom_filter_eval_time=23.77Β΅s, metadata_load_time=23.91685ms, page_index_eval_time=2.322Β΅s, row_pushdown_eval_time=1.768526ms, statistics_eval_time=204.189Β΅s, time_elapsed_opening=22.822957ms, time_elapsed_processing=48.607876ms, time_elapsed_scanning_total=35.223838ms, time_elapsed_scanning_until_data=11.836286ms] |
| | |
+-------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
1 row(s) fetched.
Elapsed 0.094 seconds.
```
</details>
The interesting bit is that this is now *faster even with predicate pushdown turned off* thanks to the late partition / stats based pruning @alamb !! For the case of a single partition it's **14x faster**:
```
> SET datafusion.execution.parquet.pushdown_filters = false;
SET datafusion.optimizer.enable_dynamic_filter_pushdown = false;
SET datafusion.execution.target_partitions = 1;
SELECT "EventTime" FROM 'benchmarks/data/hits_partitioned/' ORDER BY "EventTime" LIMIT 10;
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
+------------+
| EventTime |
+------------+
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
+------------+
10 row(s) fetched.
Elapsed 1.169 seconds.
> SET datafusion.execution.parquet.pushdown_filters = false;
SET datafusion.optimizer.enable_dynamic_filter_pushdown = true;
SET datafusion.execution.target_partitions = 1;
SELECT "EventTime" FROM 'benchmarks/data/hits_partitioned/' ORDER BY "EventTime" LIMIT 10;
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
0 row(s) fetched.
Elapsed 0.000 seconds.
+------------+
| EventTime |
+------------+
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
| 1372708800 |
+------------+
10 row(s) fetched.
Elapsed 0.082 seconds.
```
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2851724974
> @adriangb I'll complete reviewing this after merging other open PR's.
Thanks for all of the reviews @berkaysynnada. This one is now ready again.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2915135838
No worries I will try to take another look this week - I mainly wanted to make sure we didn't double 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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2970437633
> The interesting bit is that this is now faster even with predicate pushdown turned off thanks to the late partition / stats based pruning @alamb !! For the case of a single partition it's 14x faster:
That sounds pretty sweet.
I looked at the numbers and it seems like you have achieved the "at least as good performance and sometimes better" test that I think is needed for a feature like this. Hopefully we can get an additional confirmation and then we can merge this PR. That would be epic
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2970425995
> @alamb could you kick off your benchmarks run on this branch?
Sorry -- I missed this. I have queued them up and they should be running shortly
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2145280129
##########
datafusion/physical-plan/src/topk/mod.rs:
##########
@@ -207,6 +221,7 @@ impl TopK {
// Idea: filter out rows >= self.heap.max() early (before passing to `RowConverter`)
// this avoids some work and also might be better vectorizable.
let mut batch_entry = self.heap.register_batch(batch.clone());
+ let mut updated = false;
Review Comment:
Probably! Do you think we should add it in this PR or a followup?
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2150940087
##########
datafusion/physical-plan/src/execution_plan.rs:
##########
@@ -509,8 +510,22 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
///
/// The default implementation bars all parent filters from being pushed down and adds no new filters.
/// This is the safest option, making filter pushdown opt-in on a per-node pasis.
+ ///
+ /// Since this may perform deep modifications to the plan tree it is called early in the optimization phase
+ /// and is not expected to be called multiple times on the same plan.
+ ///
+ /// A quick summary of the phases is below, see [`FilterPushdownPhase`] for more details:
+ /// - [`FilterPushdownPhase::Pre`]: Filters get pushded down before most other optimizations are applied.
+ /// At this stage the plan can be modified (e.g. when [`ExecutionPlan::handle_child_pushdown_result`] is called the plan may choose to return an entirely new plan tree)
+ /// but subsequent optimizations may also rewrite the plan tree drastically, thus it is *not guaranteed* that a [`PhysicalExpr`] can hold on to a reference to the plan tree.
+ /// During this phase static filters (such as `col = 1`) are pushed down.
+ /// - [`FilterPushdownPhase::Post`]: Filters get pushed down after most other optimizations are applied.
+ /// At this stage the plan tree is expected to be stable and not change drastically, and operators that do filter pushdown during this phase should also not change the plan tree.
Review Comment:
I think the requirement is that the plan nodes don't change. Given that DynamicFilters effectively can have pointers to existing ExecutionPlan instances if a pass changes / removes / rewrites an `ExecutionPlan` that added a DynamicFilter I am not sure what will happen π€
##########
datafusion/common/src/config.rs:
##########
@@ -614,6 +614,13 @@ config_namespace! {
/// during aggregations, if possible
pub enable_topk_aggregation: bool, default = true
+ /// When set to true attempts to push down dynamic filters generated by operators into the file scan phase.
Review Comment:
Would the idea be to prune hash table state, for example, if we knew some of the groups were no longer needed?
I do think implementing more "late materialization" (aka turn on filter_pushdown) will help too
##########
datafusion/physical-plan/src/execution_plan.rs:
##########
@@ -548,10 +563,22 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
/// This can be used alongside [`FilterPushdownPropagation::with_filters`] and [`FilterPushdownPropagation::with_updated_node`]
/// to dynamically build a result with a mix of supported and unsupported filters.
///
+ /// There are two different phases in filter pushdown, which some operators may handle the same and some differently.
Review Comment:
While I love documentation, I would personally suggest not duplicating the docs here as duplicates can get out of sync, and instead leave a link to `FilterPushdownPhase` and focus on getting that documentation to be as clear as possible
##########
datafusion/physical-plan/src/filter_pushdown.rs:
##########
@@ -20,6 +20,39 @@ use std::vec::IntoIter;
use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
+#[derive(Debug, Clone, Copy)]
+pub enum FilterPushdownPhase {
+ /// Pushdown that happens before most other optimizations.
+ /// This pushdown allows static filters that do not reference any [`ExecutionPlan`]s to be pushed down.
+ /// Filters that reference an [`ExecutionPlan`] cannot be pushed down at this stage since the whole plan tree may be rewritten
+ /// by other optimizations.
+ /// Implemneters are however allowed to modify the execution plan themselves during this phase, for example by returning a completely
+ /// different [`ExecutionPlan`] from [`ExecutionPlan::handle_child_pushdown_result`].
+ ///
+ /// [`ExecutionPlan`]: crate::ExecutionPlan
+ /// [`ExecutionPlan::handle_child_pushdown_result`]: crate::ExecutionPlan::handle_child_pushdown_result
+ Pre,
+ /// Pushdown that happens after most other optimizations.
+ /// This pushdown allows filters that reference an [`ExecutionPlan`] to be pushed down.
+ /// It is guaranteed that subsequent optimizations will not make large changes to the plan tree,
Review Comment:
Aa above, I think it woudl be good to make it more precise what "large changes to the plan tree" means (basically I think it means don't remove existing ExecutionPlans ? π€ )
##########
datafusion/physical-optimizer/src/optimizer.rs:
##########
@@ -131,6 +131,8 @@ impl PhysicalOptimizer {
// replacing operators with fetching variants, or adding limits
// past operators that support limit pushdown.
Arc::new(LimitPushdown::new()),
+ // This FilterPushdown handles dynamic filters that may have references to the source ExecutionPlan
+ Arc::new(FilterPushdown::new_post_optimization()),
Review Comment:
this is so much nicer than adding it to `EnforceSorting` β€οΈ
But shouldn't this be the final pass? (maybe right before SanityCheckPlan?)
As I understand it, this filter pushdown pass has to be run after any pass that modifies the structure of the plan and `ProjectionPushdown` may actually do that π€
It also think it would be good to add a comment here explaining that `FilterPushdown::new_post_optimization()` must be run after all passes that change the structure of the plan as it can generate pointers from one plan to another
##########
datafusion/physical-optimizer/src/filter_pushdown.rs:
##########
@@ -362,17 +363,25 @@ use itertools::izip;
/// [`ProjectionExec`]: datafusion_physical_plan::projection::ProjectionExec
/// [`AggregateExec`]: datafusion_physical_plan::aggregates::AggregateExec
#[derive(Debug)]
-pub struct FilterPushdown {}
+pub struct FilterPushdown {
+ phase: FilterPushdownPhase,
+ name: String,
+}
impl FilterPushdown {
- pub fn new() -> Self {
- Self {}
+ fn new(phase: FilterPushdownPhase) -> Self {
+ Self {
+ phase,
+ name: format!("FilterPushdown({phase})"),
Review Comment:
I like `FilterPushdown` and `FilterPushdown(Dynamic)`
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2978219556
I queued up some benchmarks
Looking at naming 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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2978256702
Thank you Andrew! I will do the renames, docs edits, etc., push those tonight and we can merge this tomorrow evening if there is no more feedback.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2978416696
π€: Benchmark completed
<details><summary>Details</summary>
<p>
```
Comparing HEAD and topk-dynamic-filters
--------------------
Benchmark sort_tpch.json
--------------------
ββββββββββββββββ³βββββββββββββ³βββββββββββββββββββββββ³ββββββββββββββββ
β Query β HEAD β topk-dynamic-filters β Change β
β‘βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Q1 β 322.61 ms β 336.40 ms β no change β
β Q2 β 323.67 ms β 279.65 ms β +1.16x faster β
β Q3 β 1175.20 ms β 1152.85 ms β no change β
β Q4 β 415.50 ms β 414.71 ms β no change β
β Q5 β 419.42 ms β 429.43 ms β no change β
β Q6 β 460.61 ms β 462.88 ms β no change β
β Q7 β 939.97 ms β 928.19 ms β no change β
β Q8 β 793.68 ms β 791.83 ms β no change β
β Q9 β 830.90 ms β 832.78 ms β no change β
β Q10 β 1219.46 ms β 1237.03 ms β no change β
β Q11 β 718.13 ms β 737.31 ms β no change β
ββββββββββββββββ΄βββββββββββββ΄βββββββββββββββββββββββ΄ββββββββββββββββ
βββββββββββββββββββββββββββββββββββββββ³ββββββββββββ
β Benchmark Summary β β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 7619.16ms β
β Total Time (topk-dynamic-filters) β 7603.05ms β
β Average Time (HEAD) β 692.65ms β
β Average Time (topk-dynamic-filters) β 691.19ms β
β Queries Faster β 1 β
β Queries Slower β 0 β
β Queries with No Change β 10 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββββββ΄ββββββββββββ
```
</p>
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2981184419
π€: Benchmark completed
<details><summary>Details</summary>
<p>
```
Comparing HEAD and topk-dynamic-filters
--------------------
Benchmark run_topk_tpch.json
--------------------
ββββββββββββββββ³ββββββββββββ³βββββββββββββββββββββββ³ββββββββββββββββ
β Query β HEAD β topk-dynamic-filters β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Q1 β 41.78 ms β 28.62 ms β +1.46x faster β
β Q2 β 40.50 ms β 32.33 ms β +1.25x faster β
β Q3 β 144.67 ms β 101.83 ms β +1.42x faster β
β Q4 β 43.48 ms β 36.12 ms β +1.20x faster β
β Q5 β 31.85 ms β 26.77 ms β +1.19x faster β
β Q6 β 54.85 ms β 47.29 ms β +1.16x faster β
β Q7 β 129.61 ms β 138.38 ms β 1.07x slower β
β Q8 β 121.74 ms β 76.63 ms β +1.59x faster β
β Q9 β 158.13 ms β 114.49 ms β +1.38x faster β
β Q10 β 215.83 ms β 169.16 ms β +1.28x faster β
β Q11 β 121.35 ms β 83.83 ms β +1.45x faster β
ββββββββββββββββ΄ββββββββββββ΄βββββββββββββββββββββββ΄ββββββββββββββββ
βββββββββββββββββββββββββββββββββββββββ³ββββββββββββ
β Benchmark Summary β β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 1103.78ms β
β Total Time (topk-dynamic-filters) β 855.46ms β
β Average Time (HEAD) β 100.34ms β
β Average Time (topk-dynamic-filters) β 77.77ms β
β Queries Faster β 10 β
β Queries Slower β 1 β
β Queries with No Change β 0 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββββββ΄ββββββββββββ
```
</p>
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb merged PR #15770:
URL: https://github.com/apache/datafusion/pull/15770
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2052761831
##########
datafusion/physical-optimizer/src/push_down_filter.rs:
##########
@@ -382,7 +383,7 @@ impl PhysicalOptimizerRule for PushdownFilter {
context
.transform_up(|node| {
- if node.plan.as_any().downcast_ref::<FilterExec>().is_some() {
+ if node.plan.as_any().downcast_ref::<FilterExec>().is_some() || node.plan.as_any().downcast_ref::<SortExec>().is_some() {
Review Comment:
I propose an API something like this:
```rust
trait ExecutionPlan {
fn gather_filters_for_pushdown(
&self,
parent_filters: &[Arc<dyn ExecutionPlan>],
) -> Result<FilterPushdownPlan> {
let unsupported = vec![FilterPushdownSupport::Unsupported; parent_filters.len()];
Ok(
FilterPushdownPlan {
parent_filters_for_children: vec![unsupported; self.children().len()],
self_filters_for_children: vec![vec![]; self.children().len()],
},
)
}
fn propagate_filter_pushdown(
&self,
parent_pushdown_result: Vec<FilterPushdowChildResult>,
_self_filter_pushdown_result: Vec<FilterPushdowChildResult>,
) -> Result<FilterPushdownPropagation> {
Ok(
FilterPushdownPropagation {
parent_filter_result: parent_pushdown_result,
new_node: None,
},
)
}
}
pub struct FilterPushdownPropagation {
parent_filter_result: Vec<FilterPushdowChildResult>,
new_node: Option<Arc<dyn ExecutionPlan>>,
}
#[derive(Debug, Clone, Copy)]
pub enum FilterPushdowChildResult {
Supported,
Unsupported,
}
impl FilterPushdowChildResult {
}
#[derive(Debug, Clone)]
pub enum FilterPushdownSupport {
Supported(Arc<dyn PhysicalExpr>),
Unsupported,
}
#[derive(Debug, Clone)]
pub struct FilterPushdownPlan {
parent_filters_for_children: Vec<Vec<FilterPushdownSupport>>,
self_filters_for_children: Vec<Vec<FilterPushdownSupport>>,
}
```
The optimizer rule will have to do a bit of bookeeping and slicing correctly but this should avoid the need for any downcast matching or `retry` and minimize clones of plans. And it should do one walk down and up regardless of what ends up happening with the filters.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2083515926
##########
datafusion/core/tests/fuzz_cases/topk_filter_pushdown.rs:
##########
@@ -0,0 +1,354 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::HashMap;
+use std::sync::{Arc, LazyLock};
+
+use arrow::array::{Int32Array, StringArray, StringDictionaryBuilder};
+use arrow::datatypes::Int32Type;
+use arrow::record_batch::RecordBatch;
+use arrow::util::pretty::pretty_format_batches;
+use arrow_schema::{DataType, Field, Schema};
+use datafusion::datasource::listing::{ListingOptions, ListingTable, ListingTableConfig};
+use datafusion::prelude::{SessionConfig, SessionContext};
+use datafusion_datasource::ListingTableUrl;
+use datafusion_datasource_parquet::ParquetFormat;
+use datafusion_execution::object_store::ObjectStoreUrl;
+use itertools::Itertools;
+use object_store::memory::InMemory;
+use object_store::path::Path;
+use object_store::{ObjectStore, PutPayload};
+use parquet::arrow::ArrowWriter;
+use rand::rngs::StdRng;
+use rand::{Rng, SeedableRng};
+use tokio::sync::Mutex;
+use tokio::task::JoinSet;
+
+#[derive(Clone)]
+struct TestDataSet {
+ store: Arc<dyn ObjectStore>,
+ schema: Arc<Schema>,
+}
+
+/// List of in memory parquet files with UTF8 data
+// Use a mutex rather than LazyLock to allow for async initialization
+static TESTFILES: LazyLock<Mutex<Vec<TestDataSet>>> =
+ LazyLock::new(|| Mutex::new(vec![]));
+
+async fn test_files() -> Vec<TestDataSet> {
+ let files_mutex = &TESTFILES;
+ let mut files = files_mutex.lock().await;
+ if !files.is_empty() {
+ return (*files).clone();
+ }
+
+ let mut rng = StdRng::seed_from_u64(0);
+
+ for nulls_in_ids in [false, true] {
+ for nulls_in_names in [false, true] {
+ for nulls_in_departments in [false, true] {
+ let store = Arc::new(InMemory::new());
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, nulls_in_ids),
+ Field::new("name", DataType::Utf8, nulls_in_names),
+ Field::new(
+ "department",
+ DataType::Dictionary(
+ Box::new(DataType::Int32),
+ Box::new(DataType::Utf8),
+ ),
+ nulls_in_departments,
+ ),
+ ]));
+
+ let name_choices = if nulls_in_names {
+ [Some("Alice"), Some("Bob"), None, Some("David"), None]
+ } else {
+ [
+ Some("Alice"),
+ Some("Bob"),
+ Some("Charlie"),
+ Some("David"),
+ Some("Eve"),
+ ]
+ };
+
+ let department_choices = if nulls_in_departments {
+ [
+ Some("Theater"),
+ Some("Engineering"),
+ None,
+ Some("Arts"),
+ None,
+ ]
+ } else {
+ [
+ Some("Theater"),
+ Some("Engineering"),
+ Some("Healthcare"),
+ Some("Arts"),
+ Some("Music"),
+ ]
+ };
+
+ // Generate 5 files, some with overlapping or repeated ids some without
+ for i in 0..5 {
+ let num_batches = rng.gen_range(1..3);
+ let mut batches = Vec::with_capacity(num_batches);
+ for _ in 0..num_batches {
+ let num_rows = 25;
+ let ids = Int32Array::from_iter((0..num_rows).map(|file| {
+ if nulls_in_ids {
+ if rng.gen_bool(1.0 / 10.0) {
+ None
+ } else {
+ Some(rng.gen_range(file..file + 5))
+ }
+ } else {
+ Some(rng.gen_range(file..file + 5))
+ }
+ }));
+ let names = StringArray::from_iter((0..num_rows).map(|_| {
+ // randomly select a name
+ let idx = rng.gen_range(0..name_choices.len());
+ name_choices[idx].map(|s| s.to_string())
+ }));
+ let mut departments = StringDictionaryBuilder::<Int32Type>::new();
+ for _ in 0..num_rows {
+ // randomly select a department
+ let idx = rng.gen_range(0..department_choices.len());
+ departments.append_option(department_choices[idx].as_ref());
+ }
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(ids),
+ Arc::new(names),
+ Arc::new(departments.finish()),
+ ],
+ )
+ .unwrap();
+ batches.push(batch);
+ }
+ let mut buf = vec![];
+ {
+ let mut writer =
+ ArrowWriter::try_new(&mut buf, schema.clone(), None).unwrap();
+ for batch in batches {
+ writer.write(&batch).unwrap();
+ writer.flush().unwrap();
+ }
+ writer.flush().unwrap();
+ writer.finish().unwrap();
+ }
+ let payload = PutPayload::from(buf);
+ let path = Path::from(format!("file_{i}.parquet"));
+ store.put(&path, payload).await.unwrap();
+ }
+ files.push(TestDataSet { store, schema });
+ }
+ }
+ }
+ (*files).clone()
+}
+
+async fn run_query_with_config(
+ query: &str,
+ config: SessionConfig,
+ dataset: TestDataSet,
+) -> Vec<RecordBatch> {
+ let store = dataset.store;
+ let schema = dataset.schema;
+ let ctx = SessionContext::new_with_config(config);
+ let url = ObjectStoreUrl::parse("memory://").unwrap();
+ ctx.register_object_store(url.as_ref(), store.clone());
+
+ let format = Arc::new(
+ ParquetFormat::default()
+ .with_options(ctx.state().table_options().parquet.clone()),
+ );
+ let options = ListingOptions::new(format);
+ let table_path = ListingTableUrl::parse("memory:///").unwrap();
+ let config = ListingTableConfig::new(table_path)
+ .with_listing_options(options)
+ .with_schema(schema);
+ let table = Arc::new(ListingTable::try_new(config).unwrap());
+
+ ctx.register_table("test_table", table).unwrap();
+
+ ctx.sql(query).await.unwrap().collect().await.unwrap()
+}
+
+#[derive(Debug)]
+struct RunQueryResult {
+ query: String,
+ result: Vec<RecordBatch>,
+ expected: Vec<RecordBatch>,
+}
+
+impl RunQueryResult {
+ fn expected_formated(&self) -> String {
+ format!("{}", pretty_format_batches(&self.expected).unwrap())
+ }
+
+ fn result_formated(&self) -> String {
+ format!("{}", pretty_format_batches(&self.result).unwrap())
+ }
+
+ fn is_ok(&self) -> bool {
+ self.expected_formated() == self.result_formated()
+ }
+}
+
+async fn run_query(
+ query: String,
+ cfg: SessionConfig,
+ dataset: TestDataSet,
+) -> RunQueryResult {
+ let cfg_with_dynamic_filters = cfg
+ .clone()
+ .set_bool("datafusion.optimizer.enable_dynamic_filter_pushdown", true);
+ let cfg_without_dynamic_filters = cfg
+ .clone()
+ .set_bool("datafusion.optimizer.enable_dynamic_filter_pushdown", false);
+
+ let expected_result =
+ run_query_with_config(&query, cfg_without_dynamic_filters, dataset.clone()).await;
+ let result =
+ run_query_with_config(&query, cfg_with_dynamic_filters, dataset.clone()).await;
+
+ RunQueryResult {
+ query: query.to_string(),
+ result,
+ expected: expected_result,
+ }
+}
+
+struct TestCase {
+ query: String,
+ cfg: SessionConfig,
+ dataset: TestDataSet,
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_fuzz_topk_filter_pushdown() {
+ let order_columns = ["id", "name", "department"];
+ let order_directions = ["ASC", "DESC"];
+ let null_orders = ["NULLS FIRST", "NULLS LAST"];
+
+ let start = datafusion_common::instant::Instant::now();
+ let mut orders: HashMap<String, Vec<String>> = HashMap::new();
+ for order_column in &order_columns {
+ for order_direction in &order_directions {
+ for null_order in &null_orders {
+ // if there is a vec for this column insert the order, otherwise create a new vec
+ let ordering =
+ format!("{} {} {}", order_column, order_direction, null_order);
+ match orders.get_mut(*order_column) {
+ Some(order_vec) => {
+ order_vec.push(ordering);
+ }
+ None => {
+ orders.insert(order_column.to_string(), vec![ordering]);
+ }
+ }
+ }
+ }
+ }
+
+ let mut queries = vec![];
+
+ for limit in [1, 10] {
+ for num_order_by_columns in [1, 2, 3] {
+ for order_columns in ["id", "name", "department"]
+ .iter()
+ .combinations(num_order_by_columns)
+ {
+ for orderings in order_columns
+ .iter()
+ .map(|col| orders.get(**col).unwrap())
+ .multi_cartesian_product()
+ {
+ let query = format!(
+ "SELECT * FROM test_table ORDER BY {} LIMIT {}",
+ orderings.into_iter().join(", "),
+ limit
+ );
+ queries.push(query);
+ }
+ }
+ }
+ }
+
+ queries.sort_unstable();
+ println!(
+ "Generated {} queries in {:?}",
+ queries.len(),
+ start.elapsed()
+ );
+
+ let start = datafusion_common::instant::Instant::now();
+ let datasets = test_files().await;
+ println!("Generated test files in {:?}", start.elapsed());
+
+ let mut test_cases = vec![];
+ for enable_filter_pushdown in [true, false] {
+ for query in &queries {
+ for dataset in &datasets {
+ let mut cfg = SessionConfig::new();
+ cfg = cfg.set_bool(
+ "datafusion.optimizer.enable_dynamic_filter_pushdown",
+ enable_filter_pushdown,
+ );
+ test_cases.push(TestCase {
+ query: query.to_string(),
+ cfg,
+ dataset: dataset.clone(),
+ });
+ }
+ }
+ }
+
+ let start = datafusion_common::instant::Instant::now();
+ let mut join_set = JoinSet::new();
+ for tc in test_cases {
+ join_set.spawn(run_query(tc.query, tc.cfg, tc.dataset));
Review Comment:
Good call! I think so. Worst case we assert against the explain plan like we do in other places.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2083513975
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -86,7 +86,8 @@ logical_plan
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)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=2
Review Comment:
I think this is an improvement
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2083516211
##########
datafusion/physical-plan/src/topk/mod.rs:
##########
@@ -570,6 +680,47 @@ impl TopKHeap {
+ self.store.size()
+ self.owned_bytes
}
+
+ fn get_threshold_values(
+ &self,
+ sort_exprs: &[PhysicalSortExpr],
+ ) -> Result<Option<Vec<ScalarValue>>> {
+ // If the heap doesn't have k elements yet, we can't create thresholds
+ let max_row = match self.max() {
+ Some(row) => row,
+ None => return Ok(None),
+ };
+
+ // Get the batch that contains the max row
+ let batch_entry = match self.store.get(max_row.batch_id) {
+ Some(entry) => entry,
+ None => return internal_err!("Invalid batch ID in TopKRow"),
+ };
+
+ // Extract threshold values for each sort expression
+ let mut scalar_values = Vec::with_capacity(sort_exprs.len());
+ for sort_expr in sort_exprs {
Review Comment:
Maybe? We should measure overhead / execution time for a large case first. If itβs under a couple hundred ms itβs probably not worth the overhead. If we do parallelize I assume we can just pipe the asyncness down to here and use Tokio tasks?
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2865358754
> EnforceSorting and EnforceDistribution are initially designed to operate orthogonally, and there are many tests for that. But it seems somehow we broke it.
Maybe worth confirming that it is broken independently of this PR (on main)?
> Unless I can figure out easily to recover that orthogonality feature, I'll manually implement FilterPushdown rule to eliminate
Sounds good!
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2816927715
Marking as ready for review despite not having any numbers to substantiate performance improvement (because we need #15769) given that algorithmically and from experience in the previous PR we know this is a big win it might be okay to merge without interlocking PRs.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2051065573
##########
datafusion/physical-optimizer/src/push_down_filter.rs:
##########
@@ -382,7 +383,7 @@ impl PhysicalOptimizerRule for PushdownFilter {
context
.transform_up(|node| {
- if node.plan.as_any().downcast_ref::<FilterExec>().is_some() {
+ if node.plan.as_any().downcast_ref::<FilterExec>().is_some() || node.plan.as_any().downcast_ref::<SortExec>().is_some() {
Review Comment:
@berkaysynnada I didn't notice this in the original PR. This seems problematic. IMO doing downcast matching here is a smell that the API needs changing. It limits implementations to a hardcoded list of plans, which defeats the purpose of making DataFusion pluggable / having a `dyn ExecutionPlan`. The original implementation didn't require this. I think this goes hand-in hand with the `revisit` parameter. It seems that you were able to get from 3 methods down to 2 by replacing one of them with this downcast matching and the other with the extra recursion via the `revisit` parameter. It would be great to iterate on this and find a way to avoid the downcast matching.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2816688267
> @Dandandan I believe with this setup we should be able to achieve with a couple LOC in `insert_batch`:
>
> ```rust
> // Apply the filter to the batch before processing
> let filter = Arc::clone(&self.filter) as Arc<dyn PhysicalExpr>;
> let batch = filter_and_project(&batch, &filter, None, batch.schema_ref())?;
> if batch.num_rows() == 0 {
> return Ok(());
> }
> ```
>
> (`filter_and_project` is from `FilterExec`, we just need to make it `pub(crate)`)
Hmm. But I think
> @Dandandan I believe with this setup we should be able to achieve with a couple LOC in `insert_batch`:
>
> ```rust
> // Apply the filter to the batch before processing
> let filter = Arc::clone(&self.filter) as Arc<dyn PhysicalExpr>;
> let batch = filter_and_project(&batch, &filter, None, batch.schema_ref())?;
> if batch.num_rows() == 0 {
> return Ok(());
> }
> ```
>
> (`filter_and_project` is from `FilterExec`, we just need to make it `pub(crate)`)
I think we probably want to avoid filtering the entire batch, but indeed, if the filter expression is available it will be only a couple LOC!
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2136896235
##########
datafusion/physical-optimizer/src/enforce_sorting/sort_pushdown.rs:
##########
@@ -114,6 +118,18 @@ fn pushdown_sorts_helper(
sort_push_down.data.fetch = fetch;
sort_push_down.data.ordering_requirement =
Some(OrderingRequirements::from(sort_ordering));
+ let filter = plan
+ .as_any()
+ .downcast_ref::<SortExec>()
+ .and_then(|s| s.filter().clone());
Review Comment:
This is... not the prettiest. And is a smell that it will be hard for 3rd party plans to implement this.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2957654244
@berkaysynnada I think I made some progress. I was able to get the TopK pushdown tests passing by passing around the filter in the `EnforceSorting` rule as is already done for the `fetch` parameter. I struggled to wrap my head around this new to me bit of code and I mostly was taking stabs in the dark to try to understand it / get it to work, so as tests reflect I fear I may have broken some invariant. Could you take a look at 16fd477 and if you think that's a ballpark reasonable approach we schedule a call to try to fix it up together?
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2970734597
> QQuery 25 β 380.03 ms β 279.23 ms β +1.36x faster
```sql
SELECT SearchPhrase FROM hits WHERE SearchPhrase <> '' ORDER BY SearchPhrase LIMIT 10;
```
Looks like it might be real!
And obviously we expect Q23 to get a huge bump once we enable filter pushdown...
I think the only thing left here is for someone to review in detail π
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2145894014
##########
datafusion/physical-optimizer/src/enforce_sorting/sort_pushdown.rs:
##########
@@ -70,6 +71,8 @@ pub fn assign_initial_requirements(sort_push_down: &mut SortPushDown) {
// If the parent has a fetch value, assign it to the children
// Or use the fetch value of the child.
fetch: child.plan.fetch(),
+ // If the parent has a filter, assign it to the children
Review Comment:
That's where the actual pushdown is happening! This part is just so that that pushdown is not lost.
Basically: EnforceSorting wants to run last / late:
https://github.com/apache/datafusion/blob/61a2dfdc71508bd0b09abaefdc882d33a9e0d91e/datafusion/physical-optimizer/src/optimizer.rs#L111
Since `FilterPushdown` can absolutely modify the plan tree (e.g. by removing a `FilterExec`) it needs to run _before_ this (maybe we just shouldn't remove FilterExec and just make it a no-op, but that might loose optimizations e.g. because `CoalesceBatchesExec` can't then be removed).
But at the same time `EnforceSorting` removes and re-adds `SortExec`'s multiple times.
So what we are doing here is passing around the filters that were already pushed down so that the connection between the SortExec and the DataSourceExec is not lost. This is similar to what is happening with `limit` which is also being passed around.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2971391046
Thanks for the review @alamb!
I'll try to summarize the high level issues:
1. Mutli-column order by not working. This seems like a bug / oversight. Fundamentally I don't see any reason it shouldn't work, I'll have to investigate why.
2. Interaction with `EnforceSorting` & co. After grokking the code for a while I'm pretty convinced this is the only way to do things correctly. `SortExec` needs to establish a link (via references) with the `DataSourceExec`, and that needs to happen _before_ `EnforceSorting` runs because it might modify the plan in the process. So we have no choice but to teach `EnforceSorting` how to preserve this link. But that's the part I'm not sure I did correctly since I don't fully understand how `EnforceSorting` works. There may be other alternatives here. @berkaysynnada suggested that the right rule ordering might "just work", but neither of us could get it to. We might be able to split the filter pushdown into two steps: static (cannot make assumptions about reference links but can modify the plan tree, e.g. for `FitlerExec`) and dynamic (can make reference links but cannot modify the plan tree, e.g. `SortExec` and `HashJoin`).
3. Formatting in explain plans. I'm open to suggestions on this!
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2971510953
> 1. Mutli-column order by not working. This seems like a bug / oversight. Fundamentally I don't see any reason it shouldn't work, I'll have to investigate why.
I suggest we file a ticket explaining what is going on so we don't lose the context, but don't try and fix it in this PR
> 2. Interaction with `EnforceSorting` & co. After grokking the code for a while I'm pretty convinced this is the only way to do things correctly. `SortExec` needs to establish a link (via references) with the `DataSourceExec`, and that needs to happen _before_ `EnforceSorting` runs because it might modify the plan in the process. So we have no choice but to teach `EnforceSorting` how to preserve this link. But that's the part I'm not sure I did correctly since I don't fully understand how `EnforceSorting` works. There may be other alternatives here. @berkaysynnada suggested that the right rule ordering might "just work", but neither of us could get it to. We might be able to split the filter pushdown into two steps: static (cannot make assumptions about reference links but can modify the plan tree, e.g. for `FilterExec`) and dynamic (can make reference links but cannot modify the plan tree, e.g. `SortExec` and `HashJoin`). I think this would consist of adding a new optimizer rule
`DynamicFilterPushdown` or something like that and maybe a `Phase::{Dynamic,Static}` argument to the existing method on `ExecutionPlan`.
I just know we have also found EnforceSorting to be fragile (and there have been quite a few bugs related to limits being lost, etc) so I agree with your assesment that it is tough to know if it is correct.
I don't understand why `FilterPushdown` can't also push down DynamicFlters if it was run after `EnforceSorting` but I vaguely remember it being discussed and rejected before
> 3. Formatting in explain plans. I'm open to suggestions on this!
Likewise I think we should just file a follow on ticket. Maybe we can change the `Display` impl to be just `<DynamicFilter>` or something to make it clear it is something special (not just `false` constant)
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2971532504
Okay then I'll file a ticket for the multi-column sort and the display. But I do think we should hash out https://github.com/apache/datafusion/pull/15770#issuecomment-2971441638 in this PR.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2973096992
@alamb could you rerun the benchmarks?
Maybe also run the topk benchmark (`topk_tpch`).
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2150949273
##########
datafusion/physical-plan/src/execution_plan.rs:
##########
@@ -509,8 +510,22 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
///
/// The default implementation bars all parent filters from being pushed down and adds no new filters.
/// This is the safest option, making filter pushdown opt-in on a per-node pasis.
+ ///
+ /// Since this may perform deep modifications to the plan tree it is called early in the optimization phase
+ /// and is not expected to be called multiple times on the same plan.
+ ///
+ /// A quick summary of the phases is below, see [`FilterPushdownPhase`] for more details:
+ /// - [`FilterPushdownPhase::Pre`]: Filters get pushded down before most other optimizations are applied.
+ /// At this stage the plan can be modified (e.g. when [`ExecutionPlan::handle_child_pushdown_result`] is called the plan may choose to return an entirely new plan tree)
+ /// but subsequent optimizations may also rewrite the plan tree drastically, thus it is *not guaranteed* that a [`PhysicalExpr`] can hold on to a reference to the plan tree.
+ /// During this phase static filters (such as `col = 1`) are pushed down.
+ /// - [`FilterPushdownPhase::Post`]: Filters get pushed down after most other optimizations are applied.
+ /// At this stage the plan tree is expected to be stable and not change drastically, and operators that do filter pushdown during this phase should also not change the plan tree.
Review Comment:
Sort of. I am mincing my words in the comments because the reality is that to push down filters into `DataSourceExec` a new `DataSourceExec` has to be created and the whole tree has to be replaced "in place" to reference the new children. But the structure of the plan does not change, and it's pretty much guaranteed that `ExecutionPlan::new_with_children` does the right thing in terms of preserving internal state that might be pointed to (unlike `EnforceSorting`).
I'm not sure how to detail that in a comment, it's somewhat confusing.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2972552039
I am reapplying my PR on top of this branch, I'll report my results.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2976187450
> Sounds reasonable. Perhaps just keep it this way and see if we somehow can separate it to 2 different passes later.
>
>
>
> It seems like the branch https://github.com/pydantic/datafusion/pull/30 passes?
Yep so if you think that's reasonable I'd like to merge it and go with that instead for now even if we're not 100% sold on either direction it will let us keep iterating.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2855182639
This seems to be working and provides a tangible performance improvement BUT I don't think it's as good as it could be because of the Optimizer ordering. In particular:
```
explain SELECT "EventTime" FROM 'benchmarks/data/hits_partitioned/' WHERE "URL" LIKE '%google%' ORDER BY "EventTime" LIMIT 10;
```
```
+---------------+-------------------------------+
| plan_type | plan |
+---------------+-------------------------------+
| physical_plan | βββββββββββββββββββββββββββββ |
| | β SortPreservingMergeExec β |
| | β -------------------- β |
| | β EventTime ASC NULLS β |
| | β LASTlimit: β |
| | β 10 β |
| | βββββββββββββββ¬ββββββββββββββ |
| | βββββββββββββββ΄ββββββββββββββ |
| | β SortExec(TopK) β |
| | β -------------------- β |
| | β EventTime@0 ASC NULLS LASTβ |
| | β β |
| | β limit: 10 β |
| | βββββββββββββββ¬ββββββββββββββ |
| | βββββββββββββββ΄ββββββββββββββ |
| | β CoalesceBatchesExec β |
| | β -------------------- β |
| | β target_batch_size: β |
| | β 8192 β |
| | βββββββββββββββ¬ββββββββββββββ |
| | βββββββββββββββ΄ββββββββββββββ |
| | β ProjectionExec β |
| | β -------------------- β |
| | β EventTime: EventTime β |
| | βββββββββββββββ¬ββββββββββββββ |
| | βββββββββββββββ΄ββββββββββββββ |
| | β DataSourceExec β |
| | β -------------------- β |
| | β files: 111 β |
| | β format: parquet β |
| | β β |
| | β predicate: β |
| | β CAST(URL AS Utf8View) LIKEβ |
| | β %google% AND true β |
| | βββββββββββββββββββββββββββββ |
| | |
+---------------+-------------------------------+
```
I think the extra `CoalesceBatchesExec` may be hurting perf, but not sure... some more measurement is needed.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2816680541
@Dandandan I believe with this setup we should be able to achieve with a couple LOC in `insert_batch`:
```rust
// Apply the filter to the batch before processing
let filter = Arc::clone(&self.filter) as Arc<dyn PhysicalExpr>;
let batch = filter_and_project(&batch, &filter, None, batch.schema_ref())?;
if batch.num_rows() == 0 {
return Ok(());
}
```
(`filter_and_project` is from `FilterExec`, we just need to make it `pub(crate)`)
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2975351398
> > I think two different phases sounds good.
> > I don't really like complicating FilterPushdown with post / pre "phases". Why not creating the `PushDownDynamicFilters` one you suggested earlier?
>
> My reasoning for using an enum parameter was that there's a lot of shared implementation both on the optimizer side and ExectionPlan side. E.g. the transparent nodes like CoalesceBatchesExec don't really care which phase it is, but FilterExec needs to ignore the dynamic phase and SortExec needs to ignore the static filter pushdown phase.
Sounds reasonable. Perhaps just keep it this way and see if we somehow can separate it to 2 different passes later.
It seems like the branch https://github.com/pydantic/datafusion/pull/30 passes?
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2978330938
π€ `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running
Linux aal-dev 6.11.0-1015-gcp #15~24.04.1-Ubuntu SMP Thu Apr 24 20:41:05 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing topk-dynamic-filters (cd56084f922cccd27a7c847fc18eb9cd213578c1) to dd936cb1b25cb685e0e146f297950eb00048c64c [diff](https://github.com/apache/datafusion/compare/dd936cb1b25cb685e0e146f297950eb00048c64c..cd56084f922cccd27a7c847fc18eb9cd213578c1)
Benchmarks: tpch_mem clickbench_partitioned clickbench_extended
Results will be posted here when complete
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2974698767
> @Dandandan thank you so much for pushing this forward! Really appreciate the help and collaboration.
>
> Since you've been looking at the code do you have any thoughts on [#15770 (comment)](https://github.com/apache/datafusion/pull/15770#issuecomment-2971653072) / the discussion around implications for `EnforceSorting`, etc.?
I think two different phases sounds good.
I don't really like complicating FilterPushdown with post / pre "phases". Why not creating the `PushDownDynamicFilters` one you suggested earlier?
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2974699280
Apart from that, I agree with the sentiment to try eliminate it from `EnforceSorting`
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2981073603
> > π€: Benchmark completed
> > Details
>
> > π€: Benchmark completed
> > Details
>
> could you maybe confirm the topk benchmark results @alamb ? `topk_tpch`?
will 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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2981197663
> π€: Benchmark completed
>
> Details
Very nice improvement even without filter pushdown!
I'm going to merge this in the next couple of hours if there is no more feedback π
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2981263802
> > π€: Benchmark completed
> > Details
>
> Very nice improvement even without filter pushdown!
>
> I'm going to merge this in the next couple of hours if there is no more feedback π
This is super nice.
I think the speed up changes a bit due to more partitions as you shared earlier. But it's starting to look really nice π
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2981275327
> I think the speed up changes a bit due to more partitions
And maybe https://github.com/apache/datafusion/pull/16424 will speed up the wide partitions case by stopping those scans early!
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2981292465
I'll also run some profiling on those topk benchmarks to see if there is any further low hanging fruit.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-3025443600
High level it is described in https://github.com/apache/datafusion/issues/15037
I think @adriangb is considering a writeup about it directly here:
- https://github.com/apache/datafusion/issues/15513
(I am standing by to help with both BTW)
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "Omega359 (via GitHub)" <gi...@apache.org>.
Omega359 commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-3025461502
Awesome, thanks. Yes, I think a full blog post for this would be amazing but I'd still like to have it in the release blog post with a short summary and maybe a few stats.
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2971198838
I also filed a ticket to add a metric that we can use to see when file pruning is working:
- https://github.com/apache/datafusion/issues/16402
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#discussion_r2145903411
##########
datafusion/physical-optimizer/src/enforce_sorting/sort_pushdown.rs:
##########
@@ -70,6 +71,8 @@ pub fn assign_initial_requirements(sort_push_down: &mut SortPushDown) {
// If the parent has a fetch value, assign it to the children
// Or use the fetch value of the child.
fetch: child.plan.fetch(),
+ // If the parent has a filter, assign it to the children
Review Comment:
Thank you -- this makes sense to me
Maybe we can clarify with some additional comments to help the future readers
--
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] TopK dynamic filter pushdown attempt 2 [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #15770:
URL: https://github.com/apache/datafusion/pull/15770#issuecomment-2973326482
@Dandandan thank you so much for pushing this forward! Really appreciate the help and collaboration.
Since you've been looking at the code do you have any thoughts on https://github.com/apache/datafusion/pull/15770#issuecomment-2971653072 / the discussion around implications for `EnforceSorting`, etc.?
--
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