You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "berkaysynnada (via GitHub)" <gi...@apache.org> on 2024/08/26 13:24:57 UTC
[PR] TopK Should Update ExecutionMode [datafusion]
berkaysynnada opened a new pull request, #12174:
URL: https://github.com/apache/datafusion/pull/12174
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
Closes #.
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
Adding a fetch limit to a `SortExec` should also update the execution mode in compute properties. The mode of Top-K operator can be changed as `ExecutionMode::Bounded` if the requirements and orders are matched.
## What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
1) Add the conditional mode update part in `with_fetch()` methods.
2) There are 2 `with_fetch()` methods; one in `impl SortExec` and `impl ExecutionPlan for SortExec`. The latter calls the first one now.
3) A minor code clean-up of `evaluate_partition_by_column_values` is included in this PR also.
## Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
## Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api change` label.
-->
--
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] Unbounded SortExec (and Top-K) Implementation When Req's Are Satisfied [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #12174:
URL: https://github.com/apache/datafusion/pull/12174#issuecomment-2323319078
Thank you for responding to the feedback @berkaysynnada and @ozankabak -- sorry for my delay -- I have been out. This PR now looks good 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 Should Update ExecutionMode [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on code in PR #12174:
URL: https://github.com/apache/datafusion/pull/12174#discussion_r1732718351
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -737,9 +738,30 @@ impl SortExec {
/// This can reduce the memory pressure required by the sort
/// operation since rows that are not going to be included
/// can be dropped.
- pub fn with_fetch(mut self, fetch: Option<usize>) -> Self {
- self.fetch = fetch;
- self
+ pub fn with_fetch(&self, fetch: Option<usize>) -> Self {
+ let mut cache = self.cache.clone();
+ if fetch.is_some() {
+ // When a theoretically unnecessary sort becomes a top-K (which
Review Comment:
> I don't fully understand how a top-k sort would become bounded. I may misundersrtand what the [ExecutionMode](https://docs.rs/datafusion/latest/datafusion/physical_plan/enum.ExecutionMode.html) trait means, but it seems like `TopK` could not complete until its input completed, but if its input was unbounded the sort itself therefore would also be unbounded
I misassumed the implementation of top-k. Could you please take a look to the new 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] Unbounded SortExec (and Top-K) Implementation When Req's Are Satisfied [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #12174:
URL: https://github.com/apache/datafusion/pull/12174#discussion_r1740095136
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -737,9 +738,30 @@ impl SortExec {
/// This can reduce the memory pressure required by the sort
/// operation since rows that are not going to be included
/// can be dropped.
- pub fn with_fetch(mut self, fetch: Option<usize>) -> Self {
- self.fetch = fetch;
- self
+ pub fn with_fetch(&self, fetch: Option<usize>) -> Self {
+ let mut cache = self.cache.clone();
+ if fetch.is_some() {
+ // When a theoretically unnecessary sort becomes a top-K (which
Review Comment:
FWIW I think the comments help explain what is going on here well. Thank you
--
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] Unbounded SortExec (and Top-K) Implementation When Req's Are Satisfied [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on code in PR #12174:
URL: https://github.com/apache/datafusion/pull/12174#discussion_r1735654567
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -874,53 +892,67 @@ impl ExecutionPlan for SortExec {
trace!("End SortExec's input.execute for partition: {}", partition);
- if let Some(fetch) = self.fetch.as_ref() {
- let mut topk = TopK::try_new(
- partition,
- input.schema(),
- self.expr.clone(),
- *fetch,
- context.session_config().batch_size(),
- context.runtime_env(),
- &self.metrics_set,
- partition,
- )?;
-
- Ok(Box::pin(RecordBatchStreamAdapter::new(
- self.schema(),
- futures::stream::once(async move {
- while let Some(batch) = input.next().await {
- let batch = batch?;
- topk.insert_batch(batch)?;
- }
- topk.emit()
- })
- .try_flatten(),
- )))
- } else {
- let mut sorter = ExternalSorter::new(
- partition,
- input.schema(),
- self.expr.clone(),
- context.session_config().batch_size(),
- self.fetch,
- execution_options.sort_spill_reservation_bytes,
- execution_options.sort_in_place_threshold_bytes,
- &self.metrics_set,
- context.runtime_env(),
+ let sort_satisfied = self
Review Comment:
I have tried that but didn't work. Knowing the sort is bounded or unbounded does not mean sort is satisfied.
--
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] Unbounded SortExec (and Top-K) Implementation When Req's Are Satisfied [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #12174:
URL: https://github.com/apache/datafusion/pull/12174#discussion_r1735090072
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -933,24 +965,59 @@ impl ExecutionPlan for SortExec {
}
fn with_fetch(&self, limit: Option<usize>) -> Option<Arc<dyn ExecutionPlan>> {
- Some(Arc::new(SortExec {
- input: Arc::clone(&self.input),
- expr: self.expr.clone(),
- metrics_set: self.metrics_set.clone(),
- preserve_partitioning: self.preserve_partitioning,
- fetch: limit,
- cache: self.cache.clone(),
- }))
+ Some(Arc::new(SortExec::with_fetch(self, limit)))
}
fn fetch(&self) -> Option<usize> {
self.fetch
}
}
+struct TopKStream {
+ input: SendableRecordBatchStream,
+ schema: SchemaRef,
+ fetch: usize,
+}
+
+impl Stream for TopKStream {
Review Comment:
You are right -- sent a commit to reuse `LimitStream`
--
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] Unbounded SortExec (and Top-K) Implementation When Req's Are Satisfied [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on PR #12174:
URL: https://github.com/apache/datafusion/pull/12174#issuecomment-2315954892
Thanks for taking another look -- incorporated your feedback to leverage `LimitStream`
--
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 Should Update ExecutionMode [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #12174:
URL: https://github.com/apache/datafusion/pull/12174#discussion_r1732297413
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -737,9 +738,30 @@ impl SortExec {
/// This can reduce the memory pressure required by the sort
/// operation since rows that are not going to be included
/// can be dropped.
- pub fn with_fetch(mut self, fetch: Option<usize>) -> Self {
- self.fetch = fetch;
- self
+ pub fn with_fetch(&self, fetch: Option<usize>) -> Self {
+ let mut cache = self.cache.clone();
+ if fetch.is_some() {
+ // When a theoretically unnecessary sort becomes a top-K (which
Review Comment:
In general you are right -- however, you are missing that we turn it into `Bounded` only when the sort requirement is already satisfied. This happens when a sort "becomes" unnecessary during one of the plan optimization steps (and it will eventually get removed).
--
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] Unbounded SortExec (and Top-K) Implementation When Req's Are Satisfied [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #12174:
URL: https://github.com/apache/datafusion/pull/12174#discussion_r1734880197
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -965,12 +1032,124 @@ mod tests {
use arrow::compute::SortOptions;
use arrow::datatypes::*;
use datafusion_common::cast::as_primitive_array;
+ use datafusion_common::{assert_batches_eq, Result, ScalarValue};
use datafusion_execution::config::SessionConfig;
- use datafusion_execution::runtime_env::RuntimeEnvBuilder;
+ use datafusion_execution::runtime_env::{RuntimeConfig, RuntimeEnvBuilder};
Review Comment:
```suggestion
use datafusion_execution::runtime_env::{RuntimeEnvBuilder};
```
--
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] Unbounded SortExec (and Top-K) Implementation When Req's Are Satisfied [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #12174:
URL: https://github.com/apache/datafusion/pull/12174#discussion_r1734614514
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -933,24 +965,59 @@ impl ExecutionPlan for SortExec {
}
fn with_fetch(&self, limit: Option<usize>) -> Option<Arc<dyn ExecutionPlan>> {
- Some(Arc::new(SortExec {
- input: Arc::clone(&self.input),
- expr: self.expr.clone(),
- metrics_set: self.metrics_set.clone(),
- preserve_partitioning: self.preserve_partitioning,
- fetch: limit,
- cache: self.cache.clone(),
- }))
+ Some(Arc::new(SortExec::with_fetch(self, limit)))
}
fn fetch(&self) -> Option<usize> {
self.fetch
}
}
+struct TopKStream {
+ input: SendableRecordBatchStream,
+ schema: SchemaRef,
+ fetch: usize,
+}
+
+impl Stream for TopKStream {
Review Comment:
This looks very similar to `LimitStream` -- https://docs.rs/datafusion-physical-plan/41.0.0/src/datafusion_physical_plan/limit.rs.html#434
though limit stream has metrics and some other features
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -874,53 +892,67 @@ impl ExecutionPlan for SortExec {
trace!("End SortExec's input.execute for partition: {}", partition);
- if let Some(fetch) = self.fetch.as_ref() {
- let mut topk = TopK::try_new(
- partition,
- input.schema(),
- self.expr.clone(),
- *fetch,
- context.session_config().batch_size(),
- context.runtime_env(),
- &self.metrics_set,
- partition,
- )?;
-
- Ok(Box::pin(RecordBatchStreamAdapter::new(
- self.schema(),
- futures::stream::once(async move {
- while let Some(batch) = input.next().await {
- let batch = batch?;
- topk.insert_batch(batch)?;
- }
- topk.emit()
- })
- .try_flatten(),
- )))
- } else {
- let mut sorter = ExternalSorter::new(
- partition,
- input.schema(),
- self.expr.clone(),
- context.session_config().batch_size(),
- self.fetch,
- execution_options.sort_spill_reservation_bytes,
- execution_options.sort_in_place_threshold_bytes,
- &self.metrics_set,
- context.runtime_env(),
+ let sort_satisfied = self
Review Comment:
This is the same calculation as `self.execution_mode()`, right? Maybe we could call `self.execution_mode` here instead to be more efficient and ensure the calculations remained in sync
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -933,24 +965,59 @@ impl ExecutionPlan for SortExec {
}
fn with_fetch(&self, limit: Option<usize>) -> Option<Arc<dyn ExecutionPlan>> {
- Some(Arc::new(SortExec {
- input: Arc::clone(&self.input),
- expr: self.expr.clone(),
- metrics_set: self.metrics_set.clone(),
- preserve_partitioning: self.preserve_partitioning,
- fetch: limit,
- cache: self.cache.clone(),
- }))
+ Some(Arc::new(SortExec::with_fetch(self, limit)))
}
fn fetch(&self) -> Option<usize> {
self.fetch
}
}
+struct TopKStream {
Review Comment:
I think it would help to add documentation to this struct, specifically that explains how it is different than `TopK`
--
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] Unbounded SortExec (and Top-K) Implementation When Req's Are Satisfied [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #12174:
URL: https://github.com/apache/datafusion/pull/12174#discussion_r1734880844
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -965,12 +1032,124 @@ mod tests {
use arrow::compute::SortOptions;
use arrow::datatypes::*;
use datafusion_common::cast::as_primitive_array;
+ use datafusion_common::{assert_batches_eq, Result, ScalarValue};
use datafusion_execution::config::SessionConfig;
- use datafusion_execution::runtime_env::RuntimeEnvBuilder;
+ use datafusion_execution::runtime_env::{RuntimeEnvBuilder};
Review Comment:
```suggestion
use datafusion_execution::runtime_env::RuntimeEnvBuilder;
```
--
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] Unbounded SortExec (and Top-K) Implementation When Req's Are Satisfied [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #12174:
URL: https://github.com/apache/datafusion/pull/12174#discussion_r1734219197
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -775,6 +791,19 @@ impl SortExec {
sort_exprs: LexOrdering,
preserve_partitioning: bool,
) -> PlanProperties {
+ // Determine execution mode:
+ let input_satisfies_sort_req =
+ input.equivalence_properties().ordering_satisfy_requirement(
+ PhysicalSortRequirement::from_sort_exprs(sort_exprs.iter()).as_slice(),
+ );
+ let mode = match input.execution_mode() {
+ ExecutionMode::Unbounded if input_satisfies_sort_req => {
+ ExecutionMode::Unbounded
+ }
+ ExecutionMode::Bounded => ExecutionMode::Bounded,
+ _ => ExecutionMode::PipelineBreaking,
+ };
Review Comment:
```suggestion
let sort_satisfied =
input.equivalence_properties().ordering_satisfy_requirement(
PhysicalSortRequirement::from_sort_exprs(sort_exprs.iter()).as_slice(),
);
let mode = match input.execution_mode() {
ExecutionMode::Unbounded if sort_satisfied => {
ExecutionMode::Unbounded
}
ExecutionMode::Bounded => ExecutionMode::Bounded,
_ => ExecutionMode::PipelineBreaking,
};
```
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -965,12 +1035,125 @@ mod tests {
use arrow::compute::SortOptions;
use arrow::datatypes::*;
use datafusion_common::cast::as_primitive_array;
+ use datafusion_common::Result;
+ use datafusion_common::{assert_batches_eq, ScalarValue};
Review Comment:
```suggestion
use datafusion_common::{assert_batches_eq, Result, ScalarValue};
```
--
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 Should Update ExecutionMode [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on code in PR #12174:
URL: https://github.com/apache/datafusion/pull/12174#discussion_r1732718351
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -737,9 +738,30 @@ impl SortExec {
/// This can reduce the memory pressure required by the sort
/// operation since rows that are not going to be included
/// can be dropped.
- pub fn with_fetch(mut self, fetch: Option<usize>) -> Self {
- self.fetch = fetch;
- self
+ pub fn with_fetch(&self, fetch: Option<usize>) -> Self {
+ let mut cache = self.cache.clone();
+ if fetch.is_some() {
+ // When a theoretically unnecessary sort becomes a top-K (which
Review Comment:
> I don't fully understand how a top-k sort would become bounded. I may misundersrtand what the [ExecutionMode](https://docs.rs/datafusion/latest/datafusion/physical_plan/enum.ExecutionMode.html) trait means, but it seems like `TopK` could not complete until its input completed, but if its input was unbounded the sort itself therefore would also be unbounded
I misassumed the implementation of top-k. Could you please take a look to the new idea? I will update the PR title and body
--
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] Unbounded SortExec (and Top-K) Implementation When Req's Are Satisfied [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on PR #12174:
URL: https://github.com/apache/datafusion/pull/12174#issuecomment-2316731429
I will go ahead and merge this soon since it is a small, localized change that fixes a wrong return value to the `execution_mode` API when the input of a sort is already sorted.
If there are any lingering concerns, we will address with a quick 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] Unbounded SortExec (and Top-K) Implementation When Req's Are Satisfied [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on PR #12174:
URL: https://github.com/apache/datafusion/pull/12174#issuecomment-2315253224
> But in this case I would have expected the Sort to have been removed by one of the optimizer passes rather than the SortExec implementing a limit. 🤔
Right. The sort will actually be entirely removed in later passes. However, in the meantime (before its removal), previous rules still think this TopK is pipeline-breaking (while it is not), which results in them behaving incorrectly.
Basically, this PR does two things:
1. It makes the `execution_mode` API give a correct result when data is already sorted, so intermediate rules that look at execution modes do not falsely think that the pipeline breaks somewhere.
2. It makes TopK's execution logic behave like a limit if it somehow ends up in the plan without being removed. This is not possible in upstream DataFusion (sans bugs), but people using custom optimizers can end up in such a state. In such cases, it will also execute more efficiently.
--
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] Unbounded SortExec (and Top-K) Implementation When Req's Are Satisfied [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #12174:
URL: https://github.com/apache/datafusion/pull/12174#discussion_r1734633225
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -933,24 +965,59 @@ impl ExecutionPlan for SortExec {
}
fn with_fetch(&self, limit: Option<usize>) -> Option<Arc<dyn ExecutionPlan>> {
- Some(Arc::new(SortExec {
- input: Arc::clone(&self.input),
- expr: self.expr.clone(),
- metrics_set: self.metrics_set.clone(),
- preserve_partitioning: self.preserve_partitioning,
- fetch: limit,
- cache: self.cache.clone(),
- }))
+ Some(Arc::new(SortExec::with_fetch(self, limit)))
}
fn fetch(&self) -> Option<usize> {
self.fetch
}
}
+struct TopKStream {
+ input: SendableRecordBatchStream,
+ schema: SchemaRef,
+ fetch: usize,
+}
+
+impl Stream for TopKStream {
Review Comment:
Yes, maybe we can reuse it. We will take a look
--
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] Unbounded SortExec (and Top-K) Implementation When Req's Are Satisfied [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak commented on code in PR #12174:
URL: https://github.com/apache/datafusion/pull/12174#discussion_r1735089341
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -874,53 +892,67 @@ impl ExecutionPlan for SortExec {
trace!("End SortExec's input.execute for partition: {}", partition);
- if let Some(fetch) = self.fetch.as_ref() {
- let mut topk = TopK::try_new(
- partition,
- input.schema(),
- self.expr.clone(),
- *fetch,
- context.session_config().batch_size(),
- context.runtime_env(),
- &self.metrics_set,
- partition,
- )?;
-
- Ok(Box::pin(RecordBatchStreamAdapter::new(
- self.schema(),
- futures::stream::once(async move {
- while let Some(batch) = input.next().await {
- let batch = batch?;
- topk.insert_batch(batch)?;
- }
- topk.emit()
- })
- .try_flatten(),
- )))
- } else {
- let mut sorter = ExternalSorter::new(
- partition,
- input.schema(),
- self.expr.clone(),
- context.session_config().batch_size(),
- self.fetch,
- execution_options.sort_spill_reservation_bytes,
- execution_options.sort_in_place_threshold_bytes,
- &self.metrics_set,
- context.runtime_env(),
+ let sort_satisfied = self
Review Comment:
It is similar but not exactly the same (i.e. execution mode is derived from `sort_satisfied` but AFAIK the reverse is not possible). I think @berkaysynnada tried this but it didn't work
--
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 Should Update ExecutionMode [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #12174:
URL: https://github.com/apache/datafusion/pull/12174#discussion_r1731708046
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -737,9 +738,30 @@ impl SortExec {
/// This can reduce the memory pressure required by the sort
/// operation since rows that are not going to be included
/// can be dropped.
- pub fn with_fetch(mut self, fetch: Option<usize>) -> Self {
- self.fetch = fetch;
- self
+ pub fn with_fetch(&self, fetch: Option<usize>) -> Self {
+ let mut cache = self.cache.clone();
+ if fetch.is_some() {
+ // When a theoretically unnecessary sort becomes a top-K (which
Review Comment:
I don't fully understand how a top-k sort would become bounded. I may misundersrtand what the [ExecutionMode](https://docs.rs/datafusion/latest/datafusion/physical_plan/enum.ExecutionMode.html) trait means, but it seems like `TopK` could not complete until its input completed, but if its input was unbounded the sort itself therefore would also be unbounded
--
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] Unbounded SortExec (and Top-K) Implementation When Req's Are Satisfied [datafusion]
Posted by "ozankabak (via GitHub)" <gi...@apache.org>.
ozankabak merged PR #12174:
URL: https://github.com/apache/datafusion/pull/12174
--
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