You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "zhuqi-lucas (via GitHub)" <gi...@apache.org> on 2025/11/19 12:12:11 UTC
[PR] Draft Reverse parquet [datafusion]
zhuqi-lucas opened a new pull request, #18817:
URL: https://github.com/apache/datafusion/pull/18817
## 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.
-->
## 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.
-->
## 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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3569439710
> Thank you @2010YOUY01 for review and valid concern:
You raise valid concerns about memory overhead is what i mentioned the key risk for this approach.
However, I want to clarify that row group reversal alone cannot eliminate the SortExec - it only provides TopK filtering benefits. Without reversing rows within each row group, the data remains in the original order (e.g., ASC when we need DESC), so the sort must stay. I propose we keep the complete optimization but default enable_reverse_scan to false. Once we implement page-level caching in arrow-rs (which will reduce memory overhead significantly), we can consider enabling it by default.
Did you mean 'cannot eliminate the SortExec(TopK)'? Just to confirm there is no global sort, but it is true that we have do a `topK` on a whole row group for this naive approach.
I have a intuition that for this kind of workload, the bottleneck is on the parquet decoding speed, and an extra `TopK` won't introduce much additional overhead, so this naive approach can also get pretty fast.
It makes a lot of sense that it's very hard to implement page/row level reversal in `arrow-rs` side, so we have to figure out how to do this at row-group level.
Summary: Perhaps we can start by adding a few end-to-end benchmarks that reflect your typical production workload. If this PRโs approach shows a clear improvement over the naive approach in https://github.com/apache/datafusion/pull/18817#issuecomment-3568934764 (I'm happy to do a quick prototype), we should definitely move forward.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3600642871
Please don't hold this up for me (I'm on vacation until Thursday) but I would love to review this, it's very exciting 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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3594266375
> > run benchmarks
>
> (BTW I am testing out some new scripts: [#18115 (comment)](https://github.com/apache/datafusion/issues/18115#issuecomment-3592609246))
>
> @zhuqi-lucas / @xudong963 let me know if you would like to be added to the whitelist
Amazing @alamb , pretty cool!
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3569581329
> FYI, I'll start reviewing the PR tomorrow.
Thanks @xudong963 !
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2569295613
##########
datafusion/physical-optimizer/src/pushdown_sort.rs:
##########
@@ -0,0 +1,1235 @@
+// 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.
+
+//! Sort Pushdown Optimization
+//!
+//! This optimizer attempts to push sort requirements down to data sources that can
+//! satisfy them natively, avoiding expensive sort operations.
+//!
+//! Currently supported optimizations:
+//! - **Reverse scan**: If a data source naturally produces data in DESC order and
+//! we need ASC (or vice versa), we can reverse the scan direction instead of
+//! adding a SortExec node.
+//!
+//! Future optimizations could include:
+//! - Reordering row groups in Parquet files
+//! - Leveraging native indexes
+//! - Reordering files in multi-file scans
+use crate::PhysicalOptimizerRule;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion_common::Result;
+use datafusion_datasource::source::DataSourceExec;
+use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
+use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
+use datafusion_physical_plan::sorts::sort::SortExec;
+use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+use std::sync::Arc;
+
+/// A PhysicalOptimizerRule that attempts to push down sort requirements to data sources
+/// that can natively handle them (e.g., by reversing scan direction).
+///
+/// This optimization:
+/// 1. Detects SortExec nodes that require a specific ordering
+/// 2. Checks if the input can satisfy the ordering by reversing its scan direction
+/// 3. Pushes the sort requirement down to the data source when possible
+/// 4. Removes unnecessary sort operations when the input already satisfies the requirement
+#[derive(Debug, Clone, Default)]
+pub struct PushdownSort;
+
+impl PushdownSort {
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for PushdownSort {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ // Check if sort pushdown optimization is enabled
+ let enable_sort_pushdown = config.execution.parquet.enable_sort_pushdown;
+
+ // Return early if not enabled
+ if !enable_sort_pushdown {
+ return Ok(plan);
+ }
+
+ // Search for any SortExec nodes and try to optimize them
+ plan.transform_up(&|plan: Arc<dyn ExecutionPlan>| {
+ // First check if this is a GlobalLimitExec -> SortExec pattern
+ if let Some(limit_exec) = plan.as_any().downcast_ref::<GlobalLimitExec>() {
+ if let Some(sort_exec) =
+ limit_exec.input().as_any().downcast_ref::<SortExec>()
+ {
+ return optimize_limit_sort(limit_exec, sort_exec);
+ }
+ }
+
+ // Otherwise, check if this is just a SortExec
+ let sort_exec = match plan.as_any().downcast_ref::<SortExec>() {
+ Some(sort_exec) => sort_exec,
+ None => return Ok(Transformed::no(plan)),
+ };
+
+ optimize_sort(sort_exec)
+ })
+ .data()
+ }
+
+ fn name(&self) -> &str {
+ "PushdownSort"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// Optimize a SortExec by potentially pushing the sort down to the data source
+fn optimize_sort(sort_exec: &SortExec) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+ let required_ordering = sort_exec.expr();
+
+ // First, check if the sort is already satisfied by input ordering
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+
+ if input_eq_properties.ordering_satisfy(required_ordering.clone())? {
+ return remove_unnecessary_sort(sort_exec, sort_input);
+ }
+ }
+
+ // Try to push the sort requirement down to the data source
+ if let Some(optimized_input) = try_pushdown_sort(&sort_input, required_ordering)? {
+ // Verify that the optimized input satisfies the required ordering
+ if optimized_input
+ .equivalence_properties()
+ .ordering_satisfy(required_ordering.clone())?
+ {
+ return remove_unnecessary_sort(sort_exec, optimized_input);
+ }
+
+ // If not fully satisfied, keep the sort but with optimized input
+ return Ok(Transformed::yes(Arc::new(
+ SortExec::new(required_ordering.clone(), optimized_input)
+ .with_fetch(sort_exec.fetch())
+ .with_preserve_partitioning(sort_exec.preserve_partitioning()),
+ )));
+ }
+
+ Ok(Transformed::no(Arc::new(sort_exec.clone())))
+}
+
+/// Handle the GlobalLimitExec -> SortExec pattern
+fn optimize_limit_sort(
+ limit_exec: &GlobalLimitExec,
+ sort_exec: &SortExec,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+ let required_ordering = sort_exec.expr();
+
+ // Check if input is already sorted
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+ if input_eq_properties.ordering_satisfy(required_ordering.clone())? {
+ // Input is already sorted correctly, remove sort and keep limit
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ sort_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Try to push down the sort requirement
+ if let Some(optimized_input) = try_pushdown_sort(&sort_input, required_ordering)? {
+ if optimized_input
+ .equivalence_properties()
+ .ordering_satisfy(required_ordering.clone())?
+ {
+ // Successfully pushed down sort, now handle the limit
+ let total_fetch = limit_exec.skip() + limit_exec.fetch().unwrap_or(0);
+
+ // Try to push limit down as well if the source supports it
+ if let Some(with_fetch) = optimized_input.with_fetch(Some(total_fetch)) {
+ if limit_exec.skip() > 0 {
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ with_fetch,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ } else {
+ return Ok(Transformed::yes(with_fetch));
+ }
+ }
+
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ optimized_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Can't optimize, return original pattern
+ Ok(Transformed::no(Arc::new(GlobalLimitExec::new(
+ Arc::new(sort_exec.clone()),
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))))
+}
+
+/// Remove unnecessary sort based on the logic from EnforceSorting::analyze_immediate_sort_removal
+fn remove_unnecessary_sort(
+ sort_exec: &SortExec,
+ sort_input: Arc<dyn ExecutionPlan>,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let new_plan = if !sort_exec.preserve_partitioning()
+ && sort_input.output_partitioning().partition_count() > 1
+ {
+ // Replace the sort with a sort-preserving merge
+ Arc::new(
+ SortPreservingMergeExec::new(sort_exec.expr().clone(), sort_input)
+ .with_fetch(sort_exec.fetch()),
+ ) as _
+ } else {
+ // Remove the sort entirely
+ if let Some(fetch) = sort_exec.fetch() {
+ // If the sort has a fetch, add a limit instead
+ if sort_input.output_partitioning().partition_count() == 1 {
+ // Try to push the limit down to the source
+ if let Some(with_fetch) = sort_input.with_fetch(Some(fetch)) {
+ return Ok(Transformed::yes(with_fetch));
+ }
+ Arc::new(GlobalLimitExec::new(sort_input, 0, Some(fetch)))
+ as Arc<dyn ExecutionPlan>
+ } else {
+ Arc::new(LocalLimitExec::new(sort_input, fetch)) as Arc<dyn ExecutionPlan>
+ }
+ } else {
+ sort_input
+ }
+ };
+
+ Ok(Transformed::yes(new_plan))
+}
+
+/// Try to push down a sort requirement to an execution plan
+fn try_pushdown_sort(
+ plan: &Arc<dyn ExecutionPlan>,
+ required_ordering: &[PhysicalSortExpr],
+) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ // Check if the plan can natively handle the sort requirement
+ if let Some(data_source_exec) = plan.as_any().downcast_ref::<DataSourceExec>() {
+ return data_source_exec.try_pushdown_sort(required_ordering);
+ }
+
+ Ok(None)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow::array::{Int32Array, RecordBatch};
+ use arrow::compute::SortOptions;
+ use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+ use datafusion_physical_expr::expressions::Column;
+ use datafusion_physical_expr::Partitioning;
+ use datafusion_physical_expr_common::sort_expr::LexOrdering;
+ use datafusion_physical_plan::empty::EmptyExec;
+ use datafusion_physical_plan::memory::LazyMemoryExec;
+ use datafusion_physical_plan::repartition::RepartitionExec;
+ use parking_lot::RwLock;
+ use std::fmt;
+
+ /// Helper function to create a test schema with three columns
+ fn create_test_schema() -> Arc<Schema> {
+ Arc::new(Schema::new(vec![
+ Field::new("col1", DataType::Int32, true),
+ Field::new("col2", DataType::Utf8, true),
+ Field::new("col3", DataType::Float64, true),
+ ]))
+ }
+
+ /// Test batch generator that produces sorted data
+ #[derive(Debug, Clone)]
+ struct SortedBatchGenerator {
+ schema: SchemaRef,
+ batches_generated: usize,
+ max_batches: usize,
+ ascending: bool,
+ }
+
+ impl fmt::Display for SortedBatchGenerator {
+ fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+ write!(
+ f,
+ "SortedBatchGenerator(batches_generated={}, max_batches={}, ascending={})",
+ self.batches_generated, self.max_batches, self.ascending
+ )
+ }
+ }
+
+ impl datafusion_physical_plan::memory::LazyBatchGenerator for SortedBatchGenerator {
Review Comment:
Thank you @alamb for good suggestion, i addressed this in latest 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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3600746535
> Please don't hold this up for me (I'm on vacation until Thursday) but I would love to review this, it's very exciting work ๐
Thank you @adriangb , i am working on the benchmark first.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2581153419
##########
datafusion/datasource-parquet/src/opener.rs:
##########
@@ -752,10 +799,316 @@ fn should_enable_page_index(
.unwrap_or(false)
}
+fn reverse_batch(batch: RecordBatch) -> Result<RecordBatch> {
+ let num_rows = batch.num_rows();
+ if num_rows <= 1 {
+ return Ok(batch);
+ }
+
+ let indices = UInt32Array::from_iter_values((0..num_rows as u32).rev());
+
+ let reversed_columns = batch
+ .columns()
+ .iter()
+ .map(|col| take(col.as_ref(), &indices, None))
+ .collect::<std::result::Result<Vec<ArrayRef>, arrow::error::ArrowError>>()
+ .map_err(DataFusionError::from)?;
+
+ RecordBatch::try_new(batch.schema(), reversed_columns).map_err(DataFusionError::from)
+}
+
+/// Stream adapter for reversed parquet reading with row-group-level buffering.
+///
+/// # Architecture
+///
+/// This stream implements a sophisticated buffering strategy to achieve true reverse
+/// reading of Parquet files while maintaining compatibility with the underlying
+/// ParquetRecordBatchStream's optimizations (caching, prefetching, etc.).
+///
+/// ## Strategy Overview
+///
+/// 1. **Pre-reversed Row Groups**: Row groups are reversed BEFORE building the stream
+/// (via `row_group_indexes.reverse()`). This allows the Parquet reader to read
+/// them in reverse order while still utilizing internal optimizations.
+///
+/// 2. **Row-Group-Level Buffering**: As batches arrive from the input stream, we
+/// track which row group they belong to using cumulative row counts. This is
+/// the MINIMAL buffering unit required for correctness - we cannot reverse
+/// individual batches without knowing the complete row group context.
+///
+/// 3. **Two-Stage Reversal**: When a complete row group is collected:
+/// - Stage 1: Reverse rows within each batch (using Arrow's take kernel)
+/// - Stage 2: Reverse the order of batches within the row group
+///
+/// 4. **Progressive Output**: Reversed batches are output immediately, minimizing
+/// memory footprint. We never buffer more than one row group at a time.
+///
+/// ## Memory Characteristics
+///
+/// - **Bounded Memory**: Maximum memory usage = size of largest row group
+/// - **Typical Usage**: ~128MB (default Parquet row group size)
+/// - **Peak Usage**: During reversal of a single row group
+///
+/// ## Why Row-Group-Level Buffering is Necessary
+///
+/// Parquet organizes data into row groups (typically 128MB each), and each row group
+/// is independently compressed and encoded. When reading in reverse:
+///
+/// - We cannot reverse individual batches in isolation because they may span
+/// row group boundaries or be split arbitrarily by the batch_size parameter
+/// - We must buffer complete row groups to ensure correct ordering semantics
+/// - This is the minimum granularity that maintains correctness
+///
+/// ## Example
+///
+/// Given a file with 3 row groups, each containing 2 batches:
+///
+/// ```text
+/// Normal order: RG0[B0, B1] -> RG1[B0, B1] -> RG2[B0, B1]
+/// Reversed: RG2[B1_rev, B0_rev] -> RG1[B1_rev, B0_rev] -> RG0[B1_rev, B0_rev]
+/// ^^^^^^^^^^^^ ^^^^^^^^^^^^ ^^^^^^^^^^^^
+/// Output 1st Output 2nd Output 3rd
+/// ```
+///
+/// ## Performance Characteristics
+///
+/// - **Latency**: First batch available after reading complete first (reversed) row group
+/// - **Throughput**: Near-native speed with ~5-10% overhead for reversal operations
+/// - **Memory**: O(row_group_size), not O(file_size)
Review Comment:
It means, the memory max size is max row_group memory size, not the file_size.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3566729760
Thank you @xudong963 @suremarc, i do a lot of changes comparing our internal implementation in this PR, but i think in general the major design is similar to our internal version, the row group level reverse. But need to add more follow-up PRs to improve it further, for example, we should support customer output source, so that we can integrated it with ordered_partiton source, 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
Re: [PR] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3567876427
> Thanks -- I'll try and review this tomorrow
Thank you @alamb !
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2554059927
##########
datafusion/sqllogictest/test_files/topk.slt:
##########
@@ -338,9 +338,7 @@ physical_plan
query TT
explain select number, letter, age from partial_sorted order by number asc limit 3;
----
-physical_plan
-01)SortExec: TopK(fetch=3), expr=[number@0 ASC NULLS LAST], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, letter, age], output_ordering=[number@0 DESC, letter@1 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ]
Review Comment:
Reverse scan make it don't need to sort here, and it's very fast.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2570783569
##########
datafusion/physical-plan/src/execution_plan.rs:
##########
@@ -682,6 +684,16 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
) -> Option<Arc<dyn ExecutionPlan>> {
None
}
+
+ /// Try to create a new execution plan that satisfies the given sort ordering.
+ ///
+ /// Default implementation returns `Ok(None)`.
+ fn try_pushdown_sort(
Review Comment:
I support high level pushdown in optimizer now in latest PR, and remove datasource in it, so we use ExecutionPlan pushdown now, and the datasourceexec is one of the implementation of it, all other exec which can support pushdown sort can benefit it in future.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3569088218
I haven't looked into all of this discussion and code (I just got tagged). I've been looking into optimizing sorted scanning in DataFusion and IMO where we should land is:
1. Via metadata (`FileScanConfig` / `ORDERED BY ...` in SQL) users declare a known sort order of their files.
2. The planner uses statistics from the files + any `ORDER BY` clauses in the query to arrange file ordering to best match the query. The `FileSource` implementation can also receive the `ORDER BY` information and optimize scan order within a file (e.g. reversing the order of reads which is what I think this PR is doing).
3. If the planner is able to deduce from file level stats that the files can be ordered and the `FileSource` reports that it is able to produce batches in sorted order then the optimizer can optimize away the sort completely.
I hope that is helpful.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3592610426
run benchmarks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2554023909
##########
datafusion/common/src/config.rs:
##########
@@ -836,6 +836,13 @@ config_namespace! {
/// writing out already in-memory data, such as from a cached
/// data frame.
pub maximum_buffered_record_batches_per_stream: usize, default = 2
+
+ /// Enable reverse scan optimization for ORDER BY ... DESC queries
+ /// on sorted Parquet files. When enabled, row groups and batches
+ /// are read in reverse order to eliminate sort operations.
+ /// Note: This buffers one row group at a time (typically ~128MB).
+ /// Default: true
+ pub enable_reverse_scan: bool, default = true
Review Comment:
Note, i default to true for reverse optimization, we can default to false if you think it's risky for some cases.
The key risk is the memory overhead, because it's row group level reverse, so we need to cache the row group level batches, if we setting the row group max size big, it will use high memory.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3569492649
> > Thank you @2010YOUY01 for review and valid concern:
> > You raise valid concerns about memory overhead is what i mentioned the key risk for this approach.
> > However, I want to clarify that row group reversal alone cannot eliminate the SortExec - it only provides TopK filtering benefits. Without reversing rows within each row group, the data remains in the original order (e.g., ASC when we need DESC), so the sort must stay. I propose we keep the complete optimization but default enable_reverse_scan to false. Once we implement page-level caching in arrow-rs (which will reduce memory overhead significantly), we can consider enabling it by default.
>
> Did you mean 'cannot eliminate the SortExec(TopK)'? Just to confirm there is no global sort, but it is true that we have do a `topK` on a whole row group for this naive approach.
>
> I have a intuition that for this kind of workload, the bottleneck is on the parquet decoding speed, and an extra `TopK` won't introduce much additional overhead, so this naive approach can also get pretty fast.
>
> It makes a lot of sense that it's very hard to implement page/row level reversal in `arrow-rs` side, so we have to figure out how to do this at row-group level.
>
> Summary: Perhaps we can start by adding a few end-to-end benchmarks that reflect your typical production workload. If this PRโs approach shows a clear improvement over the naive approach in [#18817 (comment)](https://github.com/apache/datafusion/pull/18817#issuecomment-3568934764) (I'm happy to do a quick prototype), we should definitely move forward.
Nice point @2010YOUY01 , i agree most time will be decode page, i can change this PR to add the config to implement [#18817 (comment)](https://github.com/apache/datafusion/pull/18817#issuecomment-3568934764), so we can have more options to compare, i agree the easier solution is 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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3585471227
๐ค: Benchmark completed
<details><summary>Details</summary>
<p>
```
Comparing HEAD and reverse_parquet
--------------------
Benchmark clickbench_extended.json
--------------------
โโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโ
โ Query โ HEAD โ reverse_parquet โ Change โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ QQuery 0 โ 2668.00 ms โ 2676.04 ms โ no change โ
โ QQuery 1 โ 1213.05 ms โ 1370.19 ms โ 1.13x slower โ
โ QQuery 2 โ 2283.93 ms โ 2585.76 ms โ 1.13x slower โ
โ QQuery 3 โ 1156.72 ms โ 1152.17 ms โ no change โ
โ QQuery 4 โ 2408.82 ms โ 2314.85 ms โ no change โ
โ QQuery 5 โ 28210.63 ms โ 28406.53 ms โ no change โ
โ QQuery 6 โ 4080.93 ms โ 4150.36 ms โ no change โ
โ QQuery 7 โ 3636.28 ms โ 3856.61 ms โ 1.06x slower โ
โโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโ
โ Benchmark Summary โ โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ Total Time (HEAD) โ 45658.38ms โ
โ Total Time (reverse_parquet) โ 46512.51ms โ
โ Average Time (HEAD) โ 5707.30ms โ
โ Average Time (reverse_parquet) โ 5814.06ms โ
โ Queries Faster โ 0 โ
โ Queries Slower โ 3 โ
โ Queries with No Change โ 5 โ
โ Queries with Failure โ 0 โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโ
--------------------
Benchmark clickbench_partitioned.json
--------------------
โโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโ
โ Query โ HEAD โ reverse_parquet โ Change โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ QQuery 0 โ 2.19 ms โ 2.15 ms โ no change โ
โ QQuery 1 โ 47.84 ms โ 48.20 ms โ no change โ
โ QQuery 2 โ 134.69 ms โ 137.26 ms โ no change โ
โ QQuery 3 โ 162.61 ms โ 163.01 ms โ no change โ
โ QQuery 4 โ 1142.39 ms โ 1106.75 ms โ no change โ
โ QQuery 5 โ 1527.90 ms โ 1540.01 ms โ no change โ
โ QQuery 6 โ 2.08 ms โ 2.06 ms โ no change โ
โ QQuery 7 โ 53.27 ms โ 52.45 ms โ no change โ
โ QQuery 8 โ 1476.18 ms โ 1424.91 ms โ no change โ
โ QQuery 9 โ 1938.83 ms โ 1851.39 ms โ no change โ
โ QQuery 10 โ 373.39 ms โ 397.21 ms โ 1.06x slower โ
โ QQuery 11 โ 428.61 ms โ 450.19 ms โ 1.05x slower โ
โ QQuery 12 โ 1423.73 ms โ 1415.23 ms โ no change โ
โ QQuery 13 โ 2159.98 ms โ 2176.52 ms โ no change โ
โ QQuery 14 โ 1303.95 ms โ 1308.40 ms โ no change โ
โ QQuery 15 โ 1272.48 ms โ 1259.24 ms โ no change โ
โ QQuery 16 โ 2725.76 ms โ 2725.49 ms โ no change โ
โ QQuery 17 โ 2682.94 ms โ 2695.56 ms โ no change โ
โ QQuery 18 โ 5689.66 ms โ 5012.04 ms โ +1.14x faster โ
โ QQuery 19 โ 129.49 ms โ 126.21 ms โ no change โ
โ QQuery 20 โ 2040.96 ms โ 1990.01 ms โ no change โ
โ QQuery 21 โ 2373.70 ms โ 2310.40 ms โ no change โ
โ QQuery 22 โ 4035.25 ms โ 3967.98 ms โ no change โ
โ QQuery 23 โ 19052.17 ms โ 13237.40 ms โ +1.44x faster โ
โ QQuery 24 โ 210.64 ms โ 223.87 ms โ 1.06x slower โ
โ QQuery 25 โ 469.25 ms โ 498.34 ms โ 1.06x slower โ
โ QQuery 26 โ 222.01 ms โ 217.85 ms โ no change โ
โ QQuery 27 โ 2806.01 ms โ 2823.67 ms โ no change โ
โ QQuery 28 โ 23713.01 ms โ 23200.02 ms โ no change โ
โ QQuery 29 โ 985.92 ms โ 980.69 ms โ no change โ
โ QQuery 30 โ 1360.77 ms โ 1366.76 ms โ no change โ
โ QQuery 31 โ 1413.99 ms โ 1398.84 ms โ no change โ
โ QQuery 32 โ 4733.39 ms โ 4897.36 ms โ no change โ
โ QQuery 33 โ 5807.25 ms โ 5921.80 ms โ no change โ
โ QQuery 34 โ 6040.56 ms โ 5977.36 ms โ no change โ
โ QQuery 35 โ 1942.79 ms โ 1849.89 ms โ no change โ
โ QQuery 36 โ 121.13 ms โ 123.30 ms โ no change โ
โ QQuery 37 โ 51.95 ms โ 53.78 ms โ no change โ
โ QQuery 38 โ 119.80 ms โ 123.62 ms โ no change โ
โ QQuery 39 โ 197.43 ms โ 198.91 ms โ no change โ
โ QQuery 40 โ 42.36 ms โ 44.37 ms โ no change โ
โ QQuery 41 โ 40.68 ms โ 40.13 ms โ no change โ
โ QQuery 42 โ 31.11 ms โ 32.31 ms โ no change โ
โโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโ
โ Benchmark Summary โ โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ Total Time (HEAD) โ 102490.10ms โ
โ Total Time (reverse_parquet) โ 95372.97ms โ
โ Average Time (HEAD) โ 2383.49ms โ
โ Average Time (reverse_parquet) โ 2217.98ms โ
โ Queries Faster โ 2 โ
โ Queries Slower โ 4 โ
โ Queries with No Change โ 37 โ
โ Queries with Failure โ 0 โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโ
--------------------
Benchmark tpch_mem_sf1.json
--------------------
โโโโโโโโโโโโโโโโณโโโโโโโโโโโโณโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโ
โ Query โ HEAD โ reverse_parquet โ Change โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ QQuery 1 โ 134.69 ms โ 134.54 ms โ no change โ
โ QQuery 2 โ 27.59 ms โ 28.57 ms โ no change โ
โ QQuery 3 โ 38.77 ms โ 38.46 ms โ no change โ
โ QQuery 4 โ 28.02 ms โ 28.68 ms โ no change โ
โ QQuery 5 โ 85.40 ms โ 86.86 ms โ no change โ
โ QQuery 6 โ 19.66 ms โ 19.24 ms โ no change โ
โ QQuery 7 โ 223.45 ms โ 223.34 ms โ no change โ
โ QQuery 8 โ 34.67 ms โ 32.04 ms โ +1.08x faster โ
โ QQuery 9 โ 103.71 ms โ 105.25 ms โ no change โ
โ QQuery 10 โ 62.48 ms โ 63.12 ms โ no change โ
โ QQuery 11 โ 18.12 ms โ 17.36 ms โ no change โ
โ QQuery 12 โ 51.93 ms โ 53.29 ms โ no change โ
โ QQuery 13 โ 45.88 ms โ 46.05 ms โ no change โ
โ QQuery 14 โ 13.82 ms โ 13.99 ms โ no change โ
โ QQuery 15 โ 24.85 ms โ 24.15 ms โ no change โ
โ QQuery 16 โ 25.12 ms โ 24.79 ms โ no change โ
โ QQuery 17 โ 149.30 ms โ 148.86 ms โ no change โ
โ QQuery 18 โ 275.44 ms โ 276.76 ms โ no change โ
โ QQuery 19 โ 38.17 ms โ 39.08 ms โ no change โ
โ QQuery 20 โ 48.24 ms โ 49.63 ms โ no change โ
โ QQuery 21 โ 320.58 ms โ 315.53 ms โ no change โ
โ QQuery 22 โ 17.75 ms โ 17.58 ms โ no change โ
โโโโโโโโโโโโโโโโดโโโโโโโโโโโโดโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโ
โ Benchmark Summary โ โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ Total Time (HEAD) โ 1787.63ms โ
โ Total Time (reverse_parquet) โ 1787.17ms โ
โ Average Time (HEAD) โ 81.26ms โ
โ Average Time (reverse_parquet) โ 81.23ms โ
โ Queries Faster โ 1 โ
โ Queries Slower โ 0 โ
โ Queries with No Change โ 21 โ
โ 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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2569302094
##########
datafusion/datasource/src/file_scan_config.rs:
##########
@@ -718,6 +717,101 @@ impl DataSource for FileScanConfig {
}
}
}
+
+ fn try_pushdown_sort(
+ &self,
+ order: &[PhysicalSortExpr],
+ ) -> Result<Option<Arc<dyn DataSource>>> {
+ let current_ordering = match self.output_ordering.first() {
+ Some(ordering) => ordering.as_ref(),
+ None => return Ok(None),
+ };
+
+ // Only support reverse ordering pushdown
+ if !is_reverse_ordering(order, current_ordering) {
+ return Ok(None);
+ }
+
+ // Ask the file source if it can handle the sort pushdown
+ // (e.g., ParquetSource will enable reverse_scan)
+ let new_file_source = match self.file_source.try_pushdown_sort(order)? {
+ Some(source) => source,
+ None => return Ok(None),
+ };
+
+ let mut new_config = self.clone();
+
+ // Reverse file groups: when scanning in reverse, we need to read files
+ // in reverse order to maintain the correct global ordering
+ new_config.file_groups = new_config
+ .file_groups
+ .into_iter()
+ .map(|group| {
+ let mut files = group.into_inner();
+ files.reverse();
+ files.into()
+ })
+ .collect();
+
+ // Build the new output ordering by reversing each sort expression's direction
+ // E.g., [number DESC] becomes [number ASC]
+ let mut reversed_ordering = Vec::new();
+ for sort_expr in current_ordering {
+ reversed_ordering.push(PhysicalSortExpr {
+ expr: Arc::clone(&sort_expr.expr),
+ options: !sort_expr.options,
+ });
+ }
+
+ new_config.output_ordering = vec![LexOrdering::new(reversed_ordering)
+ .ok_or_else(|| {
+ DataFusionError::Plan(
+ "Failed to create ordering: invalid sort expressions".to_string(),
+ )
+ })?];
+
+ new_config.file_source = new_file_source;
+
+ Ok(Some(Arc::new(new_config)))
+ }
+}
+
+/// Check if the requested ordering can be satisfied by reversing the current ordering.
+///
+/// This function supports **prefix matching**: if the file has ordering [A DESC, B ASC]
+/// and we need [A ASC], reversing the scan gives us [A ASC, B DESC], which satisfies
+/// the requirement since [A ASC] is a prefix.
+///
+/// # Arguments
+/// * `requested` - The ordering required by the query
+/// * `current` - The natural ordering of the data source (e.g., from file metadata)
+///
+/// # Returns
+/// `true` if reversing the current ordering would satisfy the requested ordering
+///
+/// # Example
+/// ```text
+/// Current: [number DESC, letter ASC]
+/// Requested: [number ASC]
+/// Reversed: [number ASC, letter DESC] โ Prefix match!
+/// ```
+fn is_reverse_ordering(
+ requested: &[PhysicalSortExpr],
+ current: &[PhysicalSortExpr],
+) -> bool {
+ // Allow prefix matching - we can satisfy a prefix of the current ordering
+ // by reversing the scan
+ if requested.len() > current.len() {
+ return false;
+ }
+
+ requested.iter().zip(current.iter()).all(|(req, cur)| {
+ let exprs_match = req.expr.to_string() == cur.expr.to_string();
Review Comment:
Thanks @alamb , i add more comments, compare the sort exprs is not we want here, we want the sort string without sort order. Correct me if i am wrong.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2554047913
##########
datafusion/sqllogictest/test_files/topk.slt:
##########
@@ -338,9 +338,7 @@ physical_plan
query TT
explain select number, letter, age from partial_sorted order by number asc limit 3;
----
-physical_plan
-01)SortExec: TopK(fetch=3), expr=[number@0 ASC NULLS LAST], preserve_partitioning=[false]
Review Comment:
Topk also optimized to remove the sort.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3812014893
Yes @adriangb , closed it 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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3806487988
@zhuqi-lucas this work has been superseded right? Should we close 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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3577878651
I plan to review this PR carefully tomorrow
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3566722719
cc @alamb @crepererum @xudong963 The PR is ready for review 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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3586503426
> ๐ค: Benchmark completed
>
> Details
Interesting, the performance Q23 44% improvement, but it seems the parquet in benchmark is not sorted?
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3601188999
https://github.com/apache/datafusion/pull/19042#issue-3684639836
@alamb @xudong963 @2010YOUY01 @adriangb
I got some real data benchmark result for this PR comparing to main branch, it's 30X faster, so above benchmark, i can add more benchmark cases as follow-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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2582071371
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -485,6 +490,15 @@ impl ParquetSource {
)),
}
}
+
+ pub fn with_reverse_scan(mut self, reverse_scan: bool) -> Self {
+ self.reverse_scan = reverse_scan;
+ self
+ }
Review Comment:
Are these used from production code (the optimizers, etc.) or only from our tests?
##########
datafusion/common/src/config.rs:
##########
@@ -831,6 +831,15 @@ config_namespace! {
/// writing out already in-memory data, such as from a cached
/// data frame.
pub maximum_buffered_record_batches_per_stream: usize, default = 2
+
+ /// Enable sort pushdown optimization for sorted Parquet files.
+ /// Currently, this optimization only has reverse order support.
+ /// When a query requires ordering that can be satisfied by reversing
+ /// the file's natural ordering, row groups and batches are read in
+ /// reverse order to eliminate sort operations.
+ /// Note: This buffers one row group at a time (typically ~128MB).
+ /// Default: true
Review Comment:
Can this support non-reverse cases where the query order is the same as the file order? i.e. can we eliminate sorts in that simpler case as well? Or does that already happen / was already implemented?
##########
datafusion/datasource/src/file_scan_config.rs:
##########
@@ -766,6 +766,131 @@ impl DataSource for FileScanConfig {
}
}
}
+
+ fn try_pushdown_sort(
+ &self,
+ order: &[PhysicalSortExpr],
+ ) -> Result<Option<Arc<dyn DataSource>>> {
+ let current_ordering = match self.output_ordering.first() {
Review Comment:
Nit for a bigger picture plan: I think we should re-arrange things such that the ordering of each file is recorded in `PartitionedFile` and any known ordering of groups in `FileGroup`. Then `FileScanConfig` should calculate it's output ordering from that instead of being given one. And if sort pushdown is requested then `FileScanConfig` can:
1. Try to re-arrange the order by reversing groups, by re-creating groups entirely using stats. This is enough for `Inexact` in https://github.com/apache/datafusion/pull/18817/files#r2582092389.
2. Try to push down the preferred order to the `FileSource` which determines if the order of reading a given file can be reversed. Maybe it needs a reference to the files since it has to handle row group order i.e. Parquet specific stuff? But if it is able to reverse the order of the scans then the whole result becomes `Exact`.
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -806,4 +844,87 @@ mod tests {
// same value. but filter() call Arc::clone internally
assert_eq!(parquet_source.predicate(), parquet_source.filter().as_ref());
}
+
+ #[test]
+ fn test_reverse_scan_default_value() {
+ use arrow::datatypes::Schema;
+
+ let schema = Arc::new(Schema::empty());
+ let source = ParquetSource::new(schema);
+
+ assert!(!source.reverse_scan());
Review Comment:
We could just access the private field here since it's in the same module.
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -788,6 +803,29 @@ impl FileSource for ParquetSource {
fn schema_adapter_factory(&self) -> Option<Arc<dyn SchemaAdapterFactory>> {
self.schema_adapter_factory.clone()
}
+
+ /// When push down to parquet source of a sort operation is possible,
+ /// create a new ParquetSource with reverse_scan enabled.
+ /// TODO support more policies in addition to reversing the scan.
+ fn try_pushdown_sort(
+ &self,
+ _order: &[PhysicalSortExpr],
+ ) -> datafusion_common::Result<Option<Arc<dyn FileSource>>> {
+ // Note: We ignore the specific `order` parameter here because the decision
+ // about whether we can reverse is made at the FileScanConfig level.
+ // This method creates a reversed version of the current ParquetSource,
+ // and the FileScanConfig will reverse both the file list and the declared ordering.
+
+ // Clone the entire source to preserve ALL configuration including:
+ // - projection (CRITICAL: prevents schema mismatch)
+ // - predicate
+ // - batch_size
+ // - table_parquet_options
+ // - all other settings
+ let new_source = self.clone().with_reverse_scan(true);
Review Comment:
If this is the only place `with_reverse_scan` is used I suggest we set the private field directly to avoid polluting the public API.
##########
datafusion/physical-plan/src/execution_plan.rs:
##########
@@ -682,6 +684,16 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
) -> Option<Arc<dyn ExecutionPlan>> {
None
}
+
+ /// Try to create a new execution plan that satisfies the given sort ordering.
+ ///
+ /// Default implementation returns `Ok(None)`.
+ fn try_pushdown_sort(
+ &self,
+ _order: &[PhysicalSortExpr],
+ ) -> Result<Option<Arc<dyn ExecutionPlan>>> {
Review Comment:
I similarly think there should be the option to return a new plan that is optimized for the order but doesn't satisfy it enough to remove the sort (https://github.com/apache/datafusion/pull/18817/files#r2582092389)
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -788,6 +803,29 @@ impl FileSource for ParquetSource {
fn schema_adapter_factory(&self) -> Option<Arc<dyn SchemaAdapterFactory>> {
self.schema_adapter_factory.clone()
}
+
+ /// When push down to parquet source of a sort operation is possible,
+ /// create a new ParquetSource with reverse_scan enabled.
+ /// TODO support more policies in addition to reversing the scan.
+ fn try_pushdown_sort(
+ &self,
+ _order: &[PhysicalSortExpr],
+ ) -> datafusion_common::Result<Option<Arc<dyn FileSource>>> {
+ // Note: We ignore the specific `order` parameter here because the decision
+ // about whether we can reverse is made at the FileScanConfig level.
+ // This method creates a reversed version of the current ParquetSource,
+ // and the FileScanConfig will reverse both the file list and the declared ordering.
Review Comment:
Just to clarify: since `FileScanConfig` is called first we are assuming that if we were called the intention is to reverse the order? I would rather verify this / decouple these two things. That is I'd rather parse the sort order here and do whatever we need to do internally to satisfy it (or respond `None` if we can't) instead of relying on `FileScanConfig` to do it for us.
##########
datafusion/datasource/src/file.rs:
##########
@@ -129,6 +130,29 @@ pub trait FileSource: Send + Sync {
))
}
+ /// Try to create a new FileSource that can produce data in the specified sort order.
+ ///
+ /// This allows file format implementations to optimize based on the required sort order.
+ /// For example:
+ /// - ParquetSource can reverse scan direction
+ /// - Future implementations might reorder row groups or use native indexes
+ ///
+ /// # Arguments
+ /// * `order` - The desired output ordering
+ ///
+ /// # Returns
+ /// * `Ok(Some(source))` - Created a source that can satisfy the ordering
+ /// * `Ok(None)` - Cannot optimize for this ordering
+ /// * `Err(e)` - Error occurred
+ ///
+ /// Default implementation returns `Ok(None)`.
+ fn try_pushdown_sort(
+ &self,
+ _order: &[PhysicalSortExpr],
+ ) -> Result<Option<Arc<dyn FileSource>>> {
Review Comment:
I think it might be nice to make this an enum instead of an option. Specifically:
```rust
pub enum SortOrderPushdownResult<T> {
Exact { inner: T },
Inexact { inner: T },
Unsupported,
}
```
The idea being that `Inexact` means "I changed myself to better satisfy the join order, but cannot guarantee it's perfectly sorted (e.g. only re-arranged files and row groups based on stats but not the actual data stream).
This matters because e.g. `Inexact` would already be a *huge* win for TopK + dynamic filter pushdown.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2554026346
##########
datafusion/physical-optimizer/src/reverse_order.rs:
##########
@@ -0,0 +1,1446 @@
+// 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 crate::PhysicalOptimizerRule;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{
+ Transformed, TransformedResult, TreeNode, TreeNodeRecursion, TreeNodeRewriter,
+};
+use datafusion_common::Result;
+use datafusion_datasource::file::FileSource;
+use datafusion_datasource::file_scan_config::{FileScanConfig, FileScanConfigBuilder};
+use datafusion_datasource::source::{DataSource, DataSourceExec};
+use datafusion_datasource_parquet::source::ParquetSource;
+use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr};
+use datafusion_physical_plan::joins::SortMergeJoinExec;
+use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
+use datafusion_physical_plan::sorts::sort::SortExec;
+use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+use std::sync::Arc;
+
+/// A PhysicalOptimizerRule that attempts to reverse a SortExec's input if doing so would make the
+/// input ordering compatible with the SortExec's required output ordering.
+/// It also removes unnecessary sorts when the input already satisfies the required ordering.
+#[derive(Debug, Clone, Default)]
+pub struct ReverseOrder;
+
+impl ReverseOrder {
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for ReverseOrder {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ // Check if reverse scan optimization is enabled
+ let enable_reverse_scan = config.execution.parquet.enable_reverse_scan;
+
+ // Return early if not enabled
+ if !enable_reverse_scan {
+ return Ok(plan);
+ }
+
+ // Search for any SortExec nodes and try to optimize them
+ plan.transform_up(&|plan: Arc<dyn ExecutionPlan>| {
+ // First check if this is a GlobalLimitExec -> SortExec pattern
+ if let Some(limit_exec) = plan.as_any().downcast_ref::<GlobalLimitExec>() {
+ if let Some(sort_exec) =
+ limit_exec.input().as_any().downcast_ref::<SortExec>()
+ {
+ return optimize_limit_sort(limit_exec, sort_exec);
+ }
+ }
+
+ // Otherwise, check if this is just a SortExec
+ let sort_exec = match plan.as_any().downcast_ref::<SortExec>() {
+ Some(sort_exec) => sort_exec,
+ None => return Ok(Transformed::no(plan)),
+ };
+
+ let sort_input: Arc<dyn ExecutionPlan> = Arc::clone(sort_exec.input());
+
+ // First, check if the sort is already satisfied by input ordering
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+
+ // Check if input already satisfies the sort requirement
+ if input_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ return remove_unnecessary_sort(sort_exec, sort_input);
+ }
+ }
+
+ // If not satisfied, try to reverse the input
+ let reversed_eq_properties = {
+ let mut new = sort_input.properties().equivalence_properties().clone();
+ new.clear_orderings();
+
+ // Build reversed sort exprs for each ordering class
+ let reversed_orderings = sort_input
+ .equivalence_properties()
+ .oeq_class()
+ .iter()
+ .map(|ordering| {
+ ordering
+ .iter()
+ .map(|expr| expr.reverse())
+ .collect::<Vec<_>>()
+ })
+ .collect::<Vec<_>>();
+
+ new.add_orderings(reversed_orderings);
+ new
+ };
+
+ match reversed_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ true => {
+ // Reverse the input and then remove the sort
+ let reversed_input =
+ sort_input.rewrite(&mut ReverseRewriter).unwrap().data;
+
+ // After reversing, check if we can remove the sort
+ if reversed_input
+ .equivalence_properties()
+ .ordering_satisfy(sort_exec.expr().clone())?
+ {
+ remove_unnecessary_sort(sort_exec, reversed_input)
+ } else {
+ // Keep the sort but with reversed input
+ Ok(Transformed::yes(Arc::new(
+ SortExec::new(sort_exec.expr().clone(), reversed_input)
+ .with_fetch(sort_exec.fetch())
+ .with_preserve_partitioning(
+ sort_exec.preserve_partitioning(),
+ ),
+ )))
+ }
+ }
+ false => Ok(Transformed::no(plan)),
+ }
+ })
+ .data()
+ }
+
+ fn name(&self) -> &str {
+ "ReverseOrder"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// Handle the GlobalLimitExec -> SortExec pattern
+fn optimize_limit_sort(
+ limit_exec: &GlobalLimitExec,
+ sort_exec: &SortExec,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+
+ // Check if input is already sorted
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+ if input_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ // Input is already sorted correctly, remove sort and keep limit
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ sort_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Check if we can reverse the input to satisfy the sort
+ let reversed_eq_properties = {
+ let mut new = sort_input.properties().equivalence_properties().clone();
+ new.clear_orderings();
+
+ let reversed_orderings = sort_input
+ .equivalence_properties()
+ .oeq_class()
+ .iter()
+ .map(|ordering| {
+ ordering
+ .iter()
+ .map(|expr| expr.reverse())
+ .collect::<Vec<_>>()
+ })
+ .collect::<Vec<_>>();
+
+ new.add_orderings(reversed_orderings);
+ new
+ };
+
+ if reversed_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ // Can reverse! Apply reversal
+ let reversed_input = sort_input.rewrite(&mut ReverseRewriter).unwrap().data;
+
+ // Check if reversed input satisfies the sort requirement
+ if reversed_input
+ .equivalence_properties()
+ .ordering_satisfy(sort_exec.expr().clone())?
+ {
+ // Check if this is a single-partition DataSourceExec with reverse_scan enabled
+ // In that case, the limit is already handled internally by ReversedParquetStreamWithLimit
+ if is_single_partition_reverse_scan_datasource(&reversed_input) {
+ let total_fetch = limit_exec.skip() + limit_exec.fetch().unwrap_or(0);
+
+ if let Some(with_fetch) = reversed_input.with_fetch(Some(total_fetch)) {
+ if limit_exec.skip() > 0 {
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ with_fetch,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ } else {
+ return Ok(Transformed::yes(with_fetch));
+ }
+ }
+
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ reversed_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+
+ // Otherwise, remove sort but keep limit with reversed input
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ reversed_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Can't optimize, return original pattern
+ Ok(Transformed::no(Arc::new(GlobalLimitExec::new(
+ Arc::new(sort_exec.clone()),
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))))
+}
+
+/// Check if the plan is a single-partition DataSourceExec with reverse_scan enabled
+fn is_single_partition_reverse_scan_datasource(plan: &Arc<dyn ExecutionPlan>) -> bool {
+ // Only optimize for single partition
+ if plan.output_partitioning().partition_count() != 1 {
+ return false;
+ }
+
+ if let Some(data_source_exec) = plan.as_any().downcast_ref::<DataSourceExec>() {
+ if let Some(scan_config) = data_source_exec
+ .data_source()
+ .as_any()
+ .downcast_ref::<FileScanConfig>()
+ {
+ if let Some(parquet_source) = scan_config
+ .file_source
+ .as_any()
+ .downcast_ref::<ParquetSource>()
+ {
+ return parquet_source.reverse_scan();
+ }
+ }
+ }
+ false
+}
+
+/// Remove unnecessary sort based on the logic from EnforceSorting::analyze_immediate_sort_removal
+fn remove_unnecessary_sort(
Review Comment:
I add this to reverse order because after reverse order, we can optimize more to remove the sort, so we don't need to execute the enforce sort optimization again after this optimization.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3570601500
> > I haven't looked into all of this discussion and code (I just got tagged). I've been looking into optimizing sorted scanning in DataFusion and IMO where we should land is:
> >
> > 1. Via metadata (`FileScanConfig` / `ORDERED BY ...` in SQL) users declare a known sort order of their files.
> > 2. The planner uses statistics from the files + any `ORDER BY` clauses in the query to arrange file ordering to best match the query. The `FileSource` implementation can also receive the `ORDER BY` information and optimize scan order within a file (e.g. reversing the order of reads which is what I think this PR is doing).
> > 3. If the planner is able to deduce from file level stats that the files can be ordered and the `FileSource` reports that it is able to produce batches in sorted order then the optimizer can optimize away the sort completely.
> >
> > I hope that is helpful.
>
> Thank you @adriangb , it's helpful for future optimization. My current implementation focuses on a specific optimization case: when data is already sorted and we need the reverse order, we can flip the scan direction instead of reading everything and sorting. The reverse_scan in FileSource handles the files/ and within-file ordering reversal.
>
> I think these approaches are complementary - my PR handles the reverse scan optimization, while your vision provides a framework for broader sorted-scan optimizations using file-level statistics and metadata. Would be great to build toward that architecture incrementally.
My point is that instead of `enable_reverse_scan: bool` we might want to consider a more holistic approach e.g. `try_pushdown_sort(&self, order: LexOrdering) -> Result<Option<Arc<dyn ExecutionPlan>>>` either at the `ExecutionPlan` level or at the `DataSource` level.
I'm not opposed to this as a step towards that but I'm not sure how helpful it is.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2554047913
##########
datafusion/sqllogictest/test_files/topk.slt:
##########
@@ -338,9 +338,7 @@ physical_plan
query TT
explain select number, letter, age from partial_sorted order by number asc limit 3;
----
-physical_plan
-01)SortExec: TopK(fetch=3), expr=[number@0 ASC NULLS LAST], preserve_partitioning=[false]
Review Comment:
Topk also optimized to remove the sort.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3569570355
FYI, I'll start reviewing the PR tomorrow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3593375069
๐ค: Benchmark completed
<details><summary>Details</summary>
<p>
```
Comparing HEAD and reverse_parquet
--------------------
Benchmark clickbench_partitioned.json
--------------------
โโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโ
โ Query โ HEAD โ reverse_parquet โ Change โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ QQuery 0 โ 2.48 ms โ 2.22 ms โ +1.12x faster โ
โ QQuery 1 โ 50.16 ms โ 50.08 ms โ no change โ
โ QQuery 2 โ 133.40 ms โ 139.58 ms โ no change โ
โ QQuery 3 โ 164.66 ms โ 161.46 ms โ no change โ
โ QQuery 4 โ 1091.17 ms โ 1078.76 ms โ no change โ
โ QQuery 5 โ 1486.75 ms โ 1513.08 ms โ no change โ
โ QQuery 6 โ 2.30 ms โ 2.31 ms โ no change โ
โ QQuery 7 โ 55.40 ms โ 55.05 ms โ no change โ
โ QQuery 8 โ 1457.64 ms โ 1431.16 ms โ no change โ
โ QQuery 9 โ 1843.33 ms โ 1846.58 ms โ no change โ
โ QQuery 10 โ 393.34 ms โ 368.55 ms โ +1.07x faster โ
โ QQuery 11 โ 443.33 ms โ 416.22 ms โ +1.07x faster โ
โ QQuery 12 โ 1387.03 ms โ 1381.09 ms โ no change โ
โ QQuery 13 โ 2134.87 ms โ 2155.86 ms โ no change โ
โ QQuery 14 โ 1277.23 ms โ 1259.77 ms โ no change โ
โ QQuery 15 โ 1244.84 ms โ 1247.28 ms โ no change โ
โ QQuery 16 โ 2718.87 ms โ 2727.31 ms โ no change โ
โ QQuery 17 โ 2702.24 ms โ 2694.43 ms โ no change โ
โ QQuery 18 โ 5553.93 ms โ 5044.15 ms โ +1.10x faster โ
โ QQuery 19 โ 128.84 ms โ 122.91 ms โ no change โ
โ QQuery 20 โ 2091.73 ms โ 1995.54 ms โ no change โ
โ QQuery 21 โ 2357.94 ms โ 2310.57 ms โ no change โ
โ QQuery 22 โ 4018.00 ms โ 3973.11 ms โ no change โ
โ QQuery 23 โ 17232.00 ms โ 12942.16 ms โ +1.33x faster โ
โ QQuery 24 โ 229.18 ms โ 211.51 ms โ +1.08x faster โ
โ QQuery 25 โ 501.10 ms โ 473.70 ms โ +1.06x faster โ
โ QQuery 26 โ 238.12 ms โ 214.18 ms โ +1.11x faster โ
โ QQuery 27 โ 2964.51 ms โ 2751.22 ms โ +1.08x faster โ
โ QQuery 28 โ 23628.96 ms โ 23294.09 ms โ no change โ
โ QQuery 29 โ 978.17 ms โ 985.62 ms โ no change โ
โ QQuery 30 โ 1351.92 ms โ 1335.56 ms โ no change โ
โ QQuery 31 โ 1409.06 ms โ 1351.65 ms โ no change โ
โ QQuery 32 โ 5463.16 ms โ 5079.21 ms โ +1.08x faster โ
โ QQuery 33 โ 6250.86 ms โ 5939.28 ms โ no change โ
โ QQuery 34 โ 6613.53 ms โ 6093.67 ms โ +1.09x faster โ
โ QQuery 35 โ 1952.58 ms โ 1962.50 ms โ no change โ
โ QQuery 36 โ 121.34 ms โ 123.10 ms โ no change โ
โ QQuery 37 โ 51.52 ms โ 54.41 ms โ 1.06x slower โ
โ QQuery 38 โ 119.42 ms โ 119.16 ms โ no change โ
โ QQuery 39 โ 197.37 ms โ 204.51 ms โ no change โ
โ QQuery 40 โ 41.48 ms โ 42.44 ms โ no change โ
โ QQuery 41 โ 39.95 ms โ 39.71 ms โ no change โ
โ QQuery 42 โ 32.96 ms โ 33.32 ms โ no change โ
โโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโ
โ Benchmark Summary โ โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ Total Time (HEAD) โ 102156.66ms โ
โ Total Time (reverse_parquet) โ 95228.06ms โ
โ Average Time (HEAD) โ 2375.74ms โ
โ Average Time (reverse_parquet) โ 2214.61ms โ
โ Queries Faster โ 11 โ
โ Queries Slower โ 1 โ
โ Queries with No Change โ 31 โ
โ 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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2583505075
##########
datafusion/datasource-parquet/src/source.rs:
##########
@@ -485,6 +490,15 @@ impl ParquetSource {
)),
}
}
+
+ pub fn with_reverse_scan(mut self, reverse_scan: bool) -> Self {
+ self.reverse_scan = reverse_scan;
+ self
+ }
Review Comment:
The similar code already in our production for very long time, and it works 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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3588053320
Also created the ticket for adding benchmark for sorted data cases:
https://github.com/apache/datafusion/issues/18976
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3566725990
Also cc @suremarc, finally, we're contributing our reversed parquet optimization to upstream, I guess you may be interested in seeing 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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2559107427
##########
datafusion/physical-optimizer/src/pushdown_sort.rs:
##########
@@ -0,0 +1,1446 @@
+// 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 crate::PhysicalOptimizerRule;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{
+ Transformed, TransformedResult, TreeNode, TreeNodeRecursion, TreeNodeRewriter,
+};
+use datafusion_common::Result;
+use datafusion_datasource::file::FileSource;
+use datafusion_datasource::file_scan_config::{FileScanConfig, FileScanConfigBuilder};
+use datafusion_datasource::source::{DataSource, DataSourceExec};
+use datafusion_datasource_parquet::source::ParquetSource;
+use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr};
+use datafusion_physical_plan::joins::SortMergeJoinExec;
+use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
+use datafusion_physical_plan::sorts::sort::SortExec;
+use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+use std::sync::Arc;
+
+/// A PhysicalOptimizerRule that attempts to reverse a SortExec's input if doing so would make the
+/// input ordering compatible with the SortExec's required output ordering.
+/// It also removes unnecessary sorts when the input already satisfies the required ordering.
+#[derive(Debug, Clone, Default)]
+pub struct ReverseOrder;
+
+impl ReverseOrder {
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for ReverseOrder {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ // Check if reverse scan optimization is enabled
+ let enable_reverse_scan = config.execution.parquet.enable_reverse_scan;
+
+ // Return early if not enabled
+ if !enable_reverse_scan {
+ return Ok(plan);
+ }
+
+ // Search for any SortExec nodes and try to optimize them
+ plan.transform_up(&|plan: Arc<dyn ExecutionPlan>| {
+ // First check if this is a GlobalLimitExec -> SortExec pattern
+ if let Some(limit_exec) = plan.as_any().downcast_ref::<GlobalLimitExec>() {
+ if let Some(sort_exec) =
+ limit_exec.input().as_any().downcast_ref::<SortExec>()
+ {
+ return optimize_limit_sort(limit_exec, sort_exec);
+ }
+ }
+
+ // Otherwise, check if this is just a SortExec
+ let sort_exec = match plan.as_any().downcast_ref::<SortExec>() {
+ Some(sort_exec) => sort_exec,
+ None => return Ok(Transformed::no(plan)),
+ };
+
+ let sort_input: Arc<dyn ExecutionPlan> = Arc::clone(sort_exec.input());
+
+ // First, check if the sort is already satisfied by input ordering
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+
+ // Check if input already satisfies the sort requirement
+ if input_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ return remove_unnecessary_sort(sort_exec, sort_input);
+ }
+ }
+
+ // If not satisfied, try to reverse the input
+ let reversed_eq_properties = {
+ let mut new = sort_input.properties().equivalence_properties().clone();
+ new.clear_orderings();
+
+ // Build reversed sort exprs for each ordering class
+ let reversed_orderings = sort_input
+ .equivalence_properties()
+ .oeq_class()
+ .iter()
+ .map(|ordering| {
+ ordering
+ .iter()
+ .map(|expr| expr.reverse())
+ .collect::<Vec<_>>()
+ })
+ .collect::<Vec<_>>();
+
+ new.add_orderings(reversed_orderings);
+ new
+ };
+
+ match reversed_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ true => {
+ // Reverse the input and then remove the sort
+ let reversed_input =
+ sort_input.rewrite(&mut ReverseRewriter).unwrap().data;
+
+ // After reversing, check if we can remove the sort
+ if reversed_input
+ .equivalence_properties()
+ .ordering_satisfy(sort_exec.expr().clone())?
+ {
+ remove_unnecessary_sort(sort_exec, reversed_input)
+ } else {
+ // Keep the sort but with reversed input
+ Ok(Transformed::yes(Arc::new(
+ SortExec::new(sort_exec.expr().clone(), reversed_input)
+ .with_fetch(sort_exec.fetch())
+ .with_preserve_partitioning(
+ sort_exec.preserve_partitioning(),
+ ),
+ )))
+ }
+ }
+ false => Ok(Transformed::no(plan)),
+ }
+ })
+ .data()
+ }
+
+ fn name(&self) -> &str {
+ "ReverseOrder"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// Handle the GlobalLimitExec -> SortExec pattern
+fn optimize_limit_sort(
+ limit_exec: &GlobalLimitExec,
+ sort_exec: &SortExec,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+
+ // Check if input is already sorted
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+ if input_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ // Input is already sorted correctly, remove sort and keep limit
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ sort_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Check if we can reverse the input to satisfy the sort
+ let reversed_eq_properties = {
+ let mut new = sort_input.properties().equivalence_properties().clone();
+ new.clear_orderings();
+
+ let reversed_orderings = sort_input
+ .equivalence_properties()
+ .oeq_class()
+ .iter()
+ .map(|ordering| {
+ ordering
+ .iter()
+ .map(|expr| expr.reverse())
+ .collect::<Vec<_>>()
+ })
+ .collect::<Vec<_>>();
+
+ new.add_orderings(reversed_orderings);
+ new
+ };
+
+ if reversed_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ // Can reverse! Apply reversal
+ let reversed_input = sort_input.rewrite(&mut ReverseRewriter).unwrap().data;
+
+ // Check if reversed input satisfies the sort requirement
+ if reversed_input
+ .equivalence_properties()
+ .ordering_satisfy(sort_exec.expr().clone())?
+ {
+ // Check if this is a single-partition DataSourceExec with reverse_scan enabled
+ // In that case, the limit is already handled internally by ReversedParquetStreamWithLimit
+ if is_single_partition_reverse_scan_datasource(&reversed_input) {
+ let total_fetch = limit_exec.skip() + limit_exec.fetch().unwrap_or(0);
+
+ if let Some(with_fetch) = reversed_input.with_fetch(Some(total_fetch)) {
+ if limit_exec.skip() > 0 {
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ with_fetch,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ } else {
+ return Ok(Transformed::yes(with_fetch));
+ }
+ }
+
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ reversed_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+
+ // Otherwise, remove sort but keep limit with reversed input
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ reversed_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Can't optimize, return original pattern
+ Ok(Transformed::no(Arc::new(GlobalLimitExec::new(
+ Arc::new(sort_exec.clone()),
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))))
+}
+
+/// Check if the plan is a single-partition DataSourceExec with reverse_scan enabled
+fn is_single_partition_reverse_scan_datasource(plan: &Arc<dyn ExecutionPlan>) -> bool {
+ // Only optimize for single partition
+ if plan.output_partitioning().partition_count() != 1 {
+ return false;
+ }
+
+ if let Some(data_source_exec) = plan.as_any().downcast_ref::<DataSourceExec>() {
+ if let Some(scan_config) = data_source_exec
+ .data_source()
+ .as_any()
+ .downcast_ref::<FileScanConfig>()
+ {
+ if let Some(parquet_source) = scan_config
+ .file_source
+ .as_any()
+ .downcast_ref::<ParquetSource>()
+ {
+ return parquet_source.reverse_scan();
+ }
+ }
+ }
+ false
+}
+
+/// Remove unnecessary sort based on the logic from EnforceSorting::analyze_immediate_sort_removal
+fn remove_unnecessary_sort(
Review Comment:
How about putting `pushdown_sort` before `enforce_sorting`
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2559130058
##########
datafusion/physical-optimizer/src/pushdown_sort.rs:
##########
@@ -0,0 +1,1446 @@
+// 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 crate::PhysicalOptimizerRule;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{
+ Transformed, TransformedResult, TreeNode, TreeNodeRecursion, TreeNodeRewriter,
+};
+use datafusion_common::Result;
+use datafusion_datasource::file::FileSource;
+use datafusion_datasource::file_scan_config::{FileScanConfig, FileScanConfigBuilder};
+use datafusion_datasource::source::{DataSource, DataSourceExec};
+use datafusion_datasource_parquet::source::ParquetSource;
+use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr};
+use datafusion_physical_plan::joins::SortMergeJoinExec;
+use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
+use datafusion_physical_plan::sorts::sort::SortExec;
+use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+use std::sync::Arc;
+
+/// A PhysicalOptimizerRule that attempts to reverse a SortExec's input if doing so would make the
+/// input ordering compatible with the SortExec's required output ordering.
+/// It also removes unnecessary sorts when the input already satisfies the required ordering.
+#[derive(Debug, Clone, Default)]
+pub struct ReverseOrder;
+
+impl ReverseOrder {
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for ReverseOrder {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ // Check if reverse scan optimization is enabled
+ let enable_reverse_scan = config.execution.parquet.enable_reverse_scan;
+
+ // Return early if not enabled
+ if !enable_reverse_scan {
+ return Ok(plan);
+ }
+
+ // Search for any SortExec nodes and try to optimize them
+ plan.transform_up(&|plan: Arc<dyn ExecutionPlan>| {
+ // First check if this is a GlobalLimitExec -> SortExec pattern
+ if let Some(limit_exec) = plan.as_any().downcast_ref::<GlobalLimitExec>() {
+ if let Some(sort_exec) =
+ limit_exec.input().as_any().downcast_ref::<SortExec>()
+ {
+ return optimize_limit_sort(limit_exec, sort_exec);
+ }
+ }
+
+ // Otherwise, check if this is just a SortExec
+ let sort_exec = match plan.as_any().downcast_ref::<SortExec>() {
+ Some(sort_exec) => sort_exec,
+ None => return Ok(Transformed::no(plan)),
+ };
+
+ let sort_input: Arc<dyn ExecutionPlan> = Arc::clone(sort_exec.input());
+
+ // First, check if the sort is already satisfied by input ordering
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+
+ // Check if input already satisfies the sort requirement
+ if input_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ return remove_unnecessary_sort(sort_exec, sort_input);
+ }
+ }
+
+ // If not satisfied, try to reverse the input
+ let reversed_eq_properties = {
+ let mut new = sort_input.properties().equivalence_properties().clone();
+ new.clear_orderings();
+
+ // Build reversed sort exprs for each ordering class
+ let reversed_orderings = sort_input
+ .equivalence_properties()
+ .oeq_class()
+ .iter()
+ .map(|ordering| {
+ ordering
+ .iter()
+ .map(|expr| expr.reverse())
+ .collect::<Vec<_>>()
+ })
+ .collect::<Vec<_>>();
+
+ new.add_orderings(reversed_orderings);
+ new
+ };
+
+ match reversed_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ true => {
+ // Reverse the input and then remove the sort
+ let reversed_input =
+ sort_input.rewrite(&mut ReverseRewriter).unwrap().data;
+
+ // After reversing, check if we can remove the sort
+ if reversed_input
+ .equivalence_properties()
+ .ordering_satisfy(sort_exec.expr().clone())?
+ {
+ remove_unnecessary_sort(sort_exec, reversed_input)
+ } else {
+ // Keep the sort but with reversed input
+ Ok(Transformed::yes(Arc::new(
+ SortExec::new(sort_exec.expr().clone(), reversed_input)
+ .with_fetch(sort_exec.fetch())
+ .with_preserve_partitioning(
+ sort_exec.preserve_partitioning(),
+ ),
+ )))
+ }
+ }
+ false => Ok(Transformed::no(plan)),
+ }
+ })
+ .data()
+ }
+
+ fn name(&self) -> &str {
+ "ReverseOrder"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// Handle the GlobalLimitExec -> SortExec pattern
+fn optimize_limit_sort(
+ limit_exec: &GlobalLimitExec,
+ sort_exec: &SortExec,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+
+ // Check if input is already sorted
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+ if input_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ // Input is already sorted correctly, remove sort and keep limit
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ sort_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Check if we can reverse the input to satisfy the sort
+ let reversed_eq_properties = {
+ let mut new = sort_input.properties().equivalence_properties().clone();
+ new.clear_orderings();
+
+ let reversed_orderings = sort_input
+ .equivalence_properties()
+ .oeq_class()
+ .iter()
+ .map(|ordering| {
+ ordering
+ .iter()
+ .map(|expr| expr.reverse())
+ .collect::<Vec<_>>()
+ })
+ .collect::<Vec<_>>();
+
+ new.add_orderings(reversed_orderings);
+ new
+ };
+
+ if reversed_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ // Can reverse! Apply reversal
+ let reversed_input = sort_input.rewrite(&mut ReverseRewriter).unwrap().data;
+
+ // Check if reversed input satisfies the sort requirement
+ if reversed_input
+ .equivalence_properties()
+ .ordering_satisfy(sort_exec.expr().clone())?
+ {
+ // Check if this is a single-partition DataSourceExec with reverse_scan enabled
+ // In that case, the limit is already handled internally by ReversedParquetStreamWithLimit
+ if is_single_partition_reverse_scan_datasource(&reversed_input) {
+ let total_fetch = limit_exec.skip() + limit_exec.fetch().unwrap_or(0);
+
+ if let Some(with_fetch) = reversed_input.with_fetch(Some(total_fetch)) {
+ if limit_exec.skip() > 0 {
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ with_fetch,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ } else {
+ return Ok(Transformed::yes(with_fetch));
+ }
+ }
+
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ reversed_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+
+ // Otherwise, remove sort but keep limit with reversed input
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ reversed_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Can't optimize, return original pattern
+ Ok(Transformed::no(Arc::new(GlobalLimitExec::new(
+ Arc::new(sort_exec.clone()),
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))))
+}
+
+/// Check if the plan is a single-partition DataSourceExec with reverse_scan enabled
+fn is_single_partition_reverse_scan_datasource(plan: &Arc<dyn ExecutionPlan>) -> bool {
+ // Only optimize for single partition
+ if plan.output_partitioning().partition_count() != 1 {
+ return false;
+ }
+
+ if let Some(data_source_exec) = plan.as_any().downcast_ref::<DataSourceExec>() {
+ if let Some(scan_config) = data_source_exec
+ .data_source()
+ .as_any()
+ .downcast_ref::<FileScanConfig>()
+ {
+ if let Some(parquet_source) = scan_config
+ .file_source
+ .as_any()
+ .downcast_ref::<ParquetSource>()
+ {
+ return parquet_source.reverse_scan();
+ }
+ }
+ }
+ false
+}
+
+/// Remove unnecessary sort based on the logic from EnforceSorting::analyze_immediate_sort_removal
+fn remove_unnecessary_sort(
Review Comment:
Thanks @xudong963 , i tried this before, but it seems caused other problems for other optimizers issues, i can test again.
The best solution may be that we don't need a new optimizer pushdown_sort, so we can just enhance the existed optimizer to support it. I will try this later.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2568121671
##########
datafusion/physical-optimizer/src/optimizer.rs:
##########
@@ -145,11 +146,9 @@ impl PhysicalOptimizer {
// are not present, the load of executors such as join or union will be
// reduced by narrowing their input tables.
Arc::new(ProjectionPushdown::new()),
- Arc::new(EnsureCooperative::new()),
- // This FilterPushdown handles dynamic filters that may have references to the source ExecutionPlan.
- // Therefore it should be run at the end of the optimization process since any changes to the plan may break the dynamic filter's references.
- // See `FilterPushdownPhase` for more details.
- Arc::new(FilterPushdown::new_post_optimization()),
+ // ReverseOrder: Detect DESC sorts that can use reverse scan
+ // This marks reverse_scan=true on DataSourceExec
+ Arc::new(PushdownSort::new()),
// The SanityCheckPlan rule checks whether the order and
Review Comment:
this comment I think needs to be moved down too (it is no longer with the sanity check plan implementation)
##########
datafusion/datasource/src/file.rs:
##########
@@ -126,6 +127,29 @@ pub trait FileSource: Send + Sync {
))
}
+ /// Try to create a new FileSource that can produce data in the specified sort order.
+ ///
+ /// This allows file format implementations to optimize based on the required sort order.
Review Comment:
The ability to push down sorts into the file source has come up before, specifically
- https://github.com/apache/datafusion/issues/10433
##########
datafusion/physical-optimizer/src/pushdown_sort.rs:
##########
@@ -0,0 +1,1235 @@
+// 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.
+
+//! Sort Pushdown Optimization
+//!
+//! This optimizer attempts to push sort requirements down to data sources that can
+//! satisfy them natively, avoiding expensive sort operations.
+//!
+//! Currently supported optimizations:
+//! - **Reverse scan**: If a data source naturally produces data in DESC order and
+//! we need ASC (or vice versa), we can reverse the scan direction instead of
+//! adding a SortExec node.
+//!
+//! Future optimizations could include:
+//! - Reordering row groups in Parquet files
+//! - Leveraging native indexes
+//! - Reordering files in multi-file scans
+use crate::PhysicalOptimizerRule;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion_common::Result;
+use datafusion_datasource::source::DataSourceExec;
+use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
+use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
+use datafusion_physical_plan::sorts::sort::SortExec;
+use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+use std::sync::Arc;
+
+/// A PhysicalOptimizerRule that attempts to push down sort requirements to data sources
+/// that can natively handle them (e.g., by reversing scan direction).
+///
+/// This optimization:
+/// 1. Detects SortExec nodes that require a specific ordering
+/// 2. Checks if the input can satisfy the ordering by reversing its scan direction
+/// 3. Pushes the sort requirement down to the data source when possible
+/// 4. Removes unnecessary sort operations when the input already satisfies the requirement
+#[derive(Debug, Clone, Default)]
+pub struct PushdownSort;
+
+impl PushdownSort {
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for PushdownSort {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ // Check if sort pushdown optimization is enabled
+ let enable_sort_pushdown = config.execution.parquet.enable_sort_pushdown;
+
+ // Return early if not enabled
+ if !enable_sort_pushdown {
+ return Ok(plan);
+ }
+
+ // Search for any SortExec nodes and try to optimize them
+ plan.transform_up(&|plan: Arc<dyn ExecutionPlan>| {
+ // First check if this is a GlobalLimitExec -> SortExec pattern
+ if let Some(limit_exec) = plan.as_any().downcast_ref::<GlobalLimitExec>() {
+ if let Some(sort_exec) =
+ limit_exec.input().as_any().downcast_ref::<SortExec>()
+ {
+ return optimize_limit_sort(limit_exec, sort_exec);
+ }
+ }
+
+ // Otherwise, check if this is just a SortExec
+ let sort_exec = match plan.as_any().downcast_ref::<SortExec>() {
+ Some(sort_exec) => sort_exec,
+ None => return Ok(Transformed::no(plan)),
+ };
+
+ optimize_sort(sort_exec)
+ })
+ .data()
+ }
+
+ fn name(&self) -> &str {
+ "PushdownSort"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// Optimize a SortExec by potentially pushing the sort down to the data source
+fn optimize_sort(sort_exec: &SortExec) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+ let required_ordering = sort_exec.expr();
+
+ // First, check if the sort is already satisfied by input ordering
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+
+ if input_eq_properties.ordering_satisfy(required_ordering.clone())? {
+ return remove_unnecessary_sort(sort_exec, sort_input);
+ }
+ }
+
+ // Try to push the sort requirement down to the data source
+ if let Some(optimized_input) = try_pushdown_sort(&sort_input, required_ordering)? {
+ // Verify that the optimized input satisfies the required ordering
+ if optimized_input
+ .equivalence_properties()
+ .ordering_satisfy(required_ordering.clone())?
+ {
+ return remove_unnecessary_sort(sort_exec, optimized_input);
+ }
+
+ // If not fully satisfied, keep the sort but with optimized input
+ return Ok(Transformed::yes(Arc::new(
+ SortExec::new(required_ordering.clone(), optimized_input)
+ .with_fetch(sort_exec.fetch())
+ .with_preserve_partitioning(sort_exec.preserve_partitioning()),
+ )));
+ }
+
+ Ok(Transformed::no(Arc::new(sort_exec.clone())))
+}
+
+/// Handle the GlobalLimitExec -> SortExec pattern
+fn optimize_limit_sort(
+ limit_exec: &GlobalLimitExec,
+ sort_exec: &SortExec,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+ let required_ordering = sort_exec.expr();
+
+ // Check if input is already sorted
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+ if input_eq_properties.ordering_satisfy(required_ordering.clone())? {
+ // Input is already sorted correctly, remove sort and keep limit
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ sort_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Try to push down the sort requirement
+ if let Some(optimized_input) = try_pushdown_sort(&sort_input, required_ordering)? {
+ if optimized_input
+ .equivalence_properties()
+ .ordering_satisfy(required_ordering.clone())?
+ {
+ // Successfully pushed down sort, now handle the limit
+ let total_fetch = limit_exec.skip() + limit_exec.fetch().unwrap_or(0);
+
+ // Try to push limit down as well if the source supports it
+ if let Some(with_fetch) = optimized_input.with_fetch(Some(total_fetch)) {
+ if limit_exec.skip() > 0 {
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ with_fetch,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ } else {
+ return Ok(Transformed::yes(with_fetch));
+ }
+ }
+
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ optimized_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Can't optimize, return original pattern
+ Ok(Transformed::no(Arc::new(GlobalLimitExec::new(
+ Arc::new(sort_exec.clone()),
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))))
+}
+
+/// Remove unnecessary sort based on the logic from EnforceSorting::analyze_immediate_sort_removal
+fn remove_unnecessary_sort(
+ sort_exec: &SortExec,
+ sort_input: Arc<dyn ExecutionPlan>,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let new_plan = if !sort_exec.preserve_partitioning()
+ && sort_input.output_partitioning().partition_count() > 1
+ {
+ // Replace the sort with a sort-preserving merge
+ Arc::new(
+ SortPreservingMergeExec::new(sort_exec.expr().clone(), sort_input)
+ .with_fetch(sort_exec.fetch()),
+ ) as _
+ } else {
+ // Remove the sort entirely
+ if let Some(fetch) = sort_exec.fetch() {
+ // If the sort has a fetch, add a limit instead
+ if sort_input.output_partitioning().partition_count() == 1 {
+ // Try to push the limit down to the source
+ if let Some(with_fetch) = sort_input.with_fetch(Some(fetch)) {
+ return Ok(Transformed::yes(with_fetch));
+ }
+ Arc::new(GlobalLimitExec::new(sort_input, 0, Some(fetch)))
+ as Arc<dyn ExecutionPlan>
+ } else {
+ Arc::new(LocalLimitExec::new(sort_input, fetch)) as Arc<dyn ExecutionPlan>
+ }
+ } else {
+ sort_input
+ }
+ };
+
+ Ok(Transformed::yes(new_plan))
+}
+
+/// Try to push down a sort requirement to an execution plan
+fn try_pushdown_sort(
+ plan: &Arc<dyn ExecutionPlan>,
+ required_ordering: &[PhysicalSortExpr],
+) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ // Check if the plan can natively handle the sort requirement
+ if let Some(data_source_exec) = plan.as_any().downcast_ref::<DataSourceExec>() {
+ return data_source_exec.try_pushdown_sort(required_ordering);
+ }
+
+ Ok(None)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow::array::{Int32Array, RecordBatch};
+ use arrow::compute::SortOptions;
+ use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+ use datafusion_physical_expr::expressions::Column;
+ use datafusion_physical_expr::Partitioning;
+ use datafusion_physical_expr_common::sort_expr::LexOrdering;
+ use datafusion_physical_plan::empty::EmptyExec;
+ use datafusion_physical_plan::memory::LazyMemoryExec;
+ use datafusion_physical_plan::repartition::RepartitionExec;
+ use parking_lot::RwLock;
+ use std::fmt;
+
+ /// Helper function to create a test schema with three columns
+ fn create_test_schema() -> Arc<Schema> {
+ Arc::new(Schema::new(vec![
+ Field::new("col1", DataType::Int32, true),
+ Field::new("col2", DataType::Utf8, true),
+ Field::new("col3", DataType::Float64, true),
+ ]))
+ }
+
+ /// Test batch generator that produces sorted data
+ #[derive(Debug, Clone)]
+ struct SortedBatchGenerator {
+ schema: SchemaRef,
+ batches_generated: usize,
+ max_batches: usize,
+ ascending: bool,
+ }
+
+ impl fmt::Display for SortedBatchGenerator {
+ fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+ write!(
+ f,
+ "SortedBatchGenerator(batches_generated={}, max_batches={}, ascending={})",
+ self.batches_generated, self.max_batches, self.ascending
+ )
+ }
+ }
+
+ impl datafusion_physical_plan::memory::LazyBatchGenerator for SortedBatchGenerator {
Review Comment:
most other optimizer tests don't actually run plans, instead they construct a plan, run the rule, and verify the transformation is as expected.
For example, here are the tests for filter_pushdown (note this is in `core/tests`)
https://github.com/apache/datafusion/blob/main/datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs
##########
datafusion/physical-optimizer/src/pushdown_sort.rs:
##########
@@ -0,0 +1,1446 @@
+// 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 crate::PhysicalOptimizerRule;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{
+ Transformed, TransformedResult, TreeNode, TreeNodeRecursion, TreeNodeRewriter,
+};
+use datafusion_common::Result;
+use datafusion_datasource::file::FileSource;
+use datafusion_datasource::file_scan_config::{FileScanConfig, FileScanConfigBuilder};
+use datafusion_datasource::source::{DataSource, DataSourceExec};
+use datafusion_datasource_parquet::source::ParquetSource;
+use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr};
+use datafusion_physical_plan::joins::SortMergeJoinExec;
+use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
+use datafusion_physical_plan::sorts::sort::SortExec;
+use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+use std::sync::Arc;
+
+/// A PhysicalOptimizerRule that attempts to reverse a SortExec's input if doing so would make the
+/// input ordering compatible with the SortExec's required output ordering.
+/// It also removes unnecessary sorts when the input already satisfies the required ordering.
+#[derive(Debug, Clone, Default)]
+pub struct ReverseOrder;
+
+impl ReverseOrder {
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for ReverseOrder {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ // Check if reverse scan optimization is enabled
+ let enable_reverse_scan = config.execution.parquet.enable_reverse_scan;
+
+ // Return early if not enabled
+ if !enable_reverse_scan {
+ return Ok(plan);
+ }
+
+ // Search for any SortExec nodes and try to optimize them
+ plan.transform_up(&|plan: Arc<dyn ExecutionPlan>| {
+ // First check if this is a GlobalLimitExec -> SortExec pattern
+ if let Some(limit_exec) = plan.as_any().downcast_ref::<GlobalLimitExec>() {
+ if let Some(sort_exec) =
+ limit_exec.input().as_any().downcast_ref::<SortExec>()
+ {
+ return optimize_limit_sort(limit_exec, sort_exec);
+ }
+ }
+
+ // Otherwise, check if this is just a SortExec
+ let sort_exec = match plan.as_any().downcast_ref::<SortExec>() {
+ Some(sort_exec) => sort_exec,
+ None => return Ok(Transformed::no(plan)),
+ };
+
+ let sort_input: Arc<dyn ExecutionPlan> = Arc::clone(sort_exec.input());
+
+ // First, check if the sort is already satisfied by input ordering
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+
+ // Check if input already satisfies the sort requirement
+ if input_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ return remove_unnecessary_sort(sort_exec, sort_input);
+ }
+ }
+
+ // If not satisfied, try to reverse the input
+ let reversed_eq_properties = {
+ let mut new = sort_input.properties().equivalence_properties().clone();
+ new.clear_orderings();
+
+ // Build reversed sort exprs for each ordering class
+ let reversed_orderings = sort_input
+ .equivalence_properties()
+ .oeq_class()
+ .iter()
+ .map(|ordering| {
+ ordering
+ .iter()
+ .map(|expr| expr.reverse())
+ .collect::<Vec<_>>()
+ })
+ .collect::<Vec<_>>();
+
+ new.add_orderings(reversed_orderings);
+ new
+ };
+
+ match reversed_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ true => {
+ // Reverse the input and then remove the sort
+ let reversed_input =
+ sort_input.rewrite(&mut ReverseRewriter).unwrap().data;
+
+ // After reversing, check if we can remove the sort
+ if reversed_input
+ .equivalence_properties()
+ .ordering_satisfy(sort_exec.expr().clone())?
+ {
+ remove_unnecessary_sort(sort_exec, reversed_input)
+ } else {
+ // Keep the sort but with reversed input
+ Ok(Transformed::yes(Arc::new(
+ SortExec::new(sort_exec.expr().clone(), reversed_input)
+ .with_fetch(sort_exec.fetch())
+ .with_preserve_partitioning(
+ sort_exec.preserve_partitioning(),
+ ),
+ )))
+ }
+ }
+ false => Ok(Transformed::no(plan)),
+ }
+ })
+ .data()
+ }
+
+ fn name(&self) -> &str {
+ "ReverseOrder"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// Handle the GlobalLimitExec -> SortExec pattern
+fn optimize_limit_sort(
+ limit_exec: &GlobalLimitExec,
+ sort_exec: &SortExec,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+
+ // Check if input is already sorted
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+ if input_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ // Input is already sorted correctly, remove sort and keep limit
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ sort_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Check if we can reverse the input to satisfy the sort
+ let reversed_eq_properties = {
+ let mut new = sort_input.properties().equivalence_properties().clone();
+ new.clear_orderings();
+
+ let reversed_orderings = sort_input
+ .equivalence_properties()
+ .oeq_class()
+ .iter()
+ .map(|ordering| {
+ ordering
+ .iter()
+ .map(|expr| expr.reverse())
+ .collect::<Vec<_>>()
+ })
+ .collect::<Vec<_>>();
+
+ new.add_orderings(reversed_orderings);
+ new
+ };
+
+ if reversed_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ // Can reverse! Apply reversal
+ let reversed_input = sort_input.rewrite(&mut ReverseRewriter).unwrap().data;
+
+ // Check if reversed input satisfies the sort requirement
+ if reversed_input
+ .equivalence_properties()
+ .ordering_satisfy(sort_exec.expr().clone())?
+ {
+ // Check if this is a single-partition DataSourceExec with reverse_scan enabled
+ // In that case, the limit is already handled internally by ReversedParquetStreamWithLimit
+ if is_single_partition_reverse_scan_datasource(&reversed_input) {
+ let total_fetch = limit_exec.skip() + limit_exec.fetch().unwrap_or(0);
+
+ if let Some(with_fetch) = reversed_input.with_fetch(Some(total_fetch)) {
+ if limit_exec.skip() > 0 {
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ with_fetch,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ } else {
+ return Ok(Transformed::yes(with_fetch));
+ }
+ }
+
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ reversed_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+
+ // Otherwise, remove sort but keep limit with reversed input
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ reversed_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Can't optimize, return original pattern
+ Ok(Transformed::no(Arc::new(GlobalLimitExec::new(
+ Arc::new(sort_exec.clone()),
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))))
+}
+
+/// Check if the plan is a single-partition DataSourceExec with reverse_scan enabled
+fn is_single_partition_reverse_scan_datasource(plan: &Arc<dyn ExecutionPlan>) -> bool {
+ // Only optimize for single partition
+ if plan.output_partitioning().partition_count() != 1 {
+ return false;
+ }
+
+ if let Some(data_source_exec) = plan.as_any().downcast_ref::<DataSourceExec>() {
+ if let Some(scan_config) = data_source_exec
+ .data_source()
+ .as_any()
+ .downcast_ref::<FileScanConfig>()
+ {
+ if let Some(parquet_source) = scan_config
+ .file_source
+ .as_any()
+ .downcast_ref::<ParquetSource>()
+ {
+ return parquet_source.reverse_scan();
+ }
+ }
+ }
+ false
+}
+
+/// Remove unnecessary sort based on the logic from EnforceSorting::analyze_immediate_sort_removal
+fn remove_unnecessary_sort(
Review Comment:
`enforce_sorting` is a somewhat triky tricky rule because it does both:
1. Adds new `SortExec`s to ensure that the `ExecutionPlan::input_requirements` are met
2. Pushes down / eliminates unecessary sorts
I think it would be a much cleaner design (though it would be a **major** refactoring) if we split that up into two passes -- add necessary sorts and then optimizations to remove / push sorts
##########
datafusion/physical-optimizer/src/pushdown_sort.rs:
##########
@@ -0,0 +1,1235 @@
+// 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.
+
+//! Sort Pushdown Optimization
+//!
+//! This optimizer attempts to push sort requirements down to data sources that can
+//! satisfy them natively, avoiding expensive sort operations.
+//!
+//! Currently supported optimizations:
+//! - **Reverse scan**: If a data source naturally produces data in DESC order and
+//! we need ASC (or vice versa), we can reverse the scan direction instead of
+//! adding a SortExec node.
+//!
+//! Future optimizations could include:
+//! - Reordering row groups in Parquet files
+//! - Leveraging native indexes
+//! - Reordering files in multi-file scans
+use crate::PhysicalOptimizerRule;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion_common::Result;
+use datafusion_datasource::source::DataSourceExec;
+use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
+use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
+use datafusion_physical_plan::sorts::sort::SortExec;
+use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+use std::sync::Arc;
+
+/// A PhysicalOptimizerRule that attempts to push down sort requirements to data sources
+/// that can natively handle them (e.g., by reversing scan direction).
+///
+/// This optimization:
+/// 1. Detects SortExec nodes that require a specific ordering
+/// 2. Checks if the input can satisfy the ordering by reversing its scan direction
+/// 3. Pushes the sort requirement down to the data source when possible
+/// 4. Removes unnecessary sort operations when the input already satisfies the requirement
+#[derive(Debug, Clone, Default)]
+pub struct PushdownSort;
+
+impl PushdownSort {
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for PushdownSort {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ // Check if sort pushdown optimization is enabled
+ let enable_sort_pushdown = config.execution.parquet.enable_sort_pushdown;
+
+ // Return early if not enabled
+ if !enable_sort_pushdown {
+ return Ok(plan);
+ }
+
+ // Search for any SortExec nodes and try to optimize them
+ plan.transform_up(&|plan: Arc<dyn ExecutionPlan>| {
Review Comment:
should this be `transform_down` so it pushes the sorts down (rather than up)?
Or is the expectation that this is only going to work on sorts that are directly above the datasource?
##########
datafusion/physical-optimizer/Cargo.toml:
##########
@@ -43,6 +43,7 @@ recursive_protection = ["dep:recursive"]
[dependencies]
arrow = { workspace = true }
datafusion-common = { workspace = true }
+datafusion-datasource = { workspace = true }
Review Comment:
Do we really need to add this new dependncy? I think if you made the optimizer tests look more like the other optimizzer tests this wouldn't be needed
##########
datafusion/datasource/src/file_scan_config.rs:
##########
@@ -718,6 +717,101 @@ impl DataSource for FileScanConfig {
}
}
}
+
+ fn try_pushdown_sort(
+ &self,
+ order: &[PhysicalSortExpr],
+ ) -> Result<Option<Arc<dyn DataSource>>> {
+ let current_ordering = match self.output_ordering.first() {
+ Some(ordering) => ordering.as_ref(),
+ None => return Ok(None),
+ };
+
+ // Only support reverse ordering pushdown
+ if !is_reverse_ordering(order, current_ordering) {
+ return Ok(None);
+ }
+
+ // Ask the file source if it can handle the sort pushdown
+ // (e.g., ParquetSource will enable reverse_scan)
+ let new_file_source = match self.file_source.try_pushdown_sort(order)? {
+ Some(source) => source,
+ None => return Ok(None),
+ };
+
+ let mut new_config = self.clone();
+
+ // Reverse file groups: when scanning in reverse, we need to read files
+ // in reverse order to maintain the correct global ordering
+ new_config.file_groups = new_config
+ .file_groups
+ .into_iter()
+ .map(|group| {
+ let mut files = group.into_inner();
+ files.reverse();
+ files.into()
+ })
+ .collect();
+
+ // Build the new output ordering by reversing each sort expression's direction
+ // E.g., [number DESC] becomes [number ASC]
+ let mut reversed_ordering = Vec::new();
+ for sort_expr in current_ordering {
+ reversed_ordering.push(PhysicalSortExpr {
+ expr: Arc::clone(&sort_expr.expr),
+ options: !sort_expr.options,
+ });
+ }
+
+ new_config.output_ordering = vec![LexOrdering::new(reversed_ordering)
+ .ok_or_else(|| {
+ DataFusionError::Plan(
+ "Failed to create ordering: invalid sort expressions".to_string(),
+ )
+ })?];
+
+ new_config.file_source = new_file_source;
+
+ Ok(Some(Arc::new(new_config)))
+ }
+}
+
+/// Check if the requested ordering can be satisfied by reversing the current ordering.
+///
+/// This function supports **prefix matching**: if the file has ordering [A DESC, B ASC]
+/// and we need [A ASC], reversing the scan gives us [A ASC, B DESC], which satisfies
+/// the requirement since [A ASC] is a prefix.
+///
+/// # Arguments
+/// * `requested` - The ordering required by the query
+/// * `current` - The natural ordering of the data source (e.g., from file metadata)
+///
+/// # Returns
+/// `true` if reversing the current ordering would satisfy the requested ordering
+///
+/// # Example
+/// ```text
+/// Current: [number DESC, letter ASC]
+/// Requested: [number ASC]
+/// Reversed: [number ASC, letter DESC] โ Prefix match!
+/// ```
+fn is_reverse_ordering(
+ requested: &[PhysicalSortExpr],
+ current: &[PhysicalSortExpr],
+) -> bool {
+ // Allow prefix matching - we can satisfy a prefix of the current ordering
+ // by reversing the scan
+ if requested.len() > current.len() {
+ return false;
+ }
+
+ requested.iter().zip(current.iter()).all(|(req, cur)| {
+ let exprs_match = req.expr.to_string() == cur.expr.to_string();
Review Comment:
is there a reason not to compare the sort exprs directly? I think converting them to a string may lose some information (like would aliased expressions appear equal when they really arent?)
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2569297154
##########
datafusion/physical-optimizer/src/pushdown_sort.rs:
##########
@@ -0,0 +1,1235 @@
+// 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.
+
+//! Sort Pushdown Optimization
+//!
+//! This optimizer attempts to push sort requirements down to data sources that can
+//! satisfy them natively, avoiding expensive sort operations.
+//!
+//! Currently supported optimizations:
+//! - **Reverse scan**: If a data source naturally produces data in DESC order and
+//! we need ASC (or vice versa), we can reverse the scan direction instead of
+//! adding a SortExec node.
+//!
+//! Future optimizations could include:
+//! - Reordering row groups in Parquet files
+//! - Leveraging native indexes
+//! - Reordering files in multi-file scans
+use crate::PhysicalOptimizerRule;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion_common::Result;
+use datafusion_datasource::source::DataSourceExec;
+use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
+use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
+use datafusion_physical_plan::sorts::sort::SortExec;
+use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+use std::sync::Arc;
+
+/// A PhysicalOptimizerRule that attempts to push down sort requirements to data sources
+/// that can natively handle them (e.g., by reversing scan direction).
+///
+/// This optimization:
+/// 1. Detects SortExec nodes that require a specific ordering
+/// 2. Checks if the input can satisfy the ordering by reversing its scan direction
+/// 3. Pushes the sort requirement down to the data source when possible
+/// 4. Removes unnecessary sort operations when the input already satisfies the requirement
+#[derive(Debug, Clone, Default)]
+pub struct PushdownSort;
+
+impl PushdownSort {
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for PushdownSort {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ // Check if sort pushdown optimization is enabled
+ let enable_sort_pushdown = config.execution.parquet.enable_sort_pushdown;
+
+ // Return early if not enabled
+ if !enable_sort_pushdown {
+ return Ok(plan);
+ }
+
+ // Search for any SortExec nodes and try to optimize them
+ plan.transform_up(&|plan: Arc<dyn ExecutionPlan>| {
Review Comment:
Thank you @alamb for good suggestion, i addressed this in latest PR, changing to transform_down.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2571766380
##########
datafusion/physical-optimizer/src/pushdown_sort.rs:
##########
@@ -0,0 +1,1446 @@
+// 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 crate::PhysicalOptimizerRule;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{
+ Transformed, TransformedResult, TreeNode, TreeNodeRecursion, TreeNodeRewriter,
+};
+use datafusion_common::Result;
+use datafusion_datasource::file::FileSource;
+use datafusion_datasource::file_scan_config::{FileScanConfig, FileScanConfigBuilder};
+use datafusion_datasource::source::{DataSource, DataSourceExec};
+use datafusion_datasource_parquet::source::ParquetSource;
+use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr};
+use datafusion_physical_plan::joins::SortMergeJoinExec;
+use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
+use datafusion_physical_plan::sorts::sort::SortExec;
+use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+use std::sync::Arc;
+
+/// A PhysicalOptimizerRule that attempts to reverse a SortExec's input if doing so would make the
+/// input ordering compatible with the SortExec's required output ordering.
+/// It also removes unnecessary sorts when the input already satisfies the required ordering.
+#[derive(Debug, Clone, Default)]
+pub struct ReverseOrder;
+
+impl ReverseOrder {
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for ReverseOrder {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ // Check if reverse scan optimization is enabled
+ let enable_reverse_scan = config.execution.parquet.enable_reverse_scan;
+
+ // Return early if not enabled
+ if !enable_reverse_scan {
+ return Ok(plan);
+ }
+
+ // Search for any SortExec nodes and try to optimize them
+ plan.transform_up(&|plan: Arc<dyn ExecutionPlan>| {
+ // First check if this is a GlobalLimitExec -> SortExec pattern
+ if let Some(limit_exec) = plan.as_any().downcast_ref::<GlobalLimitExec>() {
+ if let Some(sort_exec) =
+ limit_exec.input().as_any().downcast_ref::<SortExec>()
+ {
+ return optimize_limit_sort(limit_exec, sort_exec);
+ }
+ }
+
+ // Otherwise, check if this is just a SortExec
+ let sort_exec = match plan.as_any().downcast_ref::<SortExec>() {
+ Some(sort_exec) => sort_exec,
+ None => return Ok(Transformed::no(plan)),
+ };
+
+ let sort_input: Arc<dyn ExecutionPlan> = Arc::clone(sort_exec.input());
+
+ // First, check if the sort is already satisfied by input ordering
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+
+ // Check if input already satisfies the sort requirement
+ if input_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ return remove_unnecessary_sort(sort_exec, sort_input);
+ }
+ }
+
+ // If not satisfied, try to reverse the input
+ let reversed_eq_properties = {
+ let mut new = sort_input.properties().equivalence_properties().clone();
+ new.clear_orderings();
+
+ // Build reversed sort exprs for each ordering class
+ let reversed_orderings = sort_input
+ .equivalence_properties()
+ .oeq_class()
+ .iter()
+ .map(|ordering| {
+ ordering
+ .iter()
+ .map(|expr| expr.reverse())
+ .collect::<Vec<_>>()
+ })
+ .collect::<Vec<_>>();
+
+ new.add_orderings(reversed_orderings);
+ new
+ };
+
+ match reversed_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ true => {
+ // Reverse the input and then remove the sort
+ let reversed_input =
+ sort_input.rewrite(&mut ReverseRewriter).unwrap().data;
+
+ // After reversing, check if we can remove the sort
+ if reversed_input
+ .equivalence_properties()
+ .ordering_satisfy(sort_exec.expr().clone())?
+ {
+ remove_unnecessary_sort(sort_exec, reversed_input)
+ } else {
+ // Keep the sort but with reversed input
+ Ok(Transformed::yes(Arc::new(
+ SortExec::new(sort_exec.expr().clone(), reversed_input)
+ .with_fetch(sort_exec.fetch())
+ .with_preserve_partitioning(
+ sort_exec.preserve_partitioning(),
+ ),
+ )))
+ }
+ }
+ false => Ok(Transformed::no(plan)),
+ }
+ })
+ .data()
+ }
+
+ fn name(&self) -> &str {
+ "ReverseOrder"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// Handle the GlobalLimitExec -> SortExec pattern
+fn optimize_limit_sort(
+ limit_exec: &GlobalLimitExec,
+ sort_exec: &SortExec,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+
+ // Check if input is already sorted
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+ if input_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ // Input is already sorted correctly, remove sort and keep limit
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ sort_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Check if we can reverse the input to satisfy the sort
+ let reversed_eq_properties = {
+ let mut new = sort_input.properties().equivalence_properties().clone();
+ new.clear_orderings();
+
+ let reversed_orderings = sort_input
+ .equivalence_properties()
+ .oeq_class()
+ .iter()
+ .map(|ordering| {
+ ordering
+ .iter()
+ .map(|expr| expr.reverse())
+ .collect::<Vec<_>>()
+ })
+ .collect::<Vec<_>>();
+
+ new.add_orderings(reversed_orderings);
+ new
+ };
+
+ if reversed_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ // Can reverse! Apply reversal
+ let reversed_input = sort_input.rewrite(&mut ReverseRewriter).unwrap().data;
+
+ // Check if reversed input satisfies the sort requirement
+ if reversed_input
+ .equivalence_properties()
+ .ordering_satisfy(sort_exec.expr().clone())?
+ {
+ // Check if this is a single-partition DataSourceExec with reverse_scan enabled
+ // In that case, the limit is already handled internally by ReversedParquetStreamWithLimit
+ if is_single_partition_reverse_scan_datasource(&reversed_input) {
+ let total_fetch = limit_exec.skip() + limit_exec.fetch().unwrap_or(0);
+
+ if let Some(with_fetch) = reversed_input.with_fetch(Some(total_fetch)) {
+ if limit_exec.skip() > 0 {
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ with_fetch,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ } else {
+ return Ok(Transformed::yes(with_fetch));
+ }
+ }
+
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ reversed_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+
+ // Otherwise, remove sort but keep limit with reversed input
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ reversed_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Can't optimize, return original pattern
+ Ok(Transformed::no(Arc::new(GlobalLimitExec::new(
+ Arc::new(sort_exec.clone()),
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))))
+}
+
+/// Check if the plan is a single-partition DataSourceExec with reverse_scan enabled
+fn is_single_partition_reverse_scan_datasource(plan: &Arc<dyn ExecutionPlan>) -> bool {
+ // Only optimize for single partition
+ if plan.output_partitioning().partition_count() != 1 {
+ return false;
+ }
+
+ if let Some(data_source_exec) = plan.as_any().downcast_ref::<DataSourceExec>() {
+ if let Some(scan_config) = data_source_exec
+ .data_source()
+ .as_any()
+ .downcast_ref::<FileScanConfig>()
+ {
+ if let Some(parquet_source) = scan_config
+ .file_source
+ .as_any()
+ .downcast_ref::<ParquetSource>()
+ {
+ return parquet_source.reverse_scan();
+ }
+ }
+ }
+ false
+}
+
+/// Remove unnecessary sort based on the logic from EnforceSorting::analyze_immediate_sort_removal
+fn remove_unnecessary_sort(
Review Comment:
@alamb This is a good point, we can do this after this PR, and split the enforce_sorting to:
EnforceSortingRequirements and OptimizeSorts.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3609815706
> I plan to try and re-review this in detail tomorrow
> > Amazing @alamb , pretty cool! Appreciate to be added, thanks!
>
> Added in [alamb/datafusion-benchmarking@c9f25c4](https://github.com/alamb/datafusion-benchmarking/commit/c9f25c40ea6e51fafaf9ba8a9ffd1ee18255d146)
Thank you @alamb.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2569293810
##########
datafusion/physical-optimizer/Cargo.toml:
##########
@@ -43,6 +43,7 @@ recursive_protection = ["dep:recursive"]
[dependencies]
arrow = { workspace = true }
datafusion-common = { workspace = true }
+datafusion-datasource = { workspace = true }
Review Comment:
Thanks for good point, currently the optimizer will use datasource to pushdown, i will look back 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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3592611260
> run benchmarks
(BTW I am testing out some new scripts: https://github.com/apache/datafusion/issues/18115#issuecomment-3592609246)
@zhuqi-lucas / @xudong963 let me know if you would like to be added to the whitelist
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2576074919
##########
datafusion/datasource-parquet/src/opener.rs:
##########
@@ -752,10 +799,316 @@ fn should_enable_page_index(
.unwrap_or(false)
}
+fn reverse_batch(batch: RecordBatch) -> Result<RecordBatch> {
+ let num_rows = batch.num_rows();
+ if num_rows <= 1 {
+ return Ok(batch);
+ }
+
+ let indices = UInt32Array::from_iter_values((0..num_rows as u32).rev());
+
+ let reversed_columns = batch
+ .columns()
+ .iter()
+ .map(|col| take(col.as_ref(), &indices, None))
+ .collect::<std::result::Result<Vec<ArrayRef>, arrow::error::ArrowError>>()
+ .map_err(DataFusionError::from)?;
+
+ RecordBatch::try_new(batch.schema(), reversed_columns).map_err(DataFusionError::from)
+}
+
+/// Stream adapter for reversed parquet reading with row-group-level buffering.
+///
+/// # Architecture
+///
+/// This stream implements a sophisticated buffering strategy to achieve true reverse
+/// reading of Parquet files while maintaining compatibility with the underlying
+/// ParquetRecordBatchStream's optimizations (caching, prefetching, etc.).
+///
+/// ## Strategy Overview
+///
+/// 1. **Pre-reversed Row Groups**: Row groups are reversed BEFORE building the stream
+/// (via `row_group_indexes.reverse()`). This allows the Parquet reader to read
+/// them in reverse order while still utilizing internal optimizations.
+///
+/// 2. **Row-Group-Level Buffering**: As batches arrive from the input stream, we
+/// track which row group they belong to using cumulative row counts. This is
+/// the MINIMAL buffering unit required for correctness - we cannot reverse
+/// individual batches without knowing the complete row group context.
+///
+/// 3. **Two-Stage Reversal**: When a complete row group is collected:
+/// - Stage 1: Reverse rows within each batch (using Arrow's take kernel)
+/// - Stage 2: Reverse the order of batches within the row group
+///
+/// 4. **Progressive Output**: Reversed batches are output immediately, minimizing
+/// memory footprint. We never buffer more than one row group at a time.
+///
+/// ## Memory Characteristics
+///
+/// - **Bounded Memory**: Maximum memory usage = size of largest row group
+/// - **Typical Usage**: ~128MB (default Parquet row group size)
+/// - **Peak Usage**: During reversal of a single row group
+///
+/// ## Why Row-Group-Level Buffering is Necessary
+///
+/// Parquet organizes data into row groups (typically 128MB each), and each row group
+/// is independently compressed and encoded. When reading in reverse:
+///
+/// - We cannot reverse individual batches in isolation because they may span
+/// row group boundaries or be split arbitrarily by the batch_size parameter
+/// - We must buffer complete row groups to ensure correct ordering semantics
+/// - This is the minimum granularity that maintains correctness
+///
+/// ## Example
+///
+/// Given a file with 3 row groups, each containing 2 batches:
+///
+/// ```text
+/// Normal order: RG0[B0, B1] -> RG1[B0, B1] -> RG2[B0, B1]
+/// Reversed: RG2[B1_rev, B0_rev] -> RG1[B1_rev, B0_rev] -> RG0[B1_rev, B0_rev]
+/// ^^^^^^^^^^^^ ^^^^^^^^^^^^ ^^^^^^^^^^^^
+/// Output 1st Output 2nd Output 3rd
+/// ```
+///
+/// ## Performance Characteristics
+///
+/// - **Latency**: First batch available after reading complete first (reversed) row group
+/// - **Throughput**: Near-native speed with ~5-10% overhead for reversal operations
Review Comment:
How did you get the "~5-10% overhead"?
##########
datafusion/datasource-parquet/src/opener.rs:
##########
@@ -752,10 +799,316 @@ fn should_enable_page_index(
.unwrap_or(false)
}
+fn reverse_batch(batch: RecordBatch) -> Result<RecordBatch> {
+ let num_rows = batch.num_rows();
+ if num_rows <= 1 {
+ return Ok(batch);
+ }
+
+ let indices = UInt32Array::from_iter_values((0..num_rows as u32).rev());
+
+ let reversed_columns = batch
+ .columns()
+ .iter()
+ .map(|col| take(col.as_ref(), &indices, None))
+ .collect::<std::result::Result<Vec<ArrayRef>, arrow::error::ArrowError>>()
+ .map_err(DataFusionError::from)?;
+
+ RecordBatch::try_new(batch.schema(), reversed_columns).map_err(DataFusionError::from)
+}
+
+/// Stream adapter for reversed parquet reading with row-group-level buffering.
+///
+/// # Architecture
+///
+/// This stream implements a sophisticated buffering strategy to achieve true reverse
+/// reading of Parquet files while maintaining compatibility with the underlying
+/// ParquetRecordBatchStream's optimizations (caching, prefetching, etc.).
+///
+/// ## Strategy Overview
+///
+/// 1. **Pre-reversed Row Groups**: Row groups are reversed BEFORE building the stream
+/// (via `row_group_indexes.reverse()`). This allows the Parquet reader to read
+/// them in reverse order while still utilizing internal optimizations.
+///
+/// 2. **Row-Group-Level Buffering**: As batches arrive from the input stream, we
+/// track which row group they belong to using cumulative row counts. This is
+/// the MINIMAL buffering unit required for correctness - we cannot reverse
+/// individual batches without knowing the complete row group context.
+///
+/// 3. **Two-Stage Reversal**: When a complete row group is collected:
+/// - Stage 1: Reverse rows within each batch (using Arrow's take kernel)
+/// - Stage 2: Reverse the order of batches within the row group
+///
+/// 4. **Progressive Output**: Reversed batches are output immediately, minimizing
+/// memory footprint. We never buffer more than one row group at a time.
+///
+/// ## Memory Characteristics
+///
+/// - **Bounded Memory**: Maximum memory usage = size of largest row group
+/// - **Typical Usage**: ~128MB (default Parquet row group size)
+/// - **Peak Usage**: During reversal of a single row group
+///
+/// ## Why Row-Group-Level Buffering is Necessary
+///
+/// Parquet organizes data into row groups (typically 128MB each), and each row group
+/// is independently compressed and encoded. When reading in reverse:
+///
+/// - We cannot reverse individual batches in isolation because they may span
+/// row group boundaries or be split arbitrarily by the batch_size parameter
+/// - We must buffer complete row groups to ensure correct ordering semantics
+/// - This is the minimum granularity that maintains correctness
+///
+/// ## Example
+///
+/// Given a file with 3 row groups, each containing 2 batches:
+///
+/// ```text
+/// Normal order: RG0[B0, B1] -> RG1[B0, B1] -> RG2[B0, B1]
+/// Reversed: RG2[B1_rev, B0_rev] -> RG1[B1_rev, B0_rev] -> RG0[B1_rev, B0_rev]
+/// ^^^^^^^^^^^^ ^^^^^^^^^^^^ ^^^^^^^^^^^^
+/// Output 1st Output 2nd Output 3rd
+/// ```
+///
+/// ## Performance Characteristics
+///
+/// - **Latency**: First batch available after reading complete first (reversed) row group
+/// - **Throughput**: Near-native speed with ~5-10% overhead for reversal operations
+/// - **Memory**: O(row_group_size), not O(file_size)
+/// TODO should we support max cache size to limit memory usage further? But if we exceed the cache size we can't reverse properly, so we need to fall back to normal reading?
+struct ReversedParquetStream<S> {
+ input: S,
+ schema: SchemaRef,
+
+ // Optional limit on the number of rows to output
+ limit: Option<usize>,
+ rows_produced: usize,
+
+ // Current row group being processed
+ current_rg_batches: Vec<RecordBatch>,
+ current_rg_rows_read: usize,
+ current_rg_total_rows: usize,
+
+ // Output buffer for reversed batches
+ output_batches: Vec<RecordBatch>,
+ output_index: usize,
+
+ // Row group metadata (each element is the number of rows in that row group)
+ // Already in reverse order since row_group_indexes was reversed
+ row_group_metadata: Vec<usize>,
+ current_rg_index: usize,
+
+ done: bool,
+
+ // Metrics
+ row_groups_reversed: Count,
+ batches_reversed: Count,
+ reverse_time: Time,
+
+ /// Pending error from batch reversal
+ pending_error: Option<DataFusionError>,
+}
+
+impl<S> ReversedParquetStream<S> {
+ fn new(
+ stream: S,
+ schema: SchemaRef,
+ row_group_metadata: Vec<usize>,
+ row_groups_reversed: Count,
+ batches_reversed: Count,
+ reverse_time: Time,
+ limit: Option<usize>,
+ ) -> Self {
+ let current_rg_total_rows = row_group_metadata.first().copied().unwrap_or(0);
+
+ Self {
+ input: stream,
+ schema,
+ limit,
+ rows_produced: 0,
+ current_rg_batches: Vec::new(),
+ current_rg_rows_read: 0,
+ current_rg_total_rows,
+ output_batches: Vec::new(),
+ output_index: 0,
+ row_group_metadata,
+ current_rg_index: 0,
+ done: false,
+ row_groups_reversed,
+ batches_reversed,
+ reverse_time,
+ pending_error: None,
+ }
+ }
+
+ /// Finalizes the current row group by performing the two-stage reversal.
+ ///
+ /// This is called when we've accumulated all batches for a row group.
+ ///
+ /// # Two-Stage Reversal Process
+ ///
+ /// 1. **Stage 1 - Reverse Rows**: For each batch in the row group, reverse
+ /// the order of rows using Arrow's `take` kernel with reversed indices.
+ ///
+ /// 2. **Stage 2 - Reverse Batches**: Reverse the order of batches within
+ /// the row group so that the last batch becomes first.
+ ///
+ /// # Example
+ ///
+ /// Input batches for a row group:
+ /// ```text
+ /// B0: [row0, row1, row2]
+ /// B1: [row3, row4, row5]
+ /// ```
+ ///
+ /// After Stage 1 (reverse rows within each batch):
+ /// ```text
+ /// B0_rev: [row2, row1, row0]
+ /// B1_rev: [row5, row4, row3]
+ /// ```
+ ///
+ /// After Stage 2 (reverse batch order):
+ /// ```text
+ /// Output: B1_rev, B0_rev
+ /// Final sequence: [row5, row4, row3, row2, row1, row0]
+ /// ```
+ fn finalize_current_row_group(&mut self) {
+ if self.current_rg_batches.is_empty() {
+ return;
+ }
+
+ // Start timing
+ let _timer = self.reverse_time.timer();
+ let batch_count = self.current_rg_batches.len();
+
+ // Step 1: Reverse rows within each batch
+ let mut reversed_batches = Vec::with_capacity(self.current_rg_batches.len());
+ for batch in self.current_rg_batches.drain(..) {
+ match reverse_batch(batch) {
+ Ok(reversed) => reversed_batches.push(reversed),
+ Err(e) => {
+ // Store error and return it on next poll
+ self.pending_error = Some(e);
+ return;
+ }
+ }
+ }
+
+ // Step 2: Reverse the order of batches
+ self.output_batches = reversed_batches.into_iter().rev().collect();
+ self.output_index = 0;
+
+ // Update metrics
+ self.row_groups_reversed.add(1);
+ self.batches_reversed.add(batch_count);
+
+ // Prepare for next row group
+ self.current_rg_rows_read = 0;
+ self.current_rg_index += 1;
+ self.current_rg_total_rows = self
+ .row_group_metadata
+ .get(self.current_rg_index)
+ .copied()
+ .unwrap_or(0);
+ }
+
+ /// Check if we've reached the limit
+ #[inline]
+ fn is_limit_reached(&self) -> bool {
+ self.limit.is_some_and(|limit| self.rows_produced >= limit)
+ }
+}
+
+impl<S> Stream for ReversedParquetStream<S>
+where
+ S: Stream<Item = Result<RecordBatch>> + Unpin,
+{
+ type Item = Result<RecordBatch>;
+
+ fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ let this = self.get_mut(); // Safe: We own the Pin and ReversedParquetStream is Unpin
+
+ // Check for pending errors first
+ if let Some(err) = this.pending_error.take() {
+ return Poll::Ready(Some(Err(err)));
+ }
+
+ if this.done || this.is_limit_reached() {
+ return Poll::Ready(None);
+ }
+
+ loop {
+ // First, output any ready batches
+ if this.output_index < this.output_batches.len() {
+ let batch = this.output_batches[this.output_index].clone();
+ this.output_index += 1;
+
+ // Apply limit if specified
+ if let Some(limit) = this.limit {
+ let remaining = limit.saturating_sub(this.rows_produced);
+
+ if batch.num_rows() <= remaining {
+ this.rows_produced += batch.num_rows();
+ return Poll::Ready(Some(Ok(batch)));
+ } else {
+ // Slice batch to fit within limit
+ let sliced = batch.slice(0, remaining);
+ this.rows_produced += remaining;
+ this.done = true;
+ return Poll::Ready(Some(Ok(sliced)));
+ }
+ } else {
+ // No limit, return full batch
+ return Poll::Ready(Some(Ok(batch)));
+ }
+ }
+
+ // Need to read more data
+ match ready!(Pin::new(&mut this.input).poll_next(cx)) {
+ Some(Ok(batch)) => {
+ let batch_rows = batch.num_rows();
+ this.current_rg_batches.push(batch);
+ this.current_rg_rows_read += batch_rows;
+
+ // Check if current row group is complete
+ if this.current_rg_rows_read >= this.current_rg_total_rows {
+ this.finalize_current_row_group();
Review Comment:
When `finalize_current_row_group()` encounters an error (e.g., during `reverse_batch`), it sets `self.pending_error` and returns early. However, it is called inside the `loop` in `poll_next`. Here does NOT check if pending_error was set.
##########
datafusion/datasource-parquet/src/opener.rs:
##########
@@ -752,10 +799,316 @@ fn should_enable_page_index(
.unwrap_or(false)
}
+fn reverse_batch(batch: RecordBatch) -> Result<RecordBatch> {
+ let num_rows = batch.num_rows();
+ if num_rows <= 1 {
+ return Ok(batch);
+ }
+
+ let indices = UInt32Array::from_iter_values((0..num_rows as u32).rev());
+
+ let reversed_columns = batch
+ .columns()
+ .iter()
+ .map(|col| take(col.as_ref(), &indices, None))
+ .collect::<std::result::Result<Vec<ArrayRef>, arrow::error::ArrowError>>()
+ .map_err(DataFusionError::from)?;
+
+ RecordBatch::try_new(batch.schema(), reversed_columns).map_err(DataFusionError::from)
+}
+
+/// Stream adapter for reversed parquet reading with row-group-level buffering.
+///
+/// # Architecture
+///
+/// This stream implements a sophisticated buffering strategy to achieve true reverse
+/// reading of Parquet files while maintaining compatibility with the underlying
+/// ParquetRecordBatchStream's optimizations (caching, prefetching, etc.).
+///
+/// ## Strategy Overview
+///
+/// 1. **Pre-reversed Row Groups**: Row groups are reversed BEFORE building the stream
+/// (via `row_group_indexes.reverse()`). This allows the Parquet reader to read
+/// them in reverse order while still utilizing internal optimizations.
+///
+/// 2. **Row-Group-Level Buffering**: As batches arrive from the input stream, we
+/// track which row group they belong to using cumulative row counts. This is
+/// the MINIMAL buffering unit required for correctness - we cannot reverse
+/// individual batches without knowing the complete row group context.
+///
+/// 3. **Two-Stage Reversal**: When a complete row group is collected:
+/// - Stage 1: Reverse rows within each batch (using Arrow's take kernel)
+/// - Stage 2: Reverse the order of batches within the row group
+///
+/// 4. **Progressive Output**: Reversed batches are output immediately, minimizing
+/// memory footprint. We never buffer more than one row group at a time.
+///
+/// ## Memory Characteristics
+///
+/// - **Bounded Memory**: Maximum memory usage = size of largest row group
+/// - **Typical Usage**: ~128MB (default Parquet row group size)
+/// - **Peak Usage**: During reversal of a single row group
+///
+/// ## Why Row-Group-Level Buffering is Necessary
+///
+/// Parquet organizes data into row groups (typically 128MB each), and each row group
+/// is independently compressed and encoded. When reading in reverse:
+///
+/// - We cannot reverse individual batches in isolation because they may span
+/// row group boundaries or be split arbitrarily by the batch_size parameter
+/// - We must buffer complete row groups to ensure correct ordering semantics
+/// - This is the minimum granularity that maintains correctness
+///
+/// ## Example
+///
+/// Given a file with 3 row groups, each containing 2 batches:
+///
+/// ```text
+/// Normal order: RG0[B0, B1] -> RG1[B0, B1] -> RG2[B0, B1]
+/// Reversed: RG2[B1_rev, B0_rev] -> RG1[B1_rev, B0_rev] -> RG0[B1_rev, B0_rev]
+/// ^^^^^^^^^^^^ ^^^^^^^^^^^^ ^^^^^^^^^^^^
+/// Output 1st Output 2nd Output 3rd
+/// ```
+///
+/// ## Performance Characteristics
+///
+/// - **Latency**: First batch available after reading complete first (reversed) row group
+/// - **Throughput**: Near-native speed with ~5-10% overhead for reversal operations
+/// - **Memory**: O(row_group_size), not O(file_size)
Review Comment:
I'm a bit confused about the line, what does it mean?
##########
datafusion/datasource/src/file_scan_config.rs:
##########
@@ -766,6 +766,131 @@ impl DataSource for FileScanConfig {
}
}
}
+
+ fn try_pushdown_sort(
+ &self,
+ order: &[PhysicalSortExpr],
+ ) -> Result<Option<Arc<dyn DataSource>>> {
+ let current_ordering = match self.output_ordering.first() {
+ Some(ordering) => ordering.as_ref(),
+ None => return Ok(None),
+ };
+
+ // Only support reverse ordering pushdown
+ if !is_reverse_ordering(order, current_ordering) {
+ return Ok(None);
+ }
+
+ // Ask the file source if it can handle the sort pushdown
+ // (e.g., ParquetSource will enable reverse_scan)
+ let new_file_source = match self.file_source.try_pushdown_sort(order)? {
+ Some(source) => source,
+ None => return Ok(None),
+ };
+
+ let mut new_config = self.clone();
+
+ // Reverse file groups: when scanning in reverse, we need to read files
+ // in reverse order to maintain the correct global ordering
+ new_config.file_groups = new_config
+ .file_groups
+ .into_iter()
+ .map(|group| {
+ let mut files = group.into_inner();
+ files.reverse();
+ files.into()
+ })
+ .collect();
+
+ // Build the new output ordering by reversing each sort expression's direction
+ // E.g., [number DESC] becomes [number ASC]
+ let mut reversed_ordering = Vec::new();
+ for sort_expr in current_ordering {
+ reversed_ordering.push(PhysicalSortExpr {
+ expr: Arc::clone(&sort_expr.expr),
+ options: !sort_expr.options,
+ });
+ }
+
+ new_config.output_ordering = vec![LexOrdering::new(reversed_ordering)
+ .ok_or_else(|| {
+ DataFusionError::Plan(
+ "Failed to create ordering: invalid sort expressions".to_string(),
+ )
+ })?];
+
+ new_config.file_source = new_file_source;
+
+ Ok(Some(Arc::new(new_config)))
+ }
+}
+
+/// Check if the requested ordering can be satisfied by reversing the current ordering.
+///
+/// This function supports **prefix matching**: if the file has ordering [A DESC, B ASC]
+/// and we need [A ASC], reversing the scan gives us [A ASC, B DESC], which satisfies
+/// the requirement since [A ASC] is a prefix.
+///
+/// # Arguments
+/// * `requested` - The ordering required by the query
+/// * `current` - The natural ordering of the data source (e.g., from file metadata)
+///
+/// # Returns
+/// `true` if reversing the current ordering would satisfy the requested ordering
+///
+/// # Example
+/// ```text
+/// Current: [number DESC, letter ASC]
+/// Requested: [number ASC]
+/// Reversed: [number ASC, letter DESC] โ Prefix match!
+/// ```
+fn is_reverse_ordering(
Review Comment:
I'm wondering if we could leverage the ordering equivalence to do the check not by string match.
##########
datafusion/datasource-parquet/src/opener.rs:
##########
@@ -752,10 +799,316 @@ fn should_enable_page_index(
.unwrap_or(false)
}
+fn reverse_batch(batch: RecordBatch) -> Result<RecordBatch> {
+ let num_rows = batch.num_rows();
+ if num_rows <= 1 {
+ return Ok(batch);
+ }
+
+ let indices = UInt32Array::from_iter_values((0..num_rows as u32).rev());
Review Comment:
The `UInt32Array` is allocated for every batch, not sure if it could be a potential cost, if it is, a cache may be good
##########
datafusion/datasource-parquet/src/opener.rs:
##########
@@ -752,10 +799,316 @@ fn should_enable_page_index(
.unwrap_or(false)
}
+fn reverse_batch(batch: RecordBatch) -> Result<RecordBatch> {
+ let num_rows = batch.num_rows();
+ if num_rows <= 1 {
+ return Ok(batch);
+ }
+
+ let indices = UInt32Array::from_iter_values((0..num_rows as u32).rev());
+
+ let reversed_columns = batch
+ .columns()
+ .iter()
+ .map(|col| take(col.as_ref(), &indices, None))
+ .collect::<std::result::Result<Vec<ArrayRef>, arrow::error::ArrowError>>()
+ .map_err(DataFusionError::from)?;
+
+ RecordBatch::try_new(batch.schema(), reversed_columns).map_err(DataFusionError::from)
+}
+
+/// Stream adapter for reversed parquet reading with row-group-level buffering.
+///
+/// # Architecture
+///
+/// This stream implements a sophisticated buffering strategy to achieve true reverse
+/// reading of Parquet files while maintaining compatibility with the underlying
+/// ParquetRecordBatchStream's optimizations (caching, prefetching, etc.).
+///
+/// ## Strategy Overview
+///
+/// 1. **Pre-reversed Row Groups**: Row groups are reversed BEFORE building the stream
+/// (via `row_group_indexes.reverse()`). This allows the Parquet reader to read
+/// them in reverse order while still utilizing internal optimizations.
+///
+/// 2. **Row-Group-Level Buffering**: As batches arrive from the input stream, we
+/// track which row group they belong to using cumulative row counts. This is
+/// the MINIMAL buffering unit required for correctness - we cannot reverse
+/// individual batches without knowing the complete row group context.
+///
+/// 3. **Two-Stage Reversal**: When a complete row group is collected:
+/// - Stage 1: Reverse rows within each batch (using Arrow's take kernel)
+/// - Stage 2: Reverse the order of batches within the row group
+///
+/// 4. **Progressive Output**: Reversed batches are output immediately, minimizing
+/// memory footprint. We never buffer more than one row group at a time.
+///
+/// ## Memory Characteristics
+///
+/// - **Bounded Memory**: Maximum memory usage = size of largest row group
+/// - **Typical Usage**: ~128MB (default Parquet row group size)
+/// - **Peak Usage**: During reversal of a single row group
+///
+/// ## Why Row-Group-Level Buffering is Necessary
+///
+/// Parquet organizes data into row groups (typically 128MB each), and each row group
+/// is independently compressed and encoded. When reading in reverse:
+///
+/// - We cannot reverse individual batches in isolation because they may span
+/// row group boundaries or be split arbitrarily by the batch_size parameter
+/// - We must buffer complete row groups to ensure correct ordering semantics
+/// - This is the minimum granularity that maintains correctness
+///
+/// ## Example
+///
+/// Given a file with 3 row groups, each containing 2 batches:
+///
+/// ```text
+/// Normal order: RG0[B0, B1] -> RG1[B0, B1] -> RG2[B0, B1]
+/// Reversed: RG2[B1_rev, B0_rev] -> RG1[B1_rev, B0_rev] -> RG0[B1_rev, B0_rev]
+/// ^^^^^^^^^^^^ ^^^^^^^^^^^^ ^^^^^^^^^^^^
+/// Output 1st Output 2nd Output 3rd
+/// ```
+///
+/// ## Performance Characteristics
+///
+/// - **Latency**: First batch available after reading complete first (reversed) row group
+/// - **Throughput**: Near-native speed with ~5-10% overhead for reversal operations
+/// - **Memory**: O(row_group_size), not O(file_size)
+/// TODO should we support max cache size to limit memory usage further? But if we exceed the cache size we can't reverse properly, so we need to fall back to normal reading?
Review Comment:
Could you please open a issue to track and descrip the todo?
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3567874839
Thanks -- I'll try and review this tomorrow
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3569024898
> Supporting scanning Parquet files in reverse order is an absolutely great idea. I have a few questions.
>
> Let me first rephrase it to make sure I understand correctly, this PR does:
>
> 1. For applicable query patterns (topK that has reverse order to the parquet existing order), reverse the row-group scanning order
> 2. For each row group, first cache all the result, then reverse the row-level order batch by batch.
>
> This implementation is quite aggressive, I think it can get a bit tricky to tune it right, to avoid excessive caching, or reversing rows batch by batch become too expensive.
>
> What if we limit the initial implementation only to reverse the row-group order, similar to what @adriangb is planning to do at file level in #17271 After scanning the last row-group, the topk dynamic filter will automatically get updated and skip the preceding row groups.
>
> * The benefits are simplicity and lower risk of regressions
> * The downside is it's too conservative and can't get the optimal performance. But once we have native reverse parquet decoding support in `arrow-rs` (that is described in the original issue [Fast parquet order inversionย #17172](https://github.com/apache/datafusion/issues/17172)), we can implement the reverse scan at the row level as follow-ups.
Thank you @2010YOUY01 for review and valid concern:
You raise valid concerns about memory overhead is what i mentioned the key risk for this approach.
However, I want to clarify that row group reversal alone cannot eliminate the SortExec - it only provides TopK filtering benefits. Without reversing rows within each row group, the data remains in the original order (e.g., ASC when we need DESC), so the sort must stay. I propose we keep the complete optimization but default enable_reverse_scan to false. Once we implement page-level caching in arrow-rs (which will reduce memory overhead significantly), we can consider enabling it by default.
And we've been running the full implementation (row group + row-level reversal) in production for very long time with excellent results: 10-100x speedups for time-series queries, well-controlled memory usage (~one row group cached at a time), but we need to note we should not make the row group size big if we enable this feature. And with very small limit, the high memory usage is very short time. Also the reverse time is very small compared to the benefit we remove all sort.
And if we want to improve the original scan to support limit the initial implementation only to reverse the row-group order, i think we can add follow-up PRs because this is another optimization which can't remove the sort for optimization so we need to do this in another PR.
What's your opinion?
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2559136898
##########
datafusion/physical-optimizer/src/pushdown_sort.rs:
##########
@@ -0,0 +1,1235 @@
+// 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.
+
+//! Sort Pushdown Optimization
+//!
+//! This optimizer attempts to push sort requirements down to data sources that can
+//! satisfy them natively, avoiding expensive sort operations.
+//!
+//! Currently supported optimizations:
+//! - **Reverse scan**: If a data source naturally produces data in DESC order and
+//! we need ASC (or vice versa), we can reverse the scan direction instead of
+//! adding a SortExec node.
+//!
+//! Future optimizations could include:
+//! - Reordering row groups in Parquet files
+//! - Leveraging native indexes
+//! - Reordering files in multi-file scans
+use crate::PhysicalOptimizerRule;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion_common::Result;
+use datafusion_datasource::source::DataSourceExec;
+use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
+use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
+use datafusion_physical_plan::sorts::sort::SortExec;
+use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+use std::sync::Arc;
+
+/// A PhysicalOptimizerRule that attempts to push down sort requirements to data sources
+/// that can natively handle them (e.g., by reversing scan direction).
+///
+/// This optimization:
+/// 1. Detects SortExec nodes that require a specific ordering
+/// 2. Checks if the input can satisfy the ordering by reversing its scan direction
+/// 3. Pushes the sort requirement down to the data source when possible
+/// 4. Removes unnecessary sort operations when the input already satisfies the requirement
+#[derive(Debug, Clone, Default)]
+pub struct PushdownSort;
+
+impl PushdownSort {
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for PushdownSort {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ // Check if sort pushdown optimization is enabled
+ let enable_sort_pushdown = config.execution.parquet.enable_sort_pushdown;
+
+ // Return early if not enabled
+ if !enable_sort_pushdown {
+ return Ok(plan);
+ }
+
+ // Search for any SortExec nodes and try to optimize them
+ plan.transform_up(&|plan: Arc<dyn ExecutionPlan>| {
+ // First check if this is a GlobalLimitExec -> SortExec pattern
+ if let Some(limit_exec) = plan.as_any().downcast_ref::<GlobalLimitExec>() {
+ if let Some(sort_exec) =
+ limit_exec.input().as_any().downcast_ref::<SortExec>()
+ {
+ return optimize_limit_sort(limit_exec, sort_exec);
+ }
+ }
+
+ // Otherwise, check if this is just a SortExec
+ let sort_exec = match plan.as_any().downcast_ref::<SortExec>() {
+ Some(sort_exec) => sort_exec,
+ None => return Ok(Transformed::no(plan)),
+ };
+
+ optimize_sort(sort_exec)
+ })
+ .data()
+ }
+
+ fn name(&self) -> &str {
+ "PushdownSort"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// Optimize a SortExec by potentially pushing the sort down to the data source
+fn optimize_sort(sort_exec: &SortExec) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+ let required_ordering = sort_exec.expr();
+
+ // First, check if the sort is already satisfied by input ordering
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+
+ if input_eq_properties.ordering_satisfy(required_ordering.clone())? {
+ return remove_unnecessary_sort(sort_exec, sort_input);
+ }
+ }
+
+ // Try to push the sort requirement down to the data source
+ if let Some(optimized_input) = try_pushdown_sort(&sort_input, required_ordering)? {
+ // Verify that the optimized input satisfies the required ordering
+ if optimized_input
+ .equivalence_properties()
+ .ordering_satisfy(required_ordering.clone())?
+ {
+ return remove_unnecessary_sort(sort_exec, optimized_input);
+ }
+
+ // If not fully satisfied, keep the sort but with optimized input
+ return Ok(Transformed::yes(Arc::new(
+ SortExec::new(required_ordering.clone(), optimized_input)
+ .with_fetch(sort_exec.fetch())
+ .with_preserve_partitioning(sort_exec.preserve_partitioning()),
+ )));
+ }
+
+ Ok(Transformed::no(Arc::new(sort_exec.clone())))
+}
+
+/// Handle the GlobalLimitExec -> SortExec pattern
+fn optimize_limit_sort(
+ limit_exec: &GlobalLimitExec,
+ sort_exec: &SortExec,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+ let required_ordering = sort_exec.expr();
+
+ // Check if input is already sorted
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+ if input_eq_properties.ordering_satisfy(required_ordering.clone())? {
+ // Input is already sorted correctly, remove sort and keep limit
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ sort_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Try to push down the sort requirement
+ if let Some(optimized_input) = try_pushdown_sort(&sort_input, required_ordering)? {
+ if optimized_input
+ .equivalence_properties()
+ .ordering_satisfy(required_ordering.clone())?
+ {
+ // Successfully pushed down sort, now handle the limit
+ let total_fetch = limit_exec.skip() + limit_exec.fetch().unwrap_or(0);
+
+ // Try to push limit down as well if the source supports it
Review Comment:
I add those logics here, because i found the optimizer order make we always need to run some of them more than one times if we remove this logic.
So i will try to add our optimizer to the existed optimizer.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3571228433
> > > I haven't looked into all of this discussion and code (I just got tagged). I've been looking into optimizing sorted scanning in DataFusion and IMO where we should land is:
> > >
> > > 1. Via metadata (`FileScanConfig` / `ORDERED BY ...` in SQL) users declare a known sort order of their files.
> > > 2. The planner uses statistics from the files + any `ORDER BY` clauses in the query to arrange file ordering to best match the query. The `FileSource` implementation can also receive the `ORDER BY` information and optimize scan order within a file (e.g. reversing the order of reads which is what I think this PR is doing).
> > > 3. If the planner is able to deduce from file level stats that the files can be ordered and the `FileSource` reports that it is able to produce batches in sorted order then the optimizer can optimize away the sort completely.
> > >
> > > I hope that is helpful.
> >
> >
> > Thank you @adriangb , it's helpful for future optimization. My current implementation focuses on a specific optimization case: when data is already sorted and we need the reverse order, we can flip the scan direction instead of reading everything and sorting. The reverse_scan in FileSource handles the files/ and within-file ordering reversal.
> > I think these approaches are complementary - my PR handles the reverse scan optimization, while your vision provides a framework for broader sorted-scan optimizations using file-level statistics and metadata. Would be great to build toward that architecture incrementally.
>
> My point is that instead of `enable_reverse_scan: bool` we might want to consider a more holistic approach e.g. `try_pushdown_sort(&self, order: LexOrdering) -> Result<Option<Arc<dyn ExecutionPlan>>>` either at the `ExecutionPlan` level or at the `DataSource` level.
>
> I'm not opposed to this as a step towards that but I'm not sure how helpful it is. Seeing something more concrete w.r.t. how this interacts with the bigger picture would be helpful IMO.
This is a great idea to have high level sort pushdown @adriangb , and reverse scan is one of the polices, i will refactor this PR to use this way, thanks!
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2559218340
##########
datafusion/physical-plan/src/execution_plan.rs:
##########
@@ -682,6 +684,16 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
) -> Option<Arc<dyn ExecutionPlan>> {
None
}
+
+ /// Try to create a new execution plan that satisfies the given sort ordering.
+ ///
+ /// Default implementation returns `Ok(None)`.
+ fn try_pushdown_sort(
Review Comment:
Do we need to add the API for `ExecutionPlan`? Is it possible to push down sort during the pushdown sort optimizer? Because we need to traverse the plan in the rule, so it looks possible to find the target node and directly give it the order.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3568934764
Supporting scanning Parquet files in reverse order is an absolutely great idea. I have a few questions.
Let me first rephrase it to make sure I understand correctly, this PR does:
1. For applicable query patterns (topK that has reverse order to the parquet existing order), reverse the row-group scanning order
2. For each row group, first cache all the result, then reverse the row-level order batch by batch.
This implementation is quite aggressive, I think it can get a bit tricky to tune it right, to avoid excessive caching, or reversing rows batch by batch become too expensive.
What if we limit the initial implementation only to reverse the row-group order, similar to what @adriangb is planning to do at file level in https://github.com/apache/datafusion/issues/17271
After scanning the last row-group, the topk dynamic filter will automatically get updated and skip the preceding row groups.
- The benefits are simplicity and lower risk of regressions
- The downside is it's too conservative and can't get the optimal performance. But once we have native reverse parquet decoding support in `arrow-rs` (that is described in the original issue https://github.com/apache/datafusion/issues/17172), we can implement the reverse scan at the row level as follow-ups.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3569112079
> I haven't looked into all of this discussion and code (I just got tagged). I've been looking into optimizing sorted scanning in DataFusion and IMO where we should land is:
>
> 1. Via metadata (`FileScanConfig` / `ORDERED BY ...` in SQL) users declare a known sort order of their files.
> 2. The planner uses statistics from the files + any `ORDER BY` clauses in the query to arrange file ordering to best match the query. The `FileSource` implementation can also receive the `ORDER BY` information and optimize scan order within a file (e.g. reversing the order of reads which is what I think this PR is doing).
> 3. If the planner is able to deduce from file level stats that the files can be ordered and the `FileSource` reports that it is able to produce batches in sorted order then the optimizer can optimize away the sort completely.
>
> I hope that is helpful.
Thank you @adriangb , it's helpful for future optimization.
My current implementation focuses on a specific optimization case: when data is already sorted and we need the reverse order, we can flip the scan direction instead of reading everything and sorting. The reverse_scan in FileSource handles the within-file ordering reversal.
I think these approaches are complementary - my PR handles the reverse scan optimization, while your vision provides a framework for broader sorted-scan optimizations using file-level statistics and metadata. Would be great to build toward that architecture incrementally.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2559080033
##########
datafusion/physical-optimizer/src/pushdown_sort.rs:
##########
@@ -0,0 +1,1235 @@
+// 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.
+
+//! Sort Pushdown Optimization
+//!
+//! This optimizer attempts to push sort requirements down to data sources that can
+//! satisfy them natively, avoiding expensive sort operations.
+//!
+//! Currently supported optimizations:
+//! - **Reverse scan**: If a data source naturally produces data in DESC order and
+//! we need ASC (or vice versa), we can reverse the scan direction instead of
+//! adding a SortExec node.
+//!
+//! Future optimizations could include:
+//! - Reordering row groups in Parquet files
+//! - Leveraging native indexes
+//! - Reordering files in multi-file scans
+use crate::PhysicalOptimizerRule;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion_common::Result;
+use datafusion_datasource::source::DataSourceExec;
+use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
+use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
+use datafusion_physical_plan::sorts::sort::SortExec;
+use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+use std::sync::Arc;
+
+/// A PhysicalOptimizerRule that attempts to push down sort requirements to data sources
+/// that can natively handle them (e.g., by reversing scan direction).
+///
+/// This optimization:
+/// 1. Detects SortExec nodes that require a specific ordering
+/// 2. Checks if the input can satisfy the ordering by reversing its scan direction
+/// 3. Pushes the sort requirement down to the data source when possible
+/// 4. Removes unnecessary sort operations when the input already satisfies the requirement
+#[derive(Debug, Clone, Default)]
+pub struct PushdownSort;
+
+impl PushdownSort {
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for PushdownSort {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ // Check if sort pushdown optimization is enabled
+ let enable_sort_pushdown = config.execution.parquet.enable_sort_pushdown;
+
+ // Return early if not enabled
+ if !enable_sort_pushdown {
+ return Ok(plan);
+ }
+
+ // Search for any SortExec nodes and try to optimize them
+ plan.transform_up(&|plan: Arc<dyn ExecutionPlan>| {
+ // First check if this is a GlobalLimitExec -> SortExec pattern
+ if let Some(limit_exec) = plan.as_any().downcast_ref::<GlobalLimitExec>() {
+ if let Some(sort_exec) =
+ limit_exec.input().as_any().downcast_ref::<SortExec>()
+ {
+ return optimize_limit_sort(limit_exec, sort_exec);
+ }
+ }
+
+ // Otherwise, check if this is just a SortExec
+ let sort_exec = match plan.as_any().downcast_ref::<SortExec>() {
+ Some(sort_exec) => sort_exec,
+ None => return Ok(Transformed::no(plan)),
+ };
+
+ optimize_sort(sort_exec)
+ })
+ .data()
+ }
+
+ fn name(&self) -> &str {
+ "PushdownSort"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// Optimize a SortExec by potentially pushing the sort down to the data source
+fn optimize_sort(sort_exec: &SortExec) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+ let required_ordering = sort_exec.expr();
+
+ // First, check if the sort is already satisfied by input ordering
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+
+ if input_eq_properties.ordering_satisfy(required_ordering.clone())? {
+ return remove_unnecessary_sort(sort_exec, sort_input);
+ }
+ }
+
+ // Try to push the sort requirement down to the data source
+ if let Some(optimized_input) = try_pushdown_sort(&sort_input, required_ordering)? {
+ // Verify that the optimized input satisfies the required ordering
+ if optimized_input
+ .equivalence_properties()
+ .ordering_satisfy(required_ordering.clone())?
+ {
+ return remove_unnecessary_sort(sort_exec, optimized_input);
+ }
+
+ // If not fully satisfied, keep the sort but with optimized input
+ return Ok(Transformed::yes(Arc::new(
+ SortExec::new(required_ordering.clone(), optimized_input)
+ .with_fetch(sort_exec.fetch())
+ .with_preserve_partitioning(sort_exec.preserve_partitioning()),
+ )));
+ }
+
+ Ok(Transformed::no(Arc::new(sort_exec.clone())))
+}
+
+/// Handle the GlobalLimitExec -> SortExec pattern
+fn optimize_limit_sort(
+ limit_exec: &GlobalLimitExec,
+ sort_exec: &SortExec,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+ let required_ordering = sort_exec.expr();
+
+ // Check if input is already sorted
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+ if input_eq_properties.ordering_satisfy(required_ordering.clone())? {
+ // Input is already sorted correctly, remove sort and keep limit
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ sort_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Try to push down the sort requirement
+ if let Some(optimized_input) = try_pushdown_sort(&sort_input, required_ordering)? {
+ if optimized_input
+ .equivalence_properties()
+ .ordering_satisfy(required_ordering.clone())?
+ {
+ // Successfully pushed down sort, now handle the limit
+ let total_fetch = limit_exec.skip() + limit_exec.fetch().unwrap_or(0);
+
+ // Try to push limit down as well if the source supports it
Review Comment:
I think the current `limit_pushdown` physical optimizer rule can do this. So do we still need to distinguish the sort and limit + sort pattern?
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2559080033
##########
datafusion/physical-optimizer/src/pushdown_sort.rs:
##########
@@ -0,0 +1,1235 @@
+// 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.
+
+//! Sort Pushdown Optimization
+//!
+//! This optimizer attempts to push sort requirements down to data sources that can
+//! satisfy them natively, avoiding expensive sort operations.
+//!
+//! Currently supported optimizations:
+//! - **Reverse scan**: If a data source naturally produces data in DESC order and
+//! we need ASC (or vice versa), we can reverse the scan direction instead of
+//! adding a SortExec node.
+//!
+//! Future optimizations could include:
+//! - Reordering row groups in Parquet files
+//! - Leveraging native indexes
+//! - Reordering files in multi-file scans
+use crate::PhysicalOptimizerRule;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion_common::Result;
+use datafusion_datasource::source::DataSourceExec;
+use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
+use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
+use datafusion_physical_plan::sorts::sort::SortExec;
+use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+use std::sync::Arc;
+
+/// A PhysicalOptimizerRule that attempts to push down sort requirements to data sources
+/// that can natively handle them (e.g., by reversing scan direction).
+///
+/// This optimization:
+/// 1. Detects SortExec nodes that require a specific ordering
+/// 2. Checks if the input can satisfy the ordering by reversing its scan direction
+/// 3. Pushes the sort requirement down to the data source when possible
+/// 4. Removes unnecessary sort operations when the input already satisfies the requirement
+#[derive(Debug, Clone, Default)]
+pub struct PushdownSort;
+
+impl PushdownSort {
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for PushdownSort {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ // Check if sort pushdown optimization is enabled
+ let enable_sort_pushdown = config.execution.parquet.enable_sort_pushdown;
+
+ // Return early if not enabled
+ if !enable_sort_pushdown {
+ return Ok(plan);
+ }
+
+ // Search for any SortExec nodes and try to optimize them
+ plan.transform_up(&|plan: Arc<dyn ExecutionPlan>| {
+ // First check if this is a GlobalLimitExec -> SortExec pattern
+ if let Some(limit_exec) = plan.as_any().downcast_ref::<GlobalLimitExec>() {
+ if let Some(sort_exec) =
+ limit_exec.input().as_any().downcast_ref::<SortExec>()
+ {
+ return optimize_limit_sort(limit_exec, sort_exec);
+ }
+ }
+
+ // Otherwise, check if this is just a SortExec
+ let sort_exec = match plan.as_any().downcast_ref::<SortExec>() {
+ Some(sort_exec) => sort_exec,
+ None => return Ok(Transformed::no(plan)),
+ };
+
+ optimize_sort(sort_exec)
+ })
+ .data()
+ }
+
+ fn name(&self) -> &str {
+ "PushdownSort"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// Optimize a SortExec by potentially pushing the sort down to the data source
+fn optimize_sort(sort_exec: &SortExec) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+ let required_ordering = sort_exec.expr();
+
+ // First, check if the sort is already satisfied by input ordering
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+
+ if input_eq_properties.ordering_satisfy(required_ordering.clone())? {
+ return remove_unnecessary_sort(sort_exec, sort_input);
+ }
+ }
+
+ // Try to push the sort requirement down to the data source
+ if let Some(optimized_input) = try_pushdown_sort(&sort_input, required_ordering)? {
+ // Verify that the optimized input satisfies the required ordering
+ if optimized_input
+ .equivalence_properties()
+ .ordering_satisfy(required_ordering.clone())?
+ {
+ return remove_unnecessary_sort(sort_exec, optimized_input);
+ }
+
+ // If not fully satisfied, keep the sort but with optimized input
+ return Ok(Transformed::yes(Arc::new(
+ SortExec::new(required_ordering.clone(), optimized_input)
+ .with_fetch(sort_exec.fetch())
+ .with_preserve_partitioning(sort_exec.preserve_partitioning()),
+ )));
+ }
+
+ Ok(Transformed::no(Arc::new(sort_exec.clone())))
+}
+
+/// Handle the GlobalLimitExec -> SortExec pattern
+fn optimize_limit_sort(
+ limit_exec: &GlobalLimitExec,
+ sort_exec: &SortExec,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+ let required_ordering = sort_exec.expr();
+
+ // Check if input is already sorted
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+ if input_eq_properties.ordering_satisfy(required_ordering.clone())? {
+ // Input is already sorted correctly, remove sort and keep limit
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ sort_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Try to push down the sort requirement
+ if let Some(optimized_input) = try_pushdown_sort(&sort_input, required_ordering)? {
+ if optimized_input
+ .equivalence_properties()
+ .ordering_satisfy(required_ordering.clone())?
+ {
+ // Successfully pushed down sort, now handle the limit
+ let total_fetch = limit_exec.skip() + limit_exec.fetch().unwrap_or(0);
+
+ // Try to push limit down as well if the source supports it
Review Comment:
I think the current `limit_pushdown` physical optimizer rule can do 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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3586500768
> This is super impressive @zhuqi-lucas ๐ I started going through it but wasn't able to finish before I ran out of time. I still need to review the code in `opener.rs` but I left the comments I have so far
>
> 1. Do you have any performance results with this optimization (in other words, do we have some benchmarks to show the improvement that comes with the complexity?)
>
> While this PR was quite easy to review (nice work!), I do think we should break it into parts to make it easier for others to comment on. Specifically, the
>
> * PushdownSort optimization and API for sorts in the scan (I think @adriangb was thinking about this too recently)
Thank you @alamb for review, i agree we can split to more PRs, and the performance test still needed as follow-up, and we may first make the default flag false for this 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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3605173547
@adriangb Created the sub-task, i will do the first step in another PR:
https://github.com/apache/datafusion/issues/19059
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3608867978
> Amazing @alamb , pretty cool! Appreciate to be added, thanks!
Added in https://github.com/alamb/datafusion-benchmarking/commit/c9f25c40ea6e51fafaf9ba8a9ffd1ee18255d146
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3605978882
I got the performance result for row group reverse with topk dynamic pushdown:
```rust
Running `/Users/zhuqi/arrow-datafusion/target/release/dfbench clickbench --iterations 5 --path /Users/zhuqi/arrow-datafusion/benchmarks/data/hits_0_sorted.parquet --queries-path /Users/zhuqi/arrow-datafusion/benchmarks/queries/clickbench/queries/sorted_data --sorted-by EventTime --sort-order ASC -o /Users/zhuqi/arrow-datafusion/benchmarks/results/issue_19059/data_sorted_clickbench.json`
Running benchmarks with the following options: RunOpt { query: None, pushdown: false, common: CommonOpt { iterations: 5, partitions: None, batch_size: None, mem_pool_type: "fair", memory_limit: None, sort_spill_reservation_bytes: None, debug: false }, path: "/Users/zhuqi/arrow-datafusion/benchmarks/data/hits_0_sorted.parquet", queries_path: "/Users/zhuqi/arrow-datafusion/benchmarks/queries/clickbench/queries/sorted_data", output_path: Some("/Users/zhuqi/arrow-datafusion/benchmarks/results/issue_19059/data_sorted_clickbench.json"), sorted_by: Some("EventTime"), sort_order: "ASC" }
โ ๏ธ Forcing target_partitions=1 to preserve sort order
โ ๏ธ (Because we want to get the pure performance benefit of sorted data to compare)
๐ Session config target_partitions: 1
Registering table with sort order: EventTime ASC
Executing: CREATE EXTERNAL TABLE hits STORED AS PARQUET LOCATION '/Users/zhuqi/arrow-datafusion/benchmarks/data/hits_0_sorted.parquet' WITH ORDER ("EventTime" ASC)
Q0: -- Must set for ClickBench hits_partitioned dataset. See https://github.com/apache/datafusion/issues/16591
-- set datafusion.execution.parquet.binary_as_string = true
SELECT * FROM hits ORDER BY "EventTime" DESC limit 10;
Query 0 iteration 0 took 21.5 ms and returned 10 rows
Query 0 iteration 1 took 12.7 ms and returned 10 rows
Query 0 iteration 2 took 11.1 ms and returned 10 rows
Query 0 iteration 3 took 10.6 ms and returned 10 rows
Query 0 iteration 4 took 9.9 ms and returned 10 rows
Query 0 avg time: 13.17 ms
+ set +x
Done
```
So it's very close to reverse parquet implementation:
The main branch result is:
300ms
The reverse parquet implementation this PR is:
9.8ms
The reverse row group with dynamic topk is https://github.com/apache/datafusion/pull/19064:
13ms
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2581146514
##########
datafusion/datasource-parquet/src/opener.rs:
##########
@@ -752,10 +799,316 @@ fn should_enable_page_index(
.unwrap_or(false)
}
+fn reverse_batch(batch: RecordBatch) -> Result<RecordBatch> {
+ let num_rows = batch.num_rows();
+ if num_rows <= 1 {
+ return Ok(batch);
+ }
+
+ let indices = UInt32Array::from_iter_values((0..num_rows as u32).rev());
+
+ let reversed_columns = batch
+ .columns()
+ .iter()
+ .map(|col| take(col.as_ref(), &indices, None))
+ .collect::<std::result::Result<Vec<ArrayRef>, arrow::error::ArrowError>>()
+ .map_err(DataFusionError::from)?;
+
+ RecordBatch::try_new(batch.schema(), reversed_columns).map_err(DataFusionError::from)
+}
+
+/// Stream adapter for reversed parquet reading with row-group-level buffering.
+///
+/// # Architecture
+///
+/// This stream implements a sophisticated buffering strategy to achieve true reverse
+/// reading of Parquet files while maintaining compatibility with the underlying
+/// ParquetRecordBatchStream's optimizations (caching, prefetching, etc.).
+///
+/// ## Strategy Overview
+///
+/// 1. **Pre-reversed Row Groups**: Row groups are reversed BEFORE building the stream
+/// (via `row_group_indexes.reverse()`). This allows the Parquet reader to read
+/// them in reverse order while still utilizing internal optimizations.
+///
+/// 2. **Row-Group-Level Buffering**: As batches arrive from the input stream, we
+/// track which row group they belong to using cumulative row counts. This is
+/// the MINIMAL buffering unit required for correctness - we cannot reverse
+/// individual batches without knowing the complete row group context.
+///
+/// 3. **Two-Stage Reversal**: When a complete row group is collected:
+/// - Stage 1: Reverse rows within each batch (using Arrow's take kernel)
+/// - Stage 2: Reverse the order of batches within the row group
+///
+/// 4. **Progressive Output**: Reversed batches are output immediately, minimizing
+/// memory footprint. We never buffer more than one row group at a time.
+///
+/// ## Memory Characteristics
+///
+/// - **Bounded Memory**: Maximum memory usage = size of largest row group
+/// - **Typical Usage**: ~128MB (default Parquet row group size)
+/// - **Peak Usage**: During reversal of a single row group
+///
+/// ## Why Row-Group-Level Buffering is Necessary
+///
+/// Parquet organizes data into row groups (typically 128MB each), and each row group
+/// is independently compressed and encoded. When reading in reverse:
+///
+/// - We cannot reverse individual batches in isolation because they may span
+/// row group boundaries or be split arbitrarily by the batch_size parameter
+/// - We must buffer complete row groups to ensure correct ordering semantics
+/// - This is the minimum granularity that maintains correctness
+///
+/// ## Example
+///
+/// Given a file with 3 row groups, each containing 2 batches:
+///
+/// ```text
+/// Normal order: RG0[B0, B1] -> RG1[B0, B1] -> RG2[B0, B1]
+/// Reversed: RG2[B1_rev, B0_rev] -> RG1[B1_rev, B0_rev] -> RG0[B1_rev, B0_rev]
+/// ^^^^^^^^^^^^ ^^^^^^^^^^^^ ^^^^^^^^^^^^
+/// Output 1st Output 2nd Output 3rd
+/// ```
+///
+/// ## Performance Characteristics
+///
+/// - **Latency**: First batch available after reading complete first (reversed) row group
+/// - **Throughput**: Near-native speed with ~5-10% overhead for reversal operations
Review Comment:
This is from my estimation, i can add create a benchmark for this operation as follow-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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas closed pull request #18817: Support reverse parquet scan and fast parquet order inversion at row group level
URL: https://github.com/apache/datafusion/pull/18817
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2554026346
##########
datafusion/physical-optimizer/src/reverse_order.rs:
##########
@@ -0,0 +1,1446 @@
+// 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 crate::PhysicalOptimizerRule;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{
+ Transformed, TransformedResult, TreeNode, TreeNodeRecursion, TreeNodeRewriter,
+};
+use datafusion_common::Result;
+use datafusion_datasource::file::FileSource;
+use datafusion_datasource::file_scan_config::{FileScanConfig, FileScanConfigBuilder};
+use datafusion_datasource::source::{DataSource, DataSourceExec};
+use datafusion_datasource_parquet::source::ParquetSource;
+use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr};
+use datafusion_physical_plan::joins::SortMergeJoinExec;
+use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
+use datafusion_physical_plan::sorts::sort::SortExec;
+use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+use std::sync::Arc;
+
+/// A PhysicalOptimizerRule that attempts to reverse a SortExec's input if doing so would make the
+/// input ordering compatible with the SortExec's required output ordering.
+/// It also removes unnecessary sorts when the input already satisfies the required ordering.
+#[derive(Debug, Clone, Default)]
+pub struct ReverseOrder;
+
+impl ReverseOrder {
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for ReverseOrder {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ // Check if reverse scan optimization is enabled
+ let enable_reverse_scan = config.execution.parquet.enable_reverse_scan;
+
+ // Return early if not enabled
+ if !enable_reverse_scan {
+ return Ok(plan);
+ }
+
+ // Search for any SortExec nodes and try to optimize them
+ plan.transform_up(&|plan: Arc<dyn ExecutionPlan>| {
+ // First check if this is a GlobalLimitExec -> SortExec pattern
+ if let Some(limit_exec) = plan.as_any().downcast_ref::<GlobalLimitExec>() {
+ if let Some(sort_exec) =
+ limit_exec.input().as_any().downcast_ref::<SortExec>()
+ {
+ return optimize_limit_sort(limit_exec, sort_exec);
+ }
+ }
+
+ // Otherwise, check if this is just a SortExec
+ let sort_exec = match plan.as_any().downcast_ref::<SortExec>() {
+ Some(sort_exec) => sort_exec,
+ None => return Ok(Transformed::no(plan)),
+ };
+
+ let sort_input: Arc<dyn ExecutionPlan> = Arc::clone(sort_exec.input());
+
+ // First, check if the sort is already satisfied by input ordering
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+
+ // Check if input already satisfies the sort requirement
+ if input_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ return remove_unnecessary_sort(sort_exec, sort_input);
+ }
+ }
+
+ // If not satisfied, try to reverse the input
+ let reversed_eq_properties = {
+ let mut new = sort_input.properties().equivalence_properties().clone();
+ new.clear_orderings();
+
+ // Build reversed sort exprs for each ordering class
+ let reversed_orderings = sort_input
+ .equivalence_properties()
+ .oeq_class()
+ .iter()
+ .map(|ordering| {
+ ordering
+ .iter()
+ .map(|expr| expr.reverse())
+ .collect::<Vec<_>>()
+ })
+ .collect::<Vec<_>>();
+
+ new.add_orderings(reversed_orderings);
+ new
+ };
+
+ match reversed_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ true => {
+ // Reverse the input and then remove the sort
+ let reversed_input =
+ sort_input.rewrite(&mut ReverseRewriter).unwrap().data;
+
+ // After reversing, check if we can remove the sort
+ if reversed_input
+ .equivalence_properties()
+ .ordering_satisfy(sort_exec.expr().clone())?
+ {
+ remove_unnecessary_sort(sort_exec, reversed_input)
+ } else {
+ // Keep the sort but with reversed input
+ Ok(Transformed::yes(Arc::new(
+ SortExec::new(sort_exec.expr().clone(), reversed_input)
+ .with_fetch(sort_exec.fetch())
+ .with_preserve_partitioning(
+ sort_exec.preserve_partitioning(),
+ ),
+ )))
+ }
+ }
+ false => Ok(Transformed::no(plan)),
+ }
+ })
+ .data()
+ }
+
+ fn name(&self) -> &str {
+ "ReverseOrder"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// Handle the GlobalLimitExec -> SortExec pattern
+fn optimize_limit_sort(
+ limit_exec: &GlobalLimitExec,
+ sort_exec: &SortExec,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+
+ // Check if input is already sorted
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+ if input_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ // Input is already sorted correctly, remove sort and keep limit
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ sort_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Check if we can reverse the input to satisfy the sort
+ let reversed_eq_properties = {
+ let mut new = sort_input.properties().equivalence_properties().clone();
+ new.clear_orderings();
+
+ let reversed_orderings = sort_input
+ .equivalence_properties()
+ .oeq_class()
+ .iter()
+ .map(|ordering| {
+ ordering
+ .iter()
+ .map(|expr| expr.reverse())
+ .collect::<Vec<_>>()
+ })
+ .collect::<Vec<_>>();
+
+ new.add_orderings(reversed_orderings);
+ new
+ };
+
+ if reversed_eq_properties.ordering_satisfy(sort_exec.expr().clone())? {
+ // Can reverse! Apply reversal
+ let reversed_input = sort_input.rewrite(&mut ReverseRewriter).unwrap().data;
+
+ // Check if reversed input satisfies the sort requirement
+ if reversed_input
+ .equivalence_properties()
+ .ordering_satisfy(sort_exec.expr().clone())?
+ {
+ // Check if this is a single-partition DataSourceExec with reverse_scan enabled
+ // In that case, the limit is already handled internally by ReversedParquetStreamWithLimit
+ if is_single_partition_reverse_scan_datasource(&reversed_input) {
+ let total_fetch = limit_exec.skip() + limit_exec.fetch().unwrap_or(0);
+
+ if let Some(with_fetch) = reversed_input.with_fetch(Some(total_fetch)) {
+ if limit_exec.skip() > 0 {
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ with_fetch,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ } else {
+ return Ok(Transformed::yes(with_fetch));
+ }
+ }
+
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ reversed_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+
+ // Otherwise, remove sort but keep limit with reversed input
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ reversed_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Can't optimize, return original pattern
+ Ok(Transformed::no(Arc::new(GlobalLimitExec::new(
+ Arc::new(sort_exec.clone()),
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))))
+}
+
+/// Check if the plan is a single-partition DataSourceExec with reverse_scan enabled
+fn is_single_partition_reverse_scan_datasource(plan: &Arc<dyn ExecutionPlan>) -> bool {
+ // Only optimize for single partition
+ if plan.output_partitioning().partition_count() != 1 {
+ return false;
+ }
+
+ if let Some(data_source_exec) = plan.as_any().downcast_ref::<DataSourceExec>() {
+ if let Some(scan_config) = data_source_exec
+ .data_source()
+ .as_any()
+ .downcast_ref::<FileScanConfig>()
+ {
+ if let Some(parquet_source) = scan_config
+ .file_source
+ .as_any()
+ .downcast_ref::<ParquetSource>()
+ {
+ return parquet_source.reverse_scan();
+ }
+ }
+ }
+ false
+}
+
+/// Remove unnecessary sort based on the logic from EnforceSorting::analyze_immediate_sort_removal
+fn remove_unnecessary_sort(
Review Comment:
Note, i add this to reverse order because after reverse order, we can optimize more to remove the sort, so we don't need to execute the enforce sort optimization again after this optimization.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3592726861
๐ค: Benchmark completed
<details><summary>Details</summary>
<p>
```
Comparing HEAD and reverse_parquet
--------------------
Benchmark clickbench_extended.json
--------------------
โโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโ
โ Query โ HEAD โ reverse_parquet โ Change โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ QQuery 0 โ 2758.70 ms โ 2664.09 ms โ no change โ
โ QQuery 1 โ 1306.02 ms โ 1225.21 ms โ +1.07x faster โ
โ QQuery 2 โ 2444.91 ms โ 2378.03 ms โ no change โ
โ QQuery 3 โ 1161.97 ms โ 1147.66 ms โ no change โ
โ QQuery 4 โ 2368.99 ms โ 2382.54 ms โ no change โ
โ QQuery 5 โ 28439.65 ms โ 28840.09 ms โ no change โ
โ QQuery 6 โ 4264.74 ms โ 4118.34 ms โ no change โ
โ QQuery 7 โ 3897.64 ms โ 3818.70 ms โ no change โ
โโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโ
โ Benchmark Summary โ โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ Total Time (HEAD) โ 46642.64ms โ
โ Total Time (reverse_parquet) โ 46574.65ms โ
โ Average Time (HEAD) โ 5830.33ms โ
โ Average Time (reverse_parquet) โ 5821.83ms โ
โ Queries Faster โ 1 โ
โ Queries Slower โ 0 โ
โ Queries with No Change โ 7 โ
โ Queries with Failure โ 0 โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโ
--------------------
Benchmark clickbench_partitioned.json
--------------------
โโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโ
โ Query โ HEAD โ reverse_parquet โ Change โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ QQuery 0 โ 2.68 ms โ 2.59 ms โ no change โ
โ QQuery 1 โ 50.49 ms โ 48.59 ms โ no change โ
โ QQuery 2 โ 137.13 ms โ 136.98 ms โ no change โ
โ QQuery 3 โ 161.78 ms โ 160.98 ms โ no change โ
โ QQuery 4 โ 1148.92 ms โ 1101.08 ms โ no change โ
โ QQuery 5 โ 1613.34 ms โ 1563.22 ms โ no change โ
โ QQuery 6 โ 2.22 ms โ 2.29 ms โ no change โ
โ QQuery 7 โ 57.11 ms โ 55.10 ms โ no change โ
โ QQuery 8 โ 1538.73 ms โ 1473.96 ms โ no change โ
โ QQuery 9 โ 1996.44 ms โ 1854.46 ms โ +1.08x faster โ
โ QQuery 10 โ 404.03 ms โ 363.43 ms โ +1.11x faster โ
โ QQuery 11 โ 454.60 ms โ 417.14 ms โ +1.09x faster โ
โ QQuery 12 โ 1447.41 ms โ 1387.72 ms โ no change โ
โ QQuery 13 โ 2271.75 ms โ 2141.22 ms โ +1.06x faster โ
โ QQuery 14 โ 1339.13 ms โ 1294.92 ms โ no change โ
โ QQuery 15 โ 1333.78 ms โ 1286.78 ms โ no change โ
โ QQuery 16 โ 2790.01 ms โ 2753.47 ms โ no change โ
โ QQuery 17 โ 2759.16 ms โ 2730.38 ms โ no change โ
โ QQuery 18 โ 5591.07 ms โ 5169.15 ms โ +1.08x faster โ
โ QQuery 19 โ 126.87 ms โ 125.69 ms โ no change โ
โ QQuery 20 โ 2026.21 ms โ 2014.34 ms โ no change โ
โ QQuery 21 โ 2376.95 ms โ 2331.82 ms โ no change โ
โ QQuery 22 โ 4126.05 ms โ 3989.54 ms โ no change โ
โ QQuery 23 โ 29485.23 ms โ 13017.00 ms โ +2.27x faster โ
โ QQuery 24 โ 242.02 ms โ 220.08 ms โ +1.10x faster โ
โ QQuery 25 โ 492.25 ms โ 486.30 ms โ no change โ
โ QQuery 26 โ 234.05 ms โ 218.64 ms โ +1.07x faster โ
โ QQuery 27 โ 2939.61 ms โ 2848.30 ms โ no change โ
โ QQuery 28 โ 24541.56 ms โ 23573.96 ms โ no change โ
โ QQuery 29 โ 949.79 ms โ 999.04 ms โ 1.05x slower โ
โ QQuery 30 โ 1406.83 ms โ 1355.84 ms โ no change โ
โ QQuery 31 โ 1454.02 ms โ 1426.96 ms โ no change โ
โ QQuery 32 โ 4805.92 ms โ 4840.46 ms โ no change โ
โ QQuery 33 โ 6107.78 ms โ 6571.56 ms โ 1.08x slower โ
โ QQuery 34 โ 6539.51 ms โ 6943.50 ms โ 1.06x slower โ
โ QQuery 35 โ 2050.16 ms โ 1944.12 ms โ +1.05x faster โ
โ QQuery 36 โ 123.55 ms โ 120.80 ms โ no change โ
โ QQuery 37 โ 56.58 ms โ 52.08 ms โ +1.09x faster โ
โ QQuery 38 โ 124.23 ms โ 122.64 ms โ no change โ
โ QQuery 39 โ 203.84 ms โ 202.28 ms โ no change โ
โ QQuery 40 โ 42.85 ms โ 43.45 ms โ no change โ
โ QQuery 41 โ 40.69 ms โ 40.84 ms โ no change โ
โ QQuery 42 โ 34.34 ms โ 32.52 ms โ +1.06x faster โ
โโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโ
โ Benchmark Summary โ โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ Total Time (HEAD) โ 115630.69ms โ
โ Total Time (reverse_parquet) โ 97465.24ms โ
โ Average Time (HEAD) โ 2689.09ms โ
โ Average Time (reverse_parquet) โ 2266.63ms โ
โ Queries Faster โ 11 โ
โ Queries Slower โ 3 โ
โ Queries with No Change โ 29 โ
โ Queries with Failure โ 0 โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโ
--------------------
Benchmark tpch_mem_sf1.json
--------------------
โโโโโโโโโโโโโโโโณโโโโโโโโโโโโณโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโ
โ Query โ HEAD โ reverse_parquet โ Change โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ QQuery 1 โ 127.92 ms โ 134.33 ms โ 1.05x slower โ
โ QQuery 2 โ 27.88 ms โ 27.86 ms โ no change โ
โ QQuery 3 โ 32.81 ms โ 39.13 ms โ 1.19x slower โ
โ QQuery 4 โ 28.06 ms โ 29.43 ms โ no change โ
โ QQuery 5 โ 85.52 ms โ 88.58 ms โ no change โ
โ QQuery 6 โ 19.37 ms โ 19.72 ms โ no change โ
โ QQuery 7 โ 223.00 ms โ 222.02 ms โ no change โ
โ QQuery 8 โ 33.75 ms โ 34.75 ms โ no change โ
โ QQuery 9 โ 103.59 ms โ 102.77 ms โ no change โ
โ QQuery 10 โ 63.12 ms โ 63.66 ms โ no change โ
โ QQuery 11 โ 17.38 ms โ 17.69 ms โ no change โ
โ QQuery 12 โ 51.73 ms โ 50.86 ms โ no change โ
โ QQuery 13 โ 46.02 ms โ 46.16 ms โ no change โ
โ QQuery 14 โ 13.63 ms โ 14.07 ms โ no change โ
โ QQuery 15 โ 24.20 ms โ 25.26 ms โ no change โ
โ QQuery 16 โ 25.66 ms โ 25.67 ms โ no change โ
โ QQuery 17 โ 148.71 ms โ 153.50 ms โ no change โ
โ QQuery 18 โ 277.18 ms โ 288.04 ms โ no change โ
โ QQuery 19 โ 37.84 ms โ 38.06 ms โ no change โ
โ QQuery 20 โ 48.59 ms โ 50.59 ms โ no change โ
โ QQuery 21 โ 312.91 ms โ 334.82 ms โ 1.07x slower โ
โ QQuery 22 โ 17.59 ms โ 18.25 ms โ no change โ
โโโโโโโโโโโโโโโโดโโโโโโโโโโโโดโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโ
โ Benchmark Summary โ โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ Total Time (HEAD) โ 1766.46ms โ
โ Total Time (reverse_parquet) โ 1825.20ms โ
โ Average Time (HEAD) โ 80.29ms โ
โ Average Time (reverse_parquet) โ 82.96ms โ
โ Queries Faster โ 0 โ
โ Queries Slower โ 3 โ
โ Queries with No Change โ 19 โ
โ Queries with Failure โ 0 โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโ
```
</p>
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3585245908
๐ค `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running
Linux aal-dev 6.14.0-1018-gcp #19~24.04.1-Ubuntu SMP Wed Sep 24 23:23:09 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing reverse_parquet (2d77d77f0d8e1688ee7c8882b98b2ce50a645f3e) to 710d14b0dfbef0d62e05baaaba3c2a73f4909296 [diff](https://github.com/apache/datafusion/compare/710d14b0dfbef0d62e05baaaba3c2a73f4909296..2d77d77f0d8e1688ee7c8882b98b2ce50a645f3e) using: tpch_mem clickbench_partitioned clickbench_extended
Results will be posted here when complete
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2559356472
##########
datafusion/physical-plan/src/execution_plan.rs:
##########
@@ -682,6 +684,16 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
) -> Option<Arc<dyn ExecutionPlan>> {
None
}
+
+ /// Try to create a new execution plan that satisfies the given sort ordering.
+ ///
+ /// Default implementation returns `Ok(None)`.
+ fn try_pushdown_sort(
Review Comment:
This is a good point.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3592627415
๐ค `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running
Linux aal-dev 6.14.0-1018-gcp #19~24.04.1-Ubuntu SMP Wed Sep 24 23:23:09 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing reverse_parquet (86e027cf33006ffe99d44ece2065e543fa073256) to 79257353be4a4dae5002ccf9450fe58e6d4ada8c [diff](https://github.com/apache/datafusion/compare/79257353be4a4dae5002ccf9450fe58e6d4ada8c..86e027cf33006ffe99d44ece2065e543fa073256) using: tpch_mem clickbench_partitioned clickbench_extended
Results will be posted here when complete
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3593304843
๐ค `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running
Linux aal-dev 6.14.0-1018-gcp #19~24.04.1-Ubuntu SMP Wed Sep 24 23:23:09 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing reverse_parquet (86e027cf33006ffe99d44ece2065e543fa073256) to 79257353be4a4dae5002ccf9450fe58e6d4ada8c [diff](https://github.com/apache/datafusion/compare/79257353be4a4dae5002ccf9450fe58e6d4ada8c..86e027cf33006ffe99d44ece2065e543fa073256) using: clickbench_partitioned
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3593297302
run benchmark clickbench_partitioned
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2575939040
##########
datafusion/physical-optimizer/src/pushdown_sort.rs:
##########
@@ -0,0 +1,1235 @@
+// 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.
+
+//! Sort Pushdown Optimization
+//!
+//! This optimizer attempts to push sort requirements down to data sources that can
+//! satisfy them natively, avoiding expensive sort operations.
+//!
+//! Currently supported optimizations:
+//! - **Reverse scan**: If a data source naturally produces data in DESC order and
+//! we need ASC (or vice versa), we can reverse the scan direction instead of
+//! adding a SortExec node.
+//!
+//! Future optimizations could include:
+//! - Reordering row groups in Parquet files
+//! - Leveraging native indexes
+//! - Reordering files in multi-file scans
+use crate::PhysicalOptimizerRule;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion_common::Result;
+use datafusion_datasource::source::DataSourceExec;
+use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
+use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
+use datafusion_physical_plan::sorts::sort::SortExec;
+use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+use std::sync::Arc;
+
+/// A PhysicalOptimizerRule that attempts to push down sort requirements to data sources
+/// that can natively handle them (e.g., by reversing scan direction).
+///
+/// This optimization:
+/// 1. Detects SortExec nodes that require a specific ordering
+/// 2. Checks if the input can satisfy the ordering by reversing its scan direction
+/// 3. Pushes the sort requirement down to the data source when possible
+/// 4. Removes unnecessary sort operations when the input already satisfies the requirement
+#[derive(Debug, Clone, Default)]
+pub struct PushdownSort;
+
+impl PushdownSort {
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for PushdownSort {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ // Check if sort pushdown optimization is enabled
+ let enable_sort_pushdown = config.execution.parquet.enable_sort_pushdown;
+
+ // Return early if not enabled
+ if !enable_sort_pushdown {
+ return Ok(plan);
+ }
+
+ // Search for any SortExec nodes and try to optimize them
+ plan.transform_up(&|plan: Arc<dyn ExecutionPlan>| {
+ // First check if this is a GlobalLimitExec -> SortExec pattern
+ if let Some(limit_exec) = plan.as_any().downcast_ref::<GlobalLimitExec>() {
+ if let Some(sort_exec) =
+ limit_exec.input().as_any().downcast_ref::<SortExec>()
+ {
+ return optimize_limit_sort(limit_exec, sort_exec);
+ }
+ }
+
+ // Otherwise, check if this is just a SortExec
+ let sort_exec = match plan.as_any().downcast_ref::<SortExec>() {
+ Some(sort_exec) => sort_exec,
+ None => return Ok(Transformed::no(plan)),
+ };
+
+ optimize_sort(sort_exec)
+ })
+ .data()
+ }
+
+ fn name(&self) -> &str {
+ "PushdownSort"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// Optimize a SortExec by potentially pushing the sort down to the data source
+fn optimize_sort(sort_exec: &SortExec) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+ let required_ordering = sort_exec.expr();
+
+ // First, check if the sort is already satisfied by input ordering
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+
+ if input_eq_properties.ordering_satisfy(required_ordering.clone())? {
+ return remove_unnecessary_sort(sort_exec, sort_input);
+ }
+ }
+
+ // Try to push the sort requirement down to the data source
+ if let Some(optimized_input) = try_pushdown_sort(&sort_input, required_ordering)? {
+ // Verify that the optimized input satisfies the required ordering
+ if optimized_input
+ .equivalence_properties()
+ .ordering_satisfy(required_ordering.clone())?
+ {
+ return remove_unnecessary_sort(sort_exec, optimized_input);
+ }
+
+ // If not fully satisfied, keep the sort but with optimized input
+ return Ok(Transformed::yes(Arc::new(
+ SortExec::new(required_ordering.clone(), optimized_input)
+ .with_fetch(sort_exec.fetch())
+ .with_preserve_partitioning(sort_exec.preserve_partitioning()),
+ )));
+ }
+
+ Ok(Transformed::no(Arc::new(sort_exec.clone())))
+}
+
+/// Handle the GlobalLimitExec -> SortExec pattern
+fn optimize_limit_sort(
+ limit_exec: &GlobalLimitExec,
+ sort_exec: &SortExec,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+ let required_ordering = sort_exec.expr();
+
+ // Check if input is already sorted
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+ if input_eq_properties.ordering_satisfy(required_ordering.clone())? {
+ // Input is already sorted correctly, remove sort and keep limit
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ sort_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Try to push down the sort requirement
+ if let Some(optimized_input) = try_pushdown_sort(&sort_input, required_ordering)? {
+ if optimized_input
+ .equivalence_properties()
+ .ordering_satisfy(required_ordering.clone())?
+ {
+ // Successfully pushed down sort, now handle the limit
+ let total_fetch = limit_exec.skip() + limit_exec.fetch().unwrap_or(0);
+
+ // Try to push limit down as well if the source supports it
Review Comment:
Which one of our optimizers?
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3594499999
I'll review another round today
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3600088081
> I think the PR as the initial version is very close
Thank you @xudong963 for review, i will try to address it. And we also need benchmark for this PR to compare the performance with original dynamic 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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2569303471
##########
datafusion/physical-optimizer/src/optimizer.rs:
##########
@@ -145,11 +146,9 @@ impl PhysicalOptimizer {
// are not present, the load of executors such as join or union will be
// reduced by narrowing their input tables.
Arc::new(ProjectionPushdown::new()),
- Arc::new(EnsureCooperative::new()),
- // This FilterPushdown handles dynamic filters that may have references to the source ExecutionPlan.
- // Therefore it should be run at the end of the optimization process since any changes to the plan may break the dynamic filter's references.
- // See `FilterPushdownPhase` for more details.
- Arc::new(FilterPushdown::new_post_optimization()),
+ // ReverseOrder: Detect DESC sorts that can use reverse scan
+ // This marks reverse_scan=true on DataSourceExec
+ Arc::new(PushdownSort::new()),
// The SanityCheckPlan rule checks whether the order and
Review Comment:
Thank you @alamb for good catch, i addressed this in latest 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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2570722799
##########
datafusion/physical-optimizer/src/pushdown_sort.rs:
##########
@@ -0,0 +1,1235 @@
+// 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.
+
+//! Sort Pushdown Optimization
+//!
+//! This optimizer attempts to push sort requirements down to data sources that can
+//! satisfy them natively, avoiding expensive sort operations.
+//!
+//! Currently supported optimizations:
+//! - **Reverse scan**: If a data source naturally produces data in DESC order and
+//! we need ASC (or vice versa), we can reverse the scan direction instead of
+//! adding a SortExec node.
+//!
+//! Future optimizations could include:
+//! - Reordering row groups in Parquet files
+//! - Leveraging native indexes
+//! - Reordering files in multi-file scans
+use crate::PhysicalOptimizerRule;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion_common::Result;
+use datafusion_datasource::source::DataSourceExec;
+use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
+use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
+use datafusion_physical_plan::sorts::sort::SortExec;
+use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+use std::sync::Arc;
+
+/// A PhysicalOptimizerRule that attempts to push down sort requirements to data sources
+/// that can natively handle them (e.g., by reversing scan direction).
+///
+/// This optimization:
+/// 1. Detects SortExec nodes that require a specific ordering
+/// 2. Checks if the input can satisfy the ordering by reversing its scan direction
+/// 3. Pushes the sort requirement down to the data source when possible
+/// 4. Removes unnecessary sort operations when the input already satisfies the requirement
+#[derive(Debug, Clone, Default)]
+pub struct PushdownSort;
+
+impl PushdownSort {
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for PushdownSort {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ // Check if sort pushdown optimization is enabled
+ let enable_sort_pushdown = config.execution.parquet.enable_sort_pushdown;
+
+ // Return early if not enabled
+ if !enable_sort_pushdown {
+ return Ok(plan);
+ }
+
+ // Search for any SortExec nodes and try to optimize them
+ plan.transform_up(&|plan: Arc<dyn ExecutionPlan>| {
+ // First check if this is a GlobalLimitExec -> SortExec pattern
+ if let Some(limit_exec) = plan.as_any().downcast_ref::<GlobalLimitExec>() {
+ if let Some(sort_exec) =
+ limit_exec.input().as_any().downcast_ref::<SortExec>()
+ {
+ return optimize_limit_sort(limit_exec, sort_exec);
+ }
+ }
+
+ // Otherwise, check if this is just a SortExec
+ let sort_exec = match plan.as_any().downcast_ref::<SortExec>() {
+ Some(sort_exec) => sort_exec,
+ None => return Ok(Transformed::no(plan)),
+ };
+
+ optimize_sort(sort_exec)
+ })
+ .data()
+ }
+
+ fn name(&self) -> &str {
+ "PushdownSort"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// Optimize a SortExec by potentially pushing the sort down to the data source
+fn optimize_sort(sort_exec: &SortExec) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+ let required_ordering = sort_exec.expr();
+
+ // First, check if the sort is already satisfied by input ordering
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+
+ if input_eq_properties.ordering_satisfy(required_ordering.clone())? {
+ return remove_unnecessary_sort(sort_exec, sort_input);
+ }
+ }
+
+ // Try to push the sort requirement down to the data source
+ if let Some(optimized_input) = try_pushdown_sort(&sort_input, required_ordering)? {
+ // Verify that the optimized input satisfies the required ordering
+ if optimized_input
+ .equivalence_properties()
+ .ordering_satisfy(required_ordering.clone())?
+ {
+ return remove_unnecessary_sort(sort_exec, optimized_input);
+ }
+
+ // If not fully satisfied, keep the sort but with optimized input
+ return Ok(Transformed::yes(Arc::new(
+ SortExec::new(required_ordering.clone(), optimized_input)
+ .with_fetch(sort_exec.fetch())
+ .with_preserve_partitioning(sort_exec.preserve_partitioning()),
+ )));
+ }
+
+ Ok(Transformed::no(Arc::new(sort_exec.clone())))
+}
+
+/// Handle the GlobalLimitExec -> SortExec pattern
+fn optimize_limit_sort(
+ limit_exec: &GlobalLimitExec,
+ sort_exec: &SortExec,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let sort_input = Arc::clone(sort_exec.input());
+ let required_ordering = sort_exec.expr();
+
+ // Check if input is already sorted
+ if let Some(_input_ordering) = sort_input.output_ordering() {
+ let input_eq_properties = sort_input.equivalence_properties();
+ if input_eq_properties.ordering_satisfy(required_ordering.clone())? {
+ // Input is already sorted correctly, remove sort and keep limit
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ sort_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Try to push down the sort requirement
+ if let Some(optimized_input) = try_pushdown_sort(&sort_input, required_ordering)? {
+ if optimized_input
+ .equivalence_properties()
+ .ordering_satisfy(required_ordering.clone())?
+ {
+ // Successfully pushed down sort, now handle the limit
+ let total_fetch = limit_exec.skip() + limit_exec.fetch().unwrap_or(0);
+
+ // Try to push limit down as well if the source supports it
+ if let Some(with_fetch) = optimized_input.with_fetch(Some(total_fetch)) {
+ if limit_exec.skip() > 0 {
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ with_fetch,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ } else {
+ return Ok(Transformed::yes(with_fetch));
+ }
+ }
+
+ return Ok(Transformed::yes(Arc::new(GlobalLimitExec::new(
+ optimized_input,
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))));
+ }
+ }
+
+ // Can't optimize, return original pattern
+ Ok(Transformed::no(Arc::new(GlobalLimitExec::new(
+ Arc::new(sort_exec.clone()),
+ limit_exec.skip(),
+ limit_exec.fetch(),
+ ))))
+}
+
+/// Remove unnecessary sort based on the logic from EnforceSorting::analyze_immediate_sort_removal
+fn remove_unnecessary_sort(
+ sort_exec: &SortExec,
+ sort_input: Arc<dyn ExecutionPlan>,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let new_plan = if !sort_exec.preserve_partitioning()
+ && sort_input.output_partitioning().partition_count() > 1
+ {
+ // Replace the sort with a sort-preserving merge
+ Arc::new(
+ SortPreservingMergeExec::new(sort_exec.expr().clone(), sort_input)
+ .with_fetch(sort_exec.fetch()),
+ ) as _
+ } else {
+ // Remove the sort entirely
+ if let Some(fetch) = sort_exec.fetch() {
+ // If the sort has a fetch, add a limit instead
+ if sort_input.output_partitioning().partition_count() == 1 {
+ // Try to push the limit down to the source
+ if let Some(with_fetch) = sort_input.with_fetch(Some(fetch)) {
+ return Ok(Transformed::yes(with_fetch));
+ }
+ Arc::new(GlobalLimitExec::new(sort_input, 0, Some(fetch)))
+ as Arc<dyn ExecutionPlan>
+ } else {
+ Arc::new(LocalLimitExec::new(sort_input, fetch)) as Arc<dyn ExecutionPlan>
+ }
+ } else {
+ sort_input
+ }
+ };
+
+ Ok(Transformed::yes(new_plan))
+}
+
+/// Try to push down a sort requirement to an execution plan
+fn try_pushdown_sort(
+ plan: &Arc<dyn ExecutionPlan>,
+ required_ordering: &[PhysicalSortExpr],
+) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ // Check if the plan can natively handle the sort requirement
+ if let Some(data_source_exec) = plan.as_any().downcast_ref::<DataSourceExec>() {
+ return data_source_exec.try_pushdown_sort(required_ordering);
+ }
+
+ Ok(None)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow::array::{Int32Array, RecordBatch};
+ use arrow::compute::SortOptions;
+ use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+ use datafusion_physical_expr::expressions::Column;
+ use datafusion_physical_expr::Partitioning;
+ use datafusion_physical_expr_common::sort_expr::LexOrdering;
+ use datafusion_physical_plan::empty::EmptyExec;
+ use datafusion_physical_plan::memory::LazyMemoryExec;
+ use datafusion_physical_plan::repartition::RepartitionExec;
+ use parking_lot::RwLock;
+ use std::fmt;
+
+ /// Helper function to create a test schema with three columns
+ fn create_test_schema() -> Arc<Schema> {
+ Arc::new(Schema::new(vec![
+ Field::new("col1", DataType::Int32, true),
+ Field::new("col2", DataType::Utf8, true),
+ Field::new("col3", DataType::Float64, true),
+ ]))
+ }
+
+ /// Test batch generator that produces sorted data
+ #[derive(Debug, Clone)]
+ struct SortedBatchGenerator {
+ schema: SchemaRef,
+ batches_generated: usize,
+ max_batches: usize,
+ ascending: bool,
+ }
+
+ impl fmt::Display for SortedBatchGenerator {
+ fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+ write!(
+ f,
+ "SortedBatchGenerator(batches_generated={}, max_batches={}, ascending={})",
+ self.batches_generated, self.max_batches, self.ascending
+ )
+ }
+ }
+
+ impl datafusion_physical_plan::memory::LazyBatchGenerator for SortedBatchGenerator {
Review Comment:
Updated, the pushdown sort test is using insta now also.
--
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] Support reverse parquet scan and fast parquet order inversion at row group level [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#issuecomment-3608857403
I plan to try and re-review this in detail tomorrow
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org