You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "alihandroid (via GitHub)" <gi...@apache.org> on 2024/07/25 15:24:50 UTC
[PR] Add LimitPushdown optimization rule and CoalesceBatchesExec fetch [datafusion]
alihandroid opened a new pull request, #11652:
URL: https://github.com/apache/datafusion/pull/11652
## Which issue does this PR close?
Closes #9792.
## Rationale for this change
Physical plans can be optimized further by pushing `GlobalLimitExec` and `LocalLimitExec` down through certain nodes, or using versions of their children nodes with fetch limits, without changing the result. This reduces unnecessary data transfer and processing for a more efficient plan execution.
`CoalesceBatchesExec` can also benefit from this improvement, and as such, a fetch limit functionality is implemented for it.
For example,
```
GlobalLimitExec: skip=0, fetch=5
StreamingTableExec: partition_sizes=1, projection=[c1, c2, c3], infinite_source=true
```
can be turned into
```
StreamingTableExec: partition_sizes=1, projection=[c1, c2, c3], infinite_source=true, fetch=5
```
and
```
GlobalLimitExec: skip=0, fetch=5
CoalescePartitionsExec
FilterExec: c3@2 > 0
RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1
StreamingTableExec: partition_sizes=1, projection=[c1, c2, c3], infinite_source=true
```
can be turned into
```
GlobalLimitExec: skip=0, fetch=5
CoalescePartitionsExec
LocalLimitExec: fetch=5
FilterExec: c3@2 > 0
RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1
StreamingTableExec: partition_sizes=1, projection=[c1, c2, c3], infinite_source=true
```
without changing the result, but using fewer resources and finishing faster
The physical plan in the following excerpt
https://github.com/apache/datafusion/blob/ecf5323eaa38869ed2f911b02f98e17aa6db639a/datafusion/sqllogictest/test_files/repartition.slt#L116-L129
will turn into
```
01)GlobalLimitExec: skip=0, fetch=5
02)--CoalescePartitionsExec
03)----CoalesceBatchesExec: target_batch_size=8192, fetch=5
04)------FilterExec: c3@2 > 0
05)--------RepartitionExec: partitioning=RoundRobinBatch(3), input_partitions=1
06)----------StreamingTableExec: partition_sizes=1, projection=[c1, c2, c3], infinite_source=true
```
Other examples can be found in the tests provided in `limit_pushdown.rs` and other .slt tests
## What changes are included in this PR?
Implement `LimitPushdown` Rule:
- Introduced new APIs in the `ExecutionPlan` trait:
- `with_fetch(&self, fetch: Option<usize>) -> Option<Arc<dyn ExecutionPlan>>`: Returns fetching version if supported, None otherwise. The default implementation returns None
- `supports_limit_pushdown(&self) -> bool`: Returns true if a node supports limit pushdown. The default implemenation returns false
Add fetch support to `CoalesceBatchesExec`:
- Add `fetch` field and `with_fetch` implementation
- Implement fetch limit functionality
## Are these changes tested?
Unit tests are provided for `LimitPushdown` and the new fetching support for `CoalesceBatchesExec`
## Are there any user-facing changes?
No. The changes only affect performance
--
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] Add LimitPushdown optimization rule and CoalesceBatchesExec fetch [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #11652:
URL: https://github.com/apache/datafusion/pull/11652#discussion_r1693887040
##########
datafusion/physical-plan/src/lib.rs:
##########
@@ -428,6 +428,18 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
fn statistics(&self) -> Result<Statistics> {
Ok(Statistics::new_unknown(&self.schema()))
}
+
+ /// Returns `true` if a limit can be safely pushed down through this
+ /// `ExecutionPlan` node.
+ fn supports_limit_pushdown(&self) -> bool {
Review Comment:
I am also not sure of the best name. If we find something that's clearly better, maybe we can change it in a follow-on before a public release.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add LimitPushdown optimization rule and CoalesceBatchesExec fetch [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #11652:
URL: https://github.com/apache/datafusion/pull/11652#discussion_r1693942386
##########
datafusion/sqllogictest/test_files/group_by.slt:
##########
@@ -4334,8 +4335,9 @@ physical_plan
01)GlobalLimitExec: skip=0, fetch=5
02)--SortPreservingMergeExec: [name@0 DESC,time_chunks@1 DESC], fetch=5
03)----ProjectionExec: expr=[name@0 as name, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 900000000000 }, ts@1) as time_chunks]
-04)------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1
-05)--------StreamingTableExec: partition_sizes=1, projection=[name, ts], infinite_source=true, output_ordering=[name@0 DESC, ts@1 DESC]
+04)------LocalLimitExec: fetch=5
+05)--------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1
Review Comment:
If I am able to pull the CoalesceBatch into Repartition I think adding support for limit in repartition will become quite easy (as the actual limit code will be reused)
##########
datafusion/core/src/physical_optimizer/limit_pushdown.rs:
##########
@@ -0,0 +1,661 @@
+// 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.
+
+//! This rule reduces the amount of data transferred by pushing down limits as much as possible.
+
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use crate::error::Result;
+use crate::physical_optimizer::PhysicalOptimizerRule;
+use crate::physical_plan::ExecutionPlan;
+
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::plan_datafusion_err;
+use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion_optimizer::push_down_limit::combine_limit;
+use datafusion_physical_plan::coalesce_partitions::CoalescePartitionsExec;
+use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
+use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
+
+/// This rule inspects [`ExecutionPlan`]'s and pushes down the fetch limit from
+/// the parent to the child if applicable.
+#[derive(Default)]
+pub struct LimitPushdown {}
+
+impl LimitPushdown {
+ #[allow(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for LimitPushdown {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ _config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ plan.transform_down(push_down_limits).data()
+ }
+
+ fn name(&self) -> &str {
+ "LimitPushdown"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// This enumeration makes `skip` and `fetch` calculations easier by providing
+/// a single API for both local and global limit operators.
+#[derive(Debug)]
+enum LimitExec {
+ Global(GlobalLimitExec),
+ Local(LocalLimitExec),
+}
+
+impl LimitExec {
+ fn input(&self) -> &Arc<dyn ExecutionPlan> {
+ match self {
+ Self::Global(global) => global.input(),
+ Self::Local(local) => local.input(),
+ }
+ }
+
+ fn fetch(&self) -> Option<usize> {
+ match self {
+ Self::Global(global) => global.fetch(),
+ Self::Local(local) => Some(local.fetch()),
+ }
+ }
+
+ fn skip(&self) -> usize {
+ match self {
+ Self::Global(global) => global.skip(),
+ Self::Local(_) => 0,
+ }
+ }
+
+ fn with_child(&self, child: Arc<dyn ExecutionPlan>) -> Self {
+ match self {
+ Self::Global(global) => {
+ Self::Global(GlobalLimitExec::new(child, global.skip(), global.fetch()))
+ }
+ Self::Local(local) => Self::Local(LocalLimitExec::new(child, local.fetch())),
+ }
+ }
+}
+
+impl From<LimitExec> for Arc<dyn ExecutionPlan> {
+ fn from(limit_exec: LimitExec) -> Self {
+ match limit_exec {
+ LimitExec::Global(global) => Arc::new(global),
+ LimitExec::Local(local) => Arc::new(local),
+ }
+ }
+}
+
+/// Pushes down the limit through the plan.
+pub fn push_down_limits(
+ plan: Arc<dyn ExecutionPlan>,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let maybe_modified = if let Some(limit_exec) = extract_limit(&plan) {
+ let child = limit_exec.input();
+ if let Some(child_limit) = extract_limit(child) {
+ let merged = merge_limits(&limit_exec, &child_limit);
+ // Revisit current node in case of consecutive pushdowns
+ Some(push_down_limits(merged)?.data)
+ } else if child.supports_limit_pushdown() {
+ try_push_down_limit(&limit_exec, child.clone())?
+ } else {
+ add_fetch_to_child(&limit_exec, child.clone())
+ }
+ } else {
+ None
+ };
+
+ Ok(maybe_modified.map_or(Transformed::no(plan), Transformed::yes))
+}
+
+/// Transforms the [`ExecutionPlan`] into a [`LimitExec`] if it is a
+/// [`GlobalLimitExec`] or a [`LocalLimitExec`].
+fn extract_limit(plan: &Arc<dyn ExecutionPlan>) -> Option<LimitExec> {
+ if let Some(global_limit) = plan.as_any().downcast_ref::<GlobalLimitExec>() {
+ Some(LimitExec::Global(GlobalLimitExec::new(
+ global_limit.input().clone(),
+ global_limit.skip(),
+ global_limit.fetch(),
+ )))
+ } else {
+ plan.as_any()
+ .downcast_ref::<LocalLimitExec>()
+ .map(|local_limit| {
+ LimitExec::Local(LocalLimitExec::new(
+ local_limit.input().clone(),
+ local_limit.fetch(),
+ ))
+ })
+ }
+}
+
+/// Merge the limits of the parent and the child. If at least one of them is a
+/// [`GlobalLimitExec`], the result is also a [`GlobalLimitExec`]. Otherwise,
+/// the result is a [`LocalLimitExec`].
+fn merge_limits(
Review Comment:
structurally this could be a method on `LimitExec` too , though this is totally fine too
--
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] Add LimitPushdown optimization rule and CoalesceBatchesExec fetch [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on PR #11652:
URL: https://github.com/apache/datafusion/pull/11652#issuecomment-2253830132
I will wait for a little bit longer to merge this in case anyone has 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] Add LimitPushdown optimization rule and CoalesceBatchesExec fetch [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #11652:
URL: https://github.com/apache/datafusion/pull/11652#issuecomment-2251274996
I believe if you merge up from main the CI will pass on 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] Add LimitPushdown optimization rule and CoalesceBatchesExec fetch [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #11652:
URL: https://github.com/apache/datafusion/pull/11652#discussion_r1693622103
##########
datafusion/physical-plan/src/lib.rs:
##########
@@ -428,6 +428,18 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
fn statistics(&self) -> Result<Statistics> {
Ok(Statistics::new_unknown(&self.schema()))
}
+
+ /// Returns `true` if a limit can be safely pushed down through this
+ /// `ExecutionPlan` node.
+ fn supports_limit_pushdown(&self) -> bool {
Review Comment:
I found this name somewhat confusing as it implied to me it was reporting if `with_fetch` was implemented, where now I see it refers to if it is ok to push a limit through the node
Perhaps we could rename it to something different. Perhaps `can_push_limit` or `preserves_limit` ?
I don't feel strongly about this
##########
datafusion/sqllogictest/test_files/explain.slt:
##########
@@ -79,10 +79,7 @@ logical_plan
01)Limit: skip=0, fetch=10
02)--Sort: aggregate_test_100_with_order.c1 ASC NULLS LAST, fetch=10
03)----TableScan: aggregate_test_100_with_order projection=[c1]
-physical_plan
-01)GlobalLimitExec: skip=0, fetch=10
-02)--CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/aggregate_test_100_order_by_c1_asc.csv]]}, projection=[c1], output_ordering=[c1@0 ASC NULLS LAST], has_header=true
-
+physical_plan CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/aggregate_test_100_order_by_c1_asc.csv]]}, projection=[c1], limit=10, output_ordering=[c1@0 ASC NULLS LAST], has_header=true
Review Comment:
The `GlobalLimitExec` is not needed here because the `CsvExec` already has a limit and there is a single partition ✅
##########
datafusion/physical-plan/src/lib.rs:
##########
@@ -428,6 +428,18 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
fn statistics(&self) -> Result<Statistics> {
Ok(Statistics::new_unknown(&self.schema()))
}
+
+ /// Returns `true` if a limit can be safely pushed down through this
+ /// `ExecutionPlan` node.
+ fn supports_limit_pushdown(&self) -> bool {
+ false
+ }
+
+ /// Returns a fetching variant of this `ExecutionPlan` node, if it supports
Review Comment:
💯
##########
datafusion/core/src/physical_optimizer/limit_pushdown.rs:
##########
@@ -0,0 +1,661 @@
+// 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.
+
+//! This rule reduces the amount of data transferred by pushing down limits as much as possible.
Review Comment:
This is the summary that is rendered on the rustdocs page, so it helps to put a link tot he actual struct
Something like
```suggestion
//! [`LimitPushdown`] pushes `LIMIT` down in `ExecutionPlan`s to reduce data transfer
```
##########
datafusion/physical-plan/src/lib.rs:
##########
@@ -428,6 +428,18 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
fn statistics(&self) -> Result<Statistics> {
Ok(Statistics::new_unknown(&self.schema()))
}
+
+ /// Returns `true` if a limit can be safely pushed down through this
+ /// `ExecutionPlan` node.
Review Comment:
I think it might help to also explain the implications of returning true. Something like
```suggestion
/// `ExecutionPlan` node.
///
/// If this method returns true, and the query contains a `FETCH`/`LIMIT`
/// at the output of this node, DataFusion will push the Fetch/Limit to
/// the input of this node
```
##########
datafusion/sqllogictest/test_files/explain.slt:
##########
@@ -304,9 +300,7 @@ CREATE EXTERNAL TABLE alltypes_plain STORED AS PARQUET LOCATION '../../parquet-t
query TT
EXPLAIN SELECT * FROM alltypes_plain limit 10;
----
-physical_plan
-01)GlobalLimitExec: skip=0, fetch=10, statistics=[Rows=Exact(8), Bytes=Absent, [(Col[0]:),(Col[1]:),(Col[2]:),(Col[3]:),(Col[4]:),(Col[5]:),(Col[6]:),(Col[7]:),(Col[8]:),(Col[9]:),(Col[10]:)]]
-02)--ParquetExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col], limit=10, statistics=[Rows=Exact(8), Bytes=Absent, [(Col[0]:),(Col[1]:),(Col[2]:),(Col[3]:),(Col[4]:),(Col[5]:),(Col[6]:),(Col[7]:),(Col[8]:),(Col[9]:),(Col[10]:)]]
+physical_plan ParquetExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col], limit=10, statistics=[Rows=Exact(8), Bytes=Absent, [(Col[0]:),(Col[1]:),(Col[2]:),(Col[3]:),(Col[4]:),(Col[5]:),(Col[6]:),(Col[7]:),(Col[8]:),(Col[9]:),(Col[10]:)]]
Review Comment:
I agree this is a better plan 👍 as it has only a single partition the global limit is unecessary
##########
datafusion/physical-plan/src/coalesce_batches.rs:
##########
@@ -160,6 +176,16 @@ impl ExecutionPlan for CoalesceBatchesExec {
fn statistics(&self) -> Result<Statistics> {
self.input.statistics()
}
+
+ fn with_fetch(&self, limit: Option<usize>) -> Option<Arc<dyn ExecutionPlan>> {
Review Comment:
It might help to document somewhere (perhaps the struct's documentation) why setting `limit` is valuable on `CoalesceBatchesExec ` -- in this case I think it is because `CoalesceBatchesExec` buffers data so it may buffer more than necessary waiting for a limit to complete
##########
datafusion/sqllogictest/test_files/group_by.slt:
##########
@@ -4334,8 +4335,9 @@ physical_plan
01)GlobalLimitExec: skip=0, fetch=5
02)--SortPreservingMergeExec: [name@0 DESC,time_chunks@1 DESC], fetch=5
03)----ProjectionExec: expr=[name@0 as name, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 900000000000 }, ts@1) as time_chunks]
-04)------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1
-05)--------StreamingTableExec: partition_sizes=1, projection=[name, ts], infinite_source=true, output_ordering=[name@0 DESC, ts@1 DESC]
+04)------LocalLimitExec: fetch=5
+05)--------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1
Review Comment:
I wonder why wasn't the limit pushed into the `RepartitionExec` 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] Add LimitPushdown optimization rule and CoalesceBatchesExec fetch [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #11652:
URL: https://github.com/apache/datafusion/pull/11652#discussion_r1692893313
##########
datafusion/core/src/physical_optimizer/limit_pushdown.rs:
##########
@@ -166,37 +154,111 @@ fn extract_limit(plan: &Arc<dyn ExecutionPlan>) -> Option<LimitExec> {
}
}
-/// Merge the limits of the parent and the child. The resulting [`ExecutionPlan`]
-/// is the same as the parent.
-fn try_merge_limits(
- limit_exec: &LimitExec,
- limit: &LocalLimitExec,
-) -> Result<Arc<dyn ExecutionPlan>> {
- let parent_skip = limit_exec.skip();
- let parent_fetch = limit_exec.fetch();
- let parent_max = parent_fetch.map(|f| f + parent_skip);
- let child_fetch = limit.fetch();
-
- if let Some(parent_max) = parent_max {
- if child_fetch >= parent_max {
- // Child fetch is larger than or equal to parent max, so we can remove the child
- Ok(limit_exec.with_child(limit.input().clone()).into())
- } else if child_fetch > parent_skip {
- // Child fetch is larger than parent skip, so we can trim the parent fetch
- Ok(limit_exec
- .with_child(limit.input().clone())
- .with_fetch(child_fetch - parent_skip)
- .into())
- } else {
- // This would return an empty result
- Err(plan_datafusion_err!("Child fetch is less than parent skip"))
+/// Merge the limits of the parent and the child. If at least one of them is a
+/// [`GlobalLimitExec`], the result is also a [`GlobalLimitExec`]. Otherwise,
+/// the result is a [`LocalLimitExec`].
+fn merge_limits(
+ parent_limit_exec: &LimitExec,
+ child_limit_exec: &LimitExec,
+) -> Arc<dyn ExecutionPlan> {
+ match (parent_limit_exec, child_limit_exec) {
+ (LimitExec::Local(parent), LimitExec::Local(child)) => {
+ // Simply use the smaller fetch
+ Arc::new(LocalLimitExec::new(
+ child_limit_exec.input().clone(),
+ parent.fetch().min(child.fetch()),
+ ))
+ }
+ _ => {
+ // We can use the same logic as push_down_limit from logical plan optimizer
+ let (skip, fetch) = combine_limit(
+ parent_limit_exec.skip(),
+ parent_limit_exec.fetch(),
+ child_limit_exec.skip(),
+ child_limit_exec.fetch(),
+ );
+
+ Arc::new(GlobalLimitExec::new(
+ child_limit_exec.input().clone(),
+ skip,
+ fetch,
+ ))
}
- } else {
- // Parent's fetch is infinite, use child's
- Ok(limit_exec.with_fetch(child_fetch).into())
}
}
+/// Combines two limits into a single
+///
+/// Returns the combined limit `(skip, fetch)`
+///
+/// # Case 0: Parent and Child are disjoint. (`child_fetch <= skip`)
+///
+/// ```text
+/// Before merging:
+/// |........skip........|---fetch-->| Parent Limit
+/// |...child_skip...|---child_fetch-->| Child Limit
+/// ```
+///
+/// After merging:
+/// ```text
+/// |.........(child_skip + skip).........|
+/// ```
+///
+/// Before merging:
+/// ```text
+/// |...skip...|------------fetch------------>| Parent Limit
+/// |...child_skip...|-------------child_fetch------------>| Child Limit
+/// ```
+///
+/// After merging:
+/// ```text
+/// |....(child_skip + skip)....|---(child_fetch - skip)-->|
+/// ```
+///
+/// # Case 1: Parent is beyond the range of Child. (`skip < child_fetch <= skip + fetch`)
+///
+/// Before merging:
+/// ```text
+/// |...skip...|------------fetch------------>| Parent Limit
+/// |...child_skip...|-------------child_fetch------------>| Child Limit
+/// ```
+///
+/// After merging:
+/// ```text
+/// |....(child_skip + skip)....|---(child_fetch - skip)-->|
+/// ```
+///
+/// # Case 2: Parent is in the range of Child. (`skip + fetch < child_fetch`)
+/// Before merging:
+/// ```text
+/// |...skip...|---fetch-->| Parent Limit
+/// |...child_skip...|-------------child_fetch------------>| Child Limit
+/// ```
+///
+/// After merging:
+/// ```text
+/// |....(child_skip + skip)....|---fetch-->|
+/// ```
Review Comment:
```suggestion
/// Computes the `skip` and `fetch` parameters of a single limit that would be
/// equivalent to two consecutive limits with the given `skip`/`fetch` parameters.
///
/// There are multiple cases to consider:
///
/// # Case 0: Parent and child are disjoint (`child_fetch <= skip`).
///
/// ```text
/// Before merging:
/// |........skip........|---fetch-->| Parent limit
/// |...child_skip...|---child_fetch-->| Child limit
/// ```
///
/// After merging:
/// ```text
/// |.........(child_skip + skip).........|
/// ```
///
/// # Case 1: Parent is beyond child's range (`skip < child_fetch <= skip + fetch`).
///
/// Before merging:
/// ```text
/// |...skip...|------------fetch------------>| Parent limit
/// |...child_skip...|-------------child_fetch------------>| Child limit
/// ```
///
/// After merging:
/// ```text
/// |....(child_skip + skip)....|---(child_fetch - skip)-->|
/// ```
///
/// # Case 2: Parent is within child's range (`skip + fetch < child_fetch`).
///
/// Before merging:
/// ```text
/// |...skip...|---fetch-->| Parent limit
/// |...child_skip...|-------------child_fetch------------>| Child limit
/// ```
///
/// After merging:
/// ```text
/// |....(child_skip + skip)....|---fetch-->|
/// ```
```
--
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] Add LimitPushdown optimization rule and CoalesceBatchesExec fetch [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #11652:
URL: https://github.com/apache/datafusion/pull/11652#discussion_r1693886004
##########
datafusion/sqllogictest/test_files/group_by.slt:
##########
@@ -4334,8 +4335,9 @@ physical_plan
01)GlobalLimitExec: skip=0, fetch=5
02)--SortPreservingMergeExec: [name@0 DESC,time_chunks@1 DESC], fetch=5
03)----ProjectionExec: expr=[name@0 as name, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 900000000000 }, ts@1) as time_chunks]
-04)------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1
-05)--------StreamingTableExec: partition_sizes=1, projection=[name, ts], infinite_source=true, output_ordering=[name@0 DESC, ts@1 DESC]
+04)------LocalLimitExec: fetch=5
+05)--------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1
Review Comment:
I think we haven't covered `RepartitionExec` yet, would be a good idea to do it in a follow-on 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] Add LimitPushdown optimization rule and CoalesceBatchesExec fetch [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak merged PR #11652:
URL: https://github.com/apache/datafusion/pull/11652
--
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