From aa7bfa9a95f1b30f20579a83b14713433d59f1f8 Mon Sep 17 00:00:00 2001 From: Qi Zhu Date: Wed, 4 Nov 2026 12:17:53 +0800 Subject: [PATCH 1/5] feat: exact reverse scan for parquet sort pushdown When a query asks for the reverse of the files' declared ordering, the parquet scan already reverses the row group order, but rows inside each row group stay in on-disk order, so the pushdown is `Inexact` and a `SortExec` (usually `TopK`) stays above the scan. Add `datafusion.execution.parquet.enable_exact_reverse_scan` (default false). With it set, `ParquetSource::try_pushdown_sort` returns `Exact` for a reversed request: the opener reverses row groups and also reverses the rows of each row group by draining its reader, reversing each batch and emitting the batches last to first. `FileScanConfig` reverses the files of each group, declares the reversed orderings, and re-validates them, falling back to `Inexact` when they do not hold (e.g. overlapping files). The `SortExec` is removed and a `LIMIT` is pushed into the scan. A limit is counted in decode order, which the reversal does not follow, so a reversing scan does not pass it to the decoder or use it for row group and page pruning; the file stream applies it to reversed output. Instead, under a limit: - a row group whose output is known up front (no predicate, or fully matched by statistics) is cut to its last `limit` rows with a row selection, and the offset index is loaded so the pages ahead of them are not fetched; - for other row groups only the last `limit` decoded rows are kept. The buffered rows are charged to the task's memory pool through a new `FileSource::with_memory_pool` hook (default: no accounting), so an oversized row group fails the scan with a resources error rather than growing the process. Row groups are not reordered by statistics in this mode, since that is not order-preserving for row groups that share a `min`. --- datafusion/common/src/config.rs | 8 + .../common/src/file_options/parquet_writer.rs | 4 + .../datasource-parquet/src/access_plan.rs | 148 +++++++ .../datasource-parquet/src/file_format.rs | 1 + .../datasource-parquet/src/opener/mod.rs | 331 +++++++++++++++- .../datasource-parquet/src/push_decoder.rs | 115 +++++- datafusion/datasource-parquet/src/source.rs | 132 ++++++- datafusion/datasource/src/file.rs | 13 + .../datasource/src/file_scan_config/mod.rs | 125 ++++++ .../src/file_scan_config/sort_pushdown.rs | 37 +- .../proto/datafusion_common.proto | 1 + datafusion/proto-common/src/from_proto/mod.rs | 1 + .../proto-common/src/generated/pbjson.rs | 18 + .../proto-common/src/generated/prost.rs | 3 + datafusion/proto-common/src/to_proto/mod.rs | 1 + .../proto-models/proto/datafusion.proto | 4 + datafusion/proto-models/src/from_proto.rs | 1 + .../src/generated/datafusion_proto_common.rs | 3 + .../proto-models/src/generated/pbjson.rs | 18 + .../proto-models/src/generated/prost.rs | 4 + datafusion/proto/tests/cases/plans/sources.rs | 47 ++- .../test_files/information_schema.slt | 2 + .../sort_pushdown_exact_reverse.slt | 366 ++++++++++++++++++ docs/source/user-guide/configs.md | 1 + 24 files changed, 1371 insertions(+), 13 deletions(-) create mode 100644 datafusion/sqllogictest/test_files/sort_pushdown_exact_reverse.slt diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 5106ea51ea6ba..fd19f7e70e776 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1354,6 +1354,14 @@ config_namespace! { /// pattern of selected rows. pub force_filter_selections: bool, default = false + /// (reading) If true, a scan whose requested ordering is the reverse of + /// the files' declared ordering reads each row group into memory and + /// emits its rows in reverse, so the scan is exactly ordered and the + /// `SortExec` above it can be removed. Peak memory per scanned file is + /// one decoded row group (after projection and filtering). If false, + /// only the row group order is reversed and the sort is kept. + pub enable_exact_reverse_scan: bool, default = false + /// (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`, /// and `Binary/BinaryLarge` with `BinaryView`. /// diff --git a/datafusion/common/src/file_options/parquet_writer.rs b/datafusion/common/src/file_options/parquet_writer.rs index 0d7d08f85c80a..ebd1bccb252ca 100644 --- a/datafusion/common/src/file_options/parquet_writer.rs +++ b/datafusion/common/src/file_options/parquet_writer.rs @@ -226,6 +226,7 @@ impl ParquetOptions { pushdown_filters: _, reorder_filters: _, force_filter_selections: _, // not used for writer props + enable_exact_reverse_scan: _, // not used for writer props allow_single_file_parallelism: _, maximum_parallel_row_group_writers: _, maximum_buffered_record_batches_per_stream: _, @@ -391,6 +392,7 @@ mod tests { pushdown_filters: defaults.pushdown_filters, reorder_filters: defaults.reorder_filters, force_filter_selections: defaults.force_filter_selections, + enable_exact_reverse_scan: defaults.enable_exact_reverse_scan, allow_single_file_parallelism: defaults.allow_single_file_parallelism, maximum_parallel_row_group_writers: defaults .maximum_parallel_row_group_writers, @@ -522,6 +524,8 @@ mod tests { pushdown_filters: global_options_defaults.pushdown_filters, reorder_filters: global_options_defaults.reorder_filters, force_filter_selections: global_options_defaults.force_filter_selections, + enable_exact_reverse_scan: global_options_defaults + .enable_exact_reverse_scan, allow_single_file_parallelism: global_options_defaults .allow_single_file_parallelism, maximum_parallel_row_group_writers: global_options_defaults diff --git a/datafusion/datasource-parquet/src/access_plan.rs b/datafusion/datasource-parquet/src/access_plan.rs index 5cf9ebe97bda7..9878c20fe54c8 100644 --- a/datafusion/datasource-parquet/src/access_plan.rs +++ b/datafusion/datasource-parquet/src/access_plan.rs @@ -879,6 +879,50 @@ impl PreparedAccessPlan { self } + /// For a scan that emits each row group's rows in reverse, select only + /// the rows a `limit` can reach, so the rows ahead of them are never + /// decoded. + /// + /// Row groups are walked in scan order. One whose output row count is + /// known up front (every row passes: `all_rows_match`, or statistics + /// proved it `fully_matched`) and that has no selection of its own + /// contributes its last `min(remaining, num_rows)` rows. The walk stops + /// at the first row group whose output is not known, since it may + /// produce fewer rows than it holds, and once `limit` rows are covered: + /// the scan stops reading before it reaches the row groups after that. + pub(crate) fn select_tail_for_limit( + mut self, + limit: usize, + all_rows_match: bool, + row_group_meta_data: &[RowGroupMetaData], + ) -> Self { + let mut remaining = limit; + for row_group in &mut self.row_groups { + if remaining == 0 + || !(all_rows_match || row_group.fully_matched) + || row_group.selection.selection().is_some() + { + break; + } + let index = row_group.selection.row_group_index(); + let num_rows = usize::try_from(row_group_meta_data[index].num_rows()) + .unwrap_or(usize::MAX); + let take = remaining.min(num_rows); + if take < num_rows { + row_group.selection = RowGroupSelection::new( + index, + Some(RowSelection::from(vec![ + RowSelector::skip(num_rows - take), + RowSelector::select(take), + ])), + ); + self.has_row_selection = true; + } + remaining -= take; + } + self + } + #[cfg(test)] pub(crate) fn row_group_indexes(&self) -> Vec { self.row_groups @@ -1144,6 +1188,110 @@ mod test { } } + /// Row groups of 10, 20, 30, 40 rows read back to front (3, 2, 1, 0). + fn reversed_plan(plan: ParquetAccessPlan) -> PreparedAccessPlan { + plan.prepare(&ROW_GROUP_METADATA).unwrap().reverse() + } + + fn selected_rows(plan: &PreparedAccessPlan) -> Vec>> { + plan.row_groups + .iter() + .map(|rg| { + rg.selection + .selection() + .map(|selection| selection.iter().copied().collect()) + }) + .collect() + } + + /// Without a predicate every row passes, so each row group's output is + /// known: the first one read (40 rows) is cut to its last 25 rows, and + /// the walk stops once the limit is covered. + #[test] + fn select_tail_for_limit_selects_last_rows() { + let plan = reversed_plan(ParquetAccessPlan::new_all(4)).select_tail_for_limit( + 25, + true, + &ROW_GROUP_METADATA, + ); + + assert_eq!( + selected_rows(&plan), + vec![ + Some(vec![RowSelector::skip(15), RowSelector::select(25)]), + None, + None, + None, + ] + ); + assert!(plan.has_row_selection); + } + + /// A limit larger than a row group takes it whole and the rest from the + /// next one. + #[test] + fn select_tail_for_limit_spans_row_groups() { + let plan = reversed_plan(ParquetAccessPlan::new_all(4)).select_tail_for_limit( + 50, + true, + &ROW_GROUP_METADATA, + ); + + assert_eq!( + selected_rows(&plan), + vec![ + None, + Some(vec![RowSelector::skip(20), RowSelector::select(10)]), + None, + None, + ] + ); + } + + /// With a predicate, only row groups that statistics proved fully + /// matched have a known output; the walk stops at the first one that + /// was not. + #[test] + fn select_tail_for_limit_stops_at_unknown_output() { + let mut access_plan = ParquetAccessPlan::new_all(4); + access_plan.mark_fully_matched(3); + let plan = reversed_plan(access_plan).select_tail_for_limit( + 50, + false, + &ROW_GROUP_METADATA, + ); + + // Row group 3 is taken whole (40 < 50), row group 2 is not fully + // matched, so neither it nor anything after it is cut. + assert_eq!(selected_rows(&plan), vec![None, None, None, None]); + assert!(!plan.has_row_selection); + } + + /// A row group that already has a selection (e.g. from the page index) + /// has an unknown output too. + #[test] + fn select_tail_for_limit_stops_at_existing_selection() { + let page_selection = + RowSelection::from(vec![RowSelector::skip(10), RowSelector::select(30)]); + let mut access_plan = ParquetAccessPlan::new_all(4); + access_plan.scan_selection(3, page_selection.clone()); + let plan = reversed_plan(access_plan).select_tail_for_limit( + 5, + true, + &ROW_GROUP_METADATA, + ); + + assert_eq!( + selected_rows(&plan), + vec![ + Some(page_selection.iter().copied().collect()), + None, + None, + None + ] + ); + } + #[test] fn test_new_from_overall_row_selection() { let row_selection = RowSelection::from(vec![ diff --git a/datafusion/datasource-parquet/src/file_format.rs b/datafusion/datasource-parquet/src/file_format.rs index cfced8bbb9903..d4686c4b636ea 100644 --- a/datafusion/datasource-parquet/src/file_format.rs +++ b/datafusion/datasource-parquet/src/file_format.rs @@ -813,6 +813,7 @@ impl From<&ParquetFormatFactory> for protobuf::TableParquetOptions { pushdown_filters: global_options.global.pushdown_filters, reorder_filters: global_options.global.reorder_filters, force_filter_selections: global_options.global.force_filter_selections, + enable_exact_reverse_scan: global_options.global.enable_exact_reverse_scan, data_pagesize_limit: global_options.global.data_pagesize_limit as u64, write_batch_size: global_options.global.write_batch_size as u64, writer_version: global_options.global.writer_version.to_string(), diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index e188b4543f284..dfe3bb45d0ce8 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -62,6 +62,7 @@ use datafusion_pruning::{FilePruner, PruningPredicate, PruningPredicateBuilder}; #[cfg(feature = "parquet_encryption")] use datafusion_common::config::EncryptionFactoryOptions; +use datafusion_execution::memory_pool::{MemoryConsumer, MemoryPool}; #[cfg(feature = "parquet_encryption")] use datafusion_execution::parquet_encryption::EncryptionFactory; use futures::{FutureExt, StreamExt, future::BoxFuture, stream::BoxStream}; @@ -296,6 +297,11 @@ pub(super) struct ParquetMorselizer { pub row_group_range_assignment: RowGroupRangeAssignment, /// Whether to read row groups in reverse order pub reverse_row_groups: bool, + /// Whether to also reverse the rows of each row group, making a reversed + /// scan exactly ordered. Buffers one decoded row group at a time. + pub reverse_rows: bool, + /// Pool charged for the row group a `reverse_rows` scan buffers. + pub memory_pool: Option>, /// Optional sort order used to reorder row groups by their min/max statistics. pub sort_order_for_reorder: Option, /// Per-scan virtual-column state (validation already performed). `None` @@ -492,6 +498,8 @@ struct PreparedParquetOpen { enable_rle_to_dictionary: bool, row_group_range_assignment: RowGroupRangeAssignment, reverse_row_groups: bool, + reverse_rows: bool, + memory_pool: Option>, sort_order_for_reorder: Option, preserve_order: bool, #[cfg(feature = "parquet_encryption")] @@ -1014,6 +1022,8 @@ impl ParquetMorselizer { enable_rle_to_dictionary: self.enable_rle_to_dictionary, row_group_range_assignment: self.row_group_range_assignment, reverse_row_groups: self.reverse_row_groups, + reverse_rows: self.reverse_rows, + memory_pool: self.memory_pool.clone(), sort_order_for_reorder: self.sort_order_for_reorder.clone(), preserve_order: self.preserve_order, #[cfg(feature = "parquet_encryption")] @@ -1426,6 +1436,29 @@ impl RowGroupsPrunedParquetOpen { return Ok(true); } } + // A row-reversing scan under a limit reads only the tail of the first + // row group it reaches when that row group's output is known up + // front (see `PreparedAccessPlan::select_tail_for_limit`). The + // offset index lets the decoder skip the pages ahead of that tail + // instead of fetching the whole column chunk. + let prepared = &self.prepared.loaded.prepared; + if prepared.reverse_rows + && let Some(limit) = prepared.limit + && let Some(first_read) = row_groups.row_group_indexes().last() + { + let row_group = parquet_metadata.row_group(first_read); + let output_known = + prepared.predicate.is_none() || row_groups.is_fully_matched()[first_read]; + if output_known + && i64::try_from(limit).is_ok_and(|limit| limit < row_group.num_rows()) + && row_group + .columns() + .iter() + .any(|column| column.offset_index_offset().is_some()) + { + return Ok(true); + } + } let Some(page_pruning_predicate) = self.prepared.page_pruning_predicate.as_ref() else { return Ok(false); @@ -1631,8 +1664,13 @@ impl RowGroupsPrunedParquetOpen { let file_metadata = Arc::clone(reader_metadata.metadata()); let rg_metadata = file_metadata.row_groups(); - // Prune by limit if limit is set and limit order is not sensitive - if let (Some(limit), false) = (prepared.limit, prepared.preserve_order) { + // Prune by limit if limit is set and limit order is not sensitive. + // Limits are counted in decode order, which a row reversal does not + // follow: the rows a reversed scan keeps are the last ones of each + // row group, so it gives up limit-based pruning and the decoder + // limit and leaves the limit to the file stream. + let decode_order_limit = prepared.limit.filter(|_| !prepared.reverse_rows); + if let (Some(limit), false) = (decode_order_limit, prepared.preserve_order) { row_groups.prune_by_limit(limit, rg_metadata, &prepared.file_metrics); } @@ -1658,7 +1696,7 @@ impl RowGroupsPrunedParquetOpen { if prepared.preserve_order { None } else { - prepared.limit + decode_order_limit }, ); access_plan = page_pruning_result.access_plan; @@ -1708,6 +1746,17 @@ impl RowGroupsPrunedParquetOpen { if prepared.reverse_row_groups { prepared_plan = prepared_plan.reverse(); } + // A row-reversing scan keeps the last rows of each row group, + // so under a limit it reads only those. + if prepared.reverse_rows + && let Some(limit) = prepared.limit + { + prepared_plan = prepared_plan.select_tail_for_limit( + limit, + prepared.predicate.is_none(), + rg_metadata, + ); + } Ok(prepared_plan) }; @@ -1751,7 +1800,7 @@ impl RowGroupsPrunedParquetOpen { batch_size: prepared.batch_size, arrow_reader_metrics: &arrow_reader_metrics, force_filter_selections: prepared.force_filter_selections, - decoder_limit: prepared.limit, + decoder_limit: decode_order_limit, }; let prepared_access_plan = prepare_access_plan(access_plan)?; @@ -1910,6 +1959,20 @@ impl RowGroupsPrunedParquetOpen { filter_installed, row_filter_skipped_fully_matched, byte_progress, + reverse_rows: prepared.reverse_rows, + reverse_limit: prepared.limit.filter(|_| prepared.reverse_rows), + reversed_batches: Vec::new(), + reversed_reservation: prepared + .memory_pool + .as_ref() + .filter(|_| prepared.reverse_rows) + .map(|pool| { + MemoryConsumer::new(format!( + "ParquetReverseRowGroup[{}]", + prepared.partition_index + )) + .register(pool) + }), } .into_stream(); @@ -2156,6 +2219,7 @@ mod test { use arrow::buffer::BooleanBuffer; use arrow::datatypes::{DataType, Field, Int32Type, Schema, SchemaRef}; use bytes::{BufMut, BytesMut}; + use datafusion_common::DataFusionError; use datafusion_common::{ ColumnStatistics, ScalarValue, Statistics, assert_contains, internal_err, stats::Precision, @@ -2168,6 +2232,7 @@ mod test { CachedFileMetadataEntry, FileMetadataCache, }; use datafusion_execution::cache::default_cache::DefaultCache; + use datafusion_execution::memory_pool::GreedyMemoryPool; use datafusion_execution::object_store::ObjectStoreUrl; use datafusion_expr::{Expr, col, lit}; use datafusion_functions::core::expr_fn::get_field; @@ -2228,6 +2293,8 @@ mod test { enable_rle_to_dictionary: bool, row_group_range_assignment: RowGroupRangeAssignment, reverse_row_groups: bool, + reverse_rows: bool, + memory_pool: Option>, preserve_order: bool, } @@ -2448,6 +2515,8 @@ mod test { enable_rle_to_dictionary: false, row_group_range_assignment: RowGroupRangeAssignment::default(), reverse_row_groups: false, + reverse_rows: false, + memory_pool: None, preserve_order: false, } } @@ -2557,6 +2626,24 @@ mod test { self } + /// Set the flag that reverses rows within each row group. + fn with_reverse_rows(mut self, enable: bool) -> Self { + self.reverse_rows = enable; + self + } + + /// Set the decoder batch size. + fn with_batch_size(mut self, batch_size: usize) -> Self { + self.batch_size = batch_size; + self + } + + /// Set the pool charged for buffered row groups. + fn with_memory_pool(mut self, memory_pool: Arc) -> Self { + self.memory_pool = Some(memory_pool); + self + } + /// Set whether the scan must preserve file order. fn with_preserve_order(mut self, enable: bool) -> Self { self.preserve_order = enable; @@ -2654,6 +2741,8 @@ mod test { enable_rle_to_dictionary: self.enable_rle_to_dictionary, row_group_range_assignment: self.row_group_range_assignment, reverse_row_groups: self.reverse_row_groups, + reverse_rows: self.reverse_rows, + memory_pool: self.memory_pool.clone(), sort_order_for_reorder: None, virtual_state, }) @@ -4438,6 +4527,240 @@ mod test { assert_eq!(batches_forward, batches_reverse); } + /// Writes `test.parquet` with values 1..=12 in row groups of 4 rows. + async fn write_twelve_rows_in_three_row_groups( + store: &Arc, + ) -> (SchemaRef, PartitionedFile) { + let batch = + record_batch!(("a", Int32, (1..=12).map(Some).collect::>())).unwrap(); + let props = WriterProperties::builder() + .set_max_row_group_row_count(Some(4)) + .build(); + let data_len = write_parquet_batches( + Arc::clone(store), + "test.parquet", + vec![batch.clone()], + Some(props), + ) + .await; + let file = PartitionedFile::new( + "test.parquet".to_string(), + u64::try_from(data_len).unwrap(), + ); + (batch.schema(), file) + } + + /// With `reverse_rows`, each row group is emitted back to front, so the + /// whole file comes out exactly reversed. A batch size smaller than the + /// row group makes each row group span several batches, which must be + /// reversed as a whole rather than one batch at a time. + #[tokio::test] + async fn test_exact_reverse_scan_reverses_rows_within_row_groups() { + let store = Arc::new(InMemory::new()) as Arc; + let (schema, file) = write_twelve_rows_in_three_row_groups(&store).await; + + let opener = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(schema) + .with_projection_indices(&[0]) + .with_batch_size(3) + .with_reverse_row_groups(true) + .with_reverse_rows(true) + .build(); + let stream = open_file(&opener, file).await.unwrap(); + + assert_eq!( + collect_int32_values(stream).await, + (1..=12).rev().collect::>() + ); + } + + /// A limit is counted in decode order, which the row reversal does not + /// follow, so a reversed scan must not hand it to the decoder: the rows + /// it keeps are the last ones of each row group, not the first. With no + /// predicate every row group's output is known, so only its last + /// `limit` rows are selected; the file stream stops after the first + /// `limit` rows of the output. + #[tokio::test] + async fn test_exact_reverse_scan_does_not_limit_in_decode_order() { + let store = Arc::new(InMemory::new()) as Arc; + let (schema, file) = write_twelve_rows_in_three_row_groups(&store).await; + + let opener = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(schema) + .with_projection_indices(&[0]) + .with_limit(2) + .with_reverse_row_groups(true) + .with_reverse_rows(true) + .build(); + let stream = open_file(&opener, file).await.unwrap(); + + assert_eq!(collect_int32_values(stream).await, [12, 11, 8, 7, 4, 3]); + } + + /// Under a pushed-down filter a row group's output is known only when + /// statistics prove every row passes. Row group `9..=12` is fully + /// matched by `a > 5` and is cut to its last 3 rows up front; `5..=8` is + /// not, so it is decoded and filtered, and only its last 3 surviving + /// rows are kept; `1..=4` has none. + #[tokio::test] + async fn test_exact_reverse_scan_with_limit_and_filter() { + let store = Arc::new(InMemory::new()) as Arc; + let (schema, file) = write_twelve_rows_in_three_row_groups(&store).await; + let predicate = logical2physical(&col("a").gt(lit(5)), &schema); + + let opener = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(schema) + .with_projection_indices(&[0]) + .with_predicate(predicate) + .with_pushdown_filters(true) + .with_limit(3) + .with_batch_size(2) + .with_reverse_row_groups(true) + .with_reverse_rows(true) + .build(); + let stream = open_file(&opener, file).await.unwrap(); + + assert_eq!(collect_int32_values(stream).await, [12, 11, 10, 8, 7, 6]); + } + + /// Selecting a row group's tail means the rows ahead of it are never + /// read: with one row per page and the offset index loaded, a limited + /// reverse scan fetches fewer bytes than reading the row group whole. + #[tokio::test] + async fn test_exact_reverse_scan_with_limit_skips_leading_pages() { + let store = Arc::new(InMemory::new()) as Arc; + let batch = + record_batch!(("a", Int32, (1..=12).map(Some).collect::>())).unwrap(); + let props = WriterProperties::builder() + .set_data_page_row_count_limit(1) + .set_write_batch_size(1) + .build(); + let data_len = write_parquet_batches( + Arc::clone(&store), + "test.parquet", + vec![batch.clone()], + Some(props), + ) + .await; + let file = PartitionedFile::new( + "test.parquet".to_string(), + u64::try_from(data_len).unwrap(), + ); + + let scan = async |limit: Option| { + let metrics = ExecutionPlanMetricsSet::new(); + let mut builder = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(batch.schema()) + .with_projection_indices(&[0]) + .with_enable_page_index(true) + .with_metrics(metrics.clone()) + .with_reverse_row_groups(true) + .with_reverse_rows(true); + if let Some(limit) = limit { + builder = builder.with_limit(limit); + } + let stream = open_file(&builder.build(), file.clone()).await.unwrap(); + let values = collect_int32_values(stream).await; + (values, counter_metric_value(&metrics, "bytes_scanned")) + }; + + let (all_values, all_bytes) = scan(None).await; + let (tail_values, tail_bytes) = scan(Some(2)).await; + + assert_eq!(all_values, (1..=12).rev().collect::>()); + assert_eq!(tail_values, [12, 11]); + assert!( + tail_bytes < all_bytes, + "limited scan read {tail_bytes} bytes, full scan {all_bytes}" + ); + } + + /// The buffered row group is charged to the memory pool, so a pool too + /// small for it fails the scan with a resources error instead of + /// letting the process grow. + #[tokio::test] + async fn test_exact_reverse_scan_fails_when_memory_pool_is_exhausted() { + let store = Arc::new(InMemory::new()) as Arc; + let (schema, file) = write_twelve_rows_in_three_row_groups(&store).await; + let pool: Arc = Arc::new(GreedyMemoryPool::new(1)); + + let opener = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(schema) + .with_projection_indices(&[0]) + .with_reverse_row_groups(true) + .with_reverse_rows(true) + .with_memory_pool(pool) + .build(); + let mut stream = open_file(&opener, file).await.unwrap(); + + let err = stream.next().await.unwrap().unwrap_err(); + assert!( + matches!(err.find_root(), DataFusionError::ResourcesExhausted(_)), + "expected a resources error, got {err}" + ); + } + + /// The reservation follows the batches held for the current row group: + /// it is non-zero while a reversed row group is being emitted and back + /// to zero once the scan has emitted everything. + #[tokio::test] + async fn test_exact_reverse_scan_releases_memory_reservation() { + let store = Arc::new(InMemory::new()) as Arc; + let (schema, file) = write_twelve_rows_in_three_row_groups(&store).await; + let pool: Arc = Arc::new(GreedyMemoryPool::new(usize::MAX)); + + let opener = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(schema) + .with_projection_indices(&[0]) + .with_batch_size(2) + .with_reverse_row_groups(true) + .with_reverse_rows(true) + .with_memory_pool(Arc::clone(&pool)) + .build(); + let mut stream = open_file(&opener, file).await.unwrap(); + + // First batch of the first row group: its second batch is still held. + stream.next().await.unwrap().unwrap(); + assert!(pool.reserved() > 0); + + while let Some(batch) = stream.next().await { + batch.unwrap(); + } + assert_eq!(pool.reserved(), 0); + } + + /// Rows removed by a pushed-down filter are dropped before the reversal, + /// and the survivors still come out back to front. + #[tokio::test] + async fn test_exact_reverse_scan_with_pushdown_filter() { + let store = Arc::new(InMemory::new()) as Arc; + let (schema, file) = write_twelve_rows_in_three_row_groups(&store).await; + let predicate = logical2physical(&col("a").gt(lit(5)), &schema); + + let opener = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(schema) + .with_projection_indices(&[0]) + .with_predicate(predicate) + .with_pushdown_filters(true) + .with_batch_size(3) + .with_reverse_row_groups(true) + .with_reverse_rows(true) + .build(); + let stream = open_file(&opener, file).await.unwrap(); + + assert_eq!( + collect_int32_values(stream).await, + (6..=12).rev().collect::>() + ); + } + #[tokio::test] async fn test_dynamic_pruning_preserves_external_row_selection() { let store = Arc::new(InMemory::new()) as Arc; diff --git a/datafusion/datasource-parquet/src/push_decoder.rs b/datafusion/datasource-parquet/src/push_decoder.rs index 57df6190daf87..22b34b953fa5e 100644 --- a/datafusion/datasource-parquet/src/push_decoder.rs +++ b/datafusion/datasource-parquet/src/push_decoder.rs @@ -25,7 +25,8 @@ use std::collections::VecDeque; use std::sync::Arc; -use arrow::array::RecordBatch; +use arrow::array::{RecordBatch, UInt32Array}; +use arrow::compute::take_record_batch; use arrow::datatypes::SchemaRef; use futures::StreamExt; use futures::stream::BoxStream; @@ -43,6 +44,7 @@ use parquet::arrow::push_decoder::{ use parquet::file::metadata::ParquetMetaData; use datafusion_common::{DataFusionError, Result, internal_err}; +use datafusion_execution::memory_pool::MemoryReservation; use datafusion_physical_expr::expressions::DynamicFilterTracking; use datafusion_physical_expr_common::physical_expr::PhysicalExpr; use datafusion_physical_plan::metrics::{BaselineMetrics, Count, Gauge}; @@ -307,6 +309,22 @@ pub(crate) struct PushDecoderStreamState { /// group at a time as they are decoded or skipped, and topped up to the /// full range when the stream is dropped. pub(crate) byte_progress: ByteProgress, + /// Emit each row group's rows in reverse order. The active reader is + /// drained into [`Self::reversed_batches`] before anything is returned, + /// so one decoded row group is held in memory at a time. + pub(crate) reverse_rows: bool, + /// The scan's row limit, if any, when `reverse_rows` is set. No more than + /// this many rows of a row group can reach the output, so only its last + /// `limit` rows are kept. + pub(crate) reverse_limit: Option, + /// The current row group's batches, each already row-reversed and kept + /// in decode order, so popping from the end yields the row group back to + /// front. + pub(crate) reversed_batches: Vec, + /// Charges the memory pool for the batches held while a row group is + /// reversed. `None` when the scan does not reverse rows or was opened + /// without a pool. + pub(crate) reversed_reservation: Option, } /// A reusable, `Arc`-shared list of prebuilt row-filter candidates. @@ -432,6 +450,25 @@ impl PushDecoderStreamState { let elapsed_compute = self.baseline_metrics.elapsed_compute().clone(); let mut timer = elapsed_compute.timer(); loop { + // Step 1a: a reversed row group still being emitted. + if let Some(batch) = self.reversed_batches.pop() { + if let Some(reservation) = &self.reversed_reservation { + reservation.shrink(batch.get_array_memory_size()); + } + return Some((Ok(batch), self)); + } + + // Step 1b: in reverse mode, drain the whole row group, reverse it + // into `reversed_batches`, and emit it from step 1a. + if self.reverse_rows + && let Some(reader) = self.active_reader.take() + { + if let Err(e) = self.buffer_reversed_row_group(reader) { + return Some((Err(e), self)); + } + continue; + } + // Step 1: drain a batch from the active reader if any. if let Some(reader) = self.active_reader.as_mut() { match reader.next() { @@ -706,6 +743,82 @@ impl PushDecoderStreamState { fn project_batch(&self, batch: &RecordBatch) -> Result { self.decoder_projection.map(batch) } + + /// Decodes every remaining batch of `reader`'s row group, projects it, + /// reverses its rows, and stores the results in + /// [`Self::reversed_batches`]. Batches stay in decode order there, so + /// popping them yields the row group's rows last to first. + /// + /// Under [`Self::reverse_limit`] only the row group's last `limit` rows + /// are kept: batches that fall entirely before them are dropped as soon + /// as later batches cover the limit, so at most `limit` rows plus one + /// batch are held while decoding. + fn buffer_reversed_row_group( + &mut self, + reader: ParquetRecordBatchReader, + ) -> Result<()> { + let mut kept: VecDeque = VecDeque::new(); + let mut kept_rows = 0; + for batch in reader { + let batch = batch.map_err(DataFusionError::from)?; + self.copy_arrow_reader_metrics(); + let batch = self.project_batch(&batch)?; + if batch.num_rows() == 0 { + continue; + } + if let Some(reservation) = &self.reversed_reservation { + reservation.try_grow(batch.get_array_memory_size())?; + } + kept_rows += batch.num_rows(); + kept.push_back(batch); + if let Some(limit) = self.reverse_limit { + while let Some(front) = kept.front() + && kept_rows - front.num_rows() >= limit + { + kept_rows -= front.num_rows(); + if let Some(dropped) = kept.pop_front() + && let Some(reservation) = &self.reversed_reservation + { + reservation.shrink(dropped.get_array_memory_size()); + } + } + } + } + if let Some(limit) = self.reverse_limit + && kept_rows > limit + && let Some(front) = kept.front_mut() + { + let excess = kept_rows - limit; + *front = front.slice(excess, front.num_rows() - excess); + } + // The reversed copies replace the decoded batches, so the + // reservation is resized to exactly what is stored, which is what + // step 1a releases batch by batch. + for batch in kept { + self.reversed_batches.push(reverse_batch_rows(&batch)?); + } + if let Some(reservation) = &self.reversed_reservation { + let stored = self + .reversed_batches + .iter() + .map(RecordBatch::get_array_memory_size) + .sum(); + reservation.try_resize(stored)?; + } + Ok(()) + } +} + +/// Returns `batch` with its rows in reverse order. +fn reverse_batch_rows(batch: &RecordBatch) -> Result { + let num_rows = u32::try_from(batch.num_rows()).map_err(|_| { + DataFusionError::Internal(format!( + "record batch of {} rows is too large to reverse", + batch.num_rows() + )) + })?; + let indices = UInt32Array::from_iter_values((0..num_rows).rev()); + Ok(take_record_batch(batch, &indices)?) } #[cfg(test)] diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index ac7e10f94e6f5..d89b16f358a87 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -57,6 +57,7 @@ use datafusion_physical_plan::metrics::Count; use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet; use log::warn; +use datafusion_execution::memory_pool::MemoryPool; #[cfg(feature = "parquet_encryption")] use datafusion_execution::parquet_encryption::EncryptionFactory; use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr}; @@ -302,10 +303,21 @@ pub struct ParquetSource { pub(crate) projection: ProjectionExprs, #[cfg(feature = "parquet_encryption")] pub(crate) encryption_factory: Option>, - /// If true, the opener flips row-group iteration order. Within- - /// row-group order is on-disk order, so the scan is `Inexact` and - /// a `SortExec` is kept in the plan. + /// If true, the opener flips row-group iteration order. Unless + /// `reverse_rows` is also set, within-row-group order is on-disk + /// order, so the scan is `Inexact` and a `SortExec` is kept in the + /// plan. reverse_row_groups: bool, + /// If true, the opener also reverses the rows of each row group, so a + /// scan with `reverse_row_groups` reads every file exactly back to + /// front. Set only by an `Exact` reverse sort pushdown (see + /// [`ParquetOptions::enable_exact_reverse_scan`]). + /// + /// [`ParquetOptions::enable_exact_reverse_scan`]: datafusion_common::config::ParquetOptions::enable_exact_reverse_scan + reverse_rows: bool, + /// Pool charged for the row group a `reverse_rows` scan buffers. Set by + /// [`FileSource::with_memory_pool`] when the scan is opened. + memory_pool: Option>, /// Sort order driving `PreparedAccessPlan::reorder_by_statistics` /// in the opener. sort_order_for_reorder: Option, @@ -335,6 +347,8 @@ impl ParquetSource { #[cfg(feature = "parquet_encryption")] encryption_factory: None, reverse_row_groups: false, + reverse_rows: false, + memory_pool: None, sort_order_for_reorder: None, } } @@ -520,6 +534,10 @@ impl ParquetSource { pub(crate) fn reverse_row_groups(&self) -> bool { self.reverse_row_groups } + #[cfg(test)] + pub(crate) fn reverse_rows(&self) -> bool { + self.reverse_rows + } } /// Allows easy conversion from ParquetSource to Arc<dyn FileSource> @@ -638,6 +656,8 @@ impl FileSource for ParquetSource { .global .row_group_range_assignment, reverse_row_groups: self.reverse_row_groups, + reverse_rows: self.reverse_rows, + memory_pool: self.memory_pool.clone(), sort_order_for_reorder: self.sort_order_for_reorder.clone(), virtual_state, })) @@ -686,6 +706,20 @@ impl FileSource for ParquetSource { Arc::new(conf) } + /// Only a `reverse_rows` scan buffers between batches: it holds one row + /// group, which is charged to the pool. + fn with_memory_pool( + &self, + memory_pool: &Arc, + ) -> Option> { + if !self.reverse_rows { + return None; + } + let mut conf = self.clone(); + conf.memory_pool = Some(Arc::clone(memory_pool)); + Some(Arc::new(conf)) + } + fn try_pushdown_projection( &self, projection: &ProjectionExprs, @@ -746,6 +780,9 @@ impl FileSource for ParquetSource { if self.reverse_row_groups { write!(f, ", reverse_row_groups=true")?; } + if self.reverse_rows { + write!(f, ", reverse_rows=true")?; + } // Plan-time marker for dynamic RG-level pruning: if the // predicate is dynamic (e.g. a TopK threshold expression), @@ -1029,6 +1066,27 @@ impl FileSource for ParquetSource { }) }); + // Exact reverse: when the reversed declared ordering satisfies the + // request and `enable_exact_reverse_scan` is set, read every file + // exactly back to front, row groups and the rows inside them, so + // the scan itself produces the requested order. Row groups keep + // their on-disk order before the flip: reordering them by + // statistics is not order-preserving when two row groups share a + // `min`. `FileScanConfig` reverses the files of each group and + // re-validates the ordering, falling back to `Inexact` if it does + // not hold. + if reversed_satisfies + && self.table_parquet_options.global.enable_exact_reverse_scan + { + let mut new_source = self.clone(); + new_source.sort_order_for_reorder = None; + new_source.reverse_row_groups = true; + new_source.reverse_rows = true; + return Ok(SortOrderPushdownResult::Exact { + inner: Arc::new(new_source) as Arc, + }); + } + if !column_in_file_schema && !reversed_satisfies { return Ok(SortOrderPushdownResult::Unsupported); } @@ -1101,6 +1159,9 @@ impl FileSource for ParquetSource { #[cfg(feature = "parquet_encryption")] encryption_factory: _, reverse_row_groups, + reverse_rows, + // Runtime only: set when the scan is opened. + memory_pool: _, sort_order_for_reorder, } = self; @@ -1133,6 +1194,7 @@ impl FileSource for ParquetSource { parquet_options: Some(table_parquet_options.try_into()?), sort_order_for_reorder, reverse_row_groups: *reverse_row_groups, + reverse_rows: *reverse_rows, metadata_size_hint, }; Ok(Some(protobuf::PhysicalPlanNode { @@ -1174,6 +1236,7 @@ impl ParquetSource { parquet_options, sort_order_for_reorder, reverse_row_groups, + reverse_rows, metadata_size_hint, } = scan; @@ -1254,6 +1317,7 @@ impl ParquetSource { .with_table_parquet_options(options); source.sort_order_for_reorder = sort_order_for_reorder; source.reverse_row_groups = *reverse_row_groups; + source.reverse_rows = *reverse_rows; source.metadata_size_hint = metadata_size_hint; if let Some(predicate) = predicate { @@ -1696,6 +1760,68 @@ mod tests { ); } + /// Source declares `[a ASC NULLS LAST]`, request is its exact reverse + /// `[a DESC NULLS FIRST]`, used by the exact reverse scan tests below. + fn exact_reverse_case() -> (Arc, EquivalenceProperties, Vec) + { + use arrow::compute::SortOptions; + use pushdown_sort_helpers::*; + + let schema = schema_with_a_int(); + let column = Arc::new(Column::new("a", 0)) as Arc; + let mut eq = EquivalenceProperties::new(Arc::clone(&schema)); + eq.add_ordering(vec![PhysicalSortExpr::new( + Arc::clone(&column), + SortOptions::new(false, false), + )]); + let order = vec![PhysicalSortExpr::new(column, SortOptions::new(true, true))]; + (schema, eq, order) + } + + /// With `enable_exact_reverse_scan`, a request that is the reverse of + /// the declared ordering is `Exact`: the source reverses both the row + /// groups and the rows inside them, and does not reorder row groups by + /// statistics, which is not order-preserving for row groups that share + /// a `min`. + #[test] + fn try_pushdown_sort_returns_exact_for_reverse_when_enabled() { + let (schema, eq, order) = exact_reverse_case(); + let mut options = TableParquetOptions::default(); + options.global.enable_exact_reverse_scan = true; + let source = ParquetSource::new(schema).with_table_parquet_options(options); + + let result = source.try_pushdown_sort(&order, &eq).unwrap(); + + let SortOrderPushdownResult::Exact { inner } = result else { + panic!("expected Exact, got a different variant"); + }; + let inner_parquet = inner + .downcast_ref::() + .expect("inner is ParquetSource"); + assert!(inner_parquet.reverse_row_groups()); + assert!(inner_parquet.reverse_rows()); + assert!(inner_parquet.sort_order_for_reorder.is_none()); + } + + /// Without the option, the same request stays `Inexact` and rows keep + /// their on-disk order within each row group. + #[test] + fn try_pushdown_sort_returns_inexact_for_reverse_by_default() { + let (schema, eq, order) = exact_reverse_case(); + let source = ParquetSource::new(schema); + + let result = source.try_pushdown_sort(&order, &eq).unwrap(); + + let SortOrderPushdownResult::Inexact { inner } = result else { + panic!("expected Inexact, got a different variant"); + }; + let inner_parquet = inner + .downcast_ref::() + .expect("inner is ParquetSource"); + assert!(inner_parquet.reverse_row_groups()); + assert!(!inner_parquet.reverse_rows()); + } + /// Regression: when the source declares `[a DESC]` and the request is /// `[a ASC]`, both `column_in_file_schema` and `reversed_satisfies` /// are true. `reverse_row_groups` must follow the *request's* diff --git a/datafusion/datasource/src/file.rs b/datafusion/datasource/src/file.rs index df6b428df940a..1e80a97668860 100644 --- a/datafusion/datasource/src/file.rs +++ b/datafusion/datasource/src/file.rs @@ -18,6 +18,7 @@ use crate::schema_adapter::SchemaAdapterFactory; use datafusion_common::config::ConfigOptions; use datafusion_common::tree_node::TreeNodeRecursion; use datafusion_common::{Result, not_impl_err}; +use datafusion_execution::memory_pool::MemoryPool; use datafusion_physical_expr::projection::ProjectionExprs; use datafusion_physical_expr::{EquivalenceProperties, LexOrdering, PhysicalExpr}; use datafusion_physical_plan::DisplayFormatType; @@ -94,6 +95,18 @@ pub trait FileSource: Any + Send + Sync { /// Initialize new type with batch size configuration fn with_batch_size(&self, batch_size: usize) -> Arc; + /// Returns a copy of this source that charges the memory it buffers + /// while scanning to `pool`, or `None` (the default) when the source + /// keeps nothing worth accounting for between the batches it emits. + /// + /// Called with the task's memory pool before the scan is opened. + fn with_memory_pool( + &self, + _memory_pool: &Arc, + ) -> Option> { + None + } + /// Returns the filter expression that will be applied *during* the file scan. /// /// These expressions are in terms of the unprojected [`Self::table_schema`]. diff --git a/datafusion/datasource/src/file_scan_config/mod.rs b/datafusion/datasource/src/file_scan_config/mod.rs index 384fbbf3c4450..7805a440d4f7f 100644 --- a/datafusion/datasource/src/file_scan_config/mod.rs +++ b/datafusion/datasource/src/file_scan_config/mod.rs @@ -760,6 +760,9 @@ impl DataSource for FileScanConfig { .unwrap_or_else(|| context.session_config().batch_size()); let source = self.file_source.with_batch_size(batch_size); + let source = source + .with_memory_pool(context.memory_pool()) + .unwrap_or(source); let morselizer = source.create_morselizer(object_store, self, partition)?; @@ -3734,6 +3737,128 @@ mod tests { Ok(()) } + /// A source that is `Exact` for the reverse of the declared ordering + /// (it reads every file back to front): the files of each group are + /// reversed and the scan declares the reversed ordering. + #[test] + fn sort_pushdown_exact_reverse_reverses_files_and_ordering() -> Result<()> { + let file_schema = + Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); + let table_schema = TableSchema::from(&file_schema); + let file_source = Arc::new(ExactSortPushdownSource::new(table_schema)); + let sort_expr = PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0))); + let file_groups = vec![FileGroup::new(vec![ + make_file_with_stats("file1", 0.0, 9.0), + make_file_with_stats("file2", 10.0, 19.0), + make_file_with_stats("file3", 20.0, 30.0), + ])]; + let config = + FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), file_source) + .with_file_groups(file_groups) + .with_output_ordering(vec![ + LexOrdering::new(vec![sort_expr.clone()]).unwrap(), + ]) + .build(); + + let reversed = sort_expr.reverse(); + let result = config.try_pushdown_sort(std::slice::from_ref(&reversed))?; + + let SortOrderPushdownResult::Exact { inner } = result else { + panic!("Expected Exact result, got {result:?}"); + }; + let pushed_config = inner + .downcast_ref::() + .expect("Expected FileScanConfig"); + let names: Vec<_> = pushed_config.file_groups[0] + .files() + .iter() + .map(|f| f.object_meta.location.as_ref()) + .collect(); + assert_eq!(names, ["file3", "file2", "file1"]); + assert_eq!( + pushed_config.output_ordering, + vec![LexOrdering::new(vec![reversed]).unwrap()] + ); + Ok(()) + } + + /// Reading overlapping files back to front does not give an ordered + /// stream, so an `Exact` reverse claim is downgraded to `Inexact`. + #[test] + fn sort_pushdown_exact_reverse_overlapping_downgraded_to_inexact() -> Result<()> { + let file_schema = + Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); + let table_schema = TableSchema::from(&file_schema); + let file_source = Arc::new(ExactSortPushdownSource::new(table_schema)); + let sort_expr = PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0))); + let file_groups = vec![FileGroup::new(vec![ + make_file_with_stats("file1", 0.0, 15.0), + make_file_with_stats("file2", 10.0, 19.0), + ])]; + let config = + FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), file_source) + .with_file_groups(file_groups) + .with_output_ordering(vec![ + LexOrdering::new(vec![sort_expr.clone()]).unwrap(), + ]) + .build(); + + let result = config.try_pushdown_sort(&[sort_expr.reverse()])?; + + let SortOrderPushdownResult::Inexact { inner } = result else { + panic!("Expected Inexact result, got {result:?}"); + }; + let pushed_config = inner + .downcast_ref::() + .expect("Expected FileScanConfig"); + assert!(pushed_config.output_ordering.is_empty()); + Ok(()) + } + + /// A source can be `Exact` for a request that only its equivalence + /// reasoning recognises as the reverse of the declared ordering, here + /// `a DESC NULLS LAST` on a non-nullable `a` declared `a ASC NULLS LAST`. + /// `FileScanConfig` compares the orderings literally, so it sorts the + /// files by statistics instead of taking its reverse path, and the + /// forward ordering it would keep no longer describes the scan: the + /// claim is downgraded to `Inexact` and the sort is kept. + #[test] + fn sort_pushdown_exact_without_file_reversal_downgraded_to_inexact() -> Result<()> { + let file_schema = + Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); + let table_schema = TableSchema::from(&file_schema); + let file_source = Arc::new(ExactSortPushdownSource::new(table_schema)); + let column = Arc::new(Column::new("a", 0)) as Arc; + let asc_nulls_last = PhysicalSortExpr::new( + Arc::clone(&column), + arrow::compute::SortOptions::new(false, false), + ); + let file_groups = vec![FileGroup::new(vec![ + make_file_with_stats("file1", 0.0, 9.0), + make_file_with_stats("file2", 10.0, 19.0), + ])]; + let config = + FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), file_source) + .with_file_groups(file_groups) + .with_output_ordering(vec![ + LexOrdering::new(vec![asc_nulls_last]).unwrap(), + ]) + .build(); + let desc_nulls_last = + PhysicalSortExpr::new(column, arrow::compute::SortOptions::new(true, false)); + + let result = config.try_pushdown_sort(&[desc_nulls_last])?; + + let SortOrderPushdownResult::Inexact { inner } = result else { + panic!("Expected Inexact result, got {result:?}"); + }; + let pushed_config = inner + .downcast_ref::() + .expect("Expected FileScanConfig"); + assert!(pushed_config.output_ordering.is_empty()); + Ok(()) + } + #[test] fn sort_pushdown_exact_source_overlapping_downgraded_to_inexact() -> Result<()> { let file_schema = diff --git a/datafusion/datasource/src/file_scan_config/sort_pushdown.rs b/datafusion/datasource/src/file_scan_config/sort_pushdown.rs index 1c023354cfd10..6aa90e805e80a 100644 --- a/datafusion/datasource/src/file_scan_config/sort_pushdown.rs +++ b/datafusion/datasource/src/file_scan_config/sort_pushdown.rs @@ -97,6 +97,29 @@ impl FileScanConfig { new_config.file_source = new_file_source; + // Exact reverse: the source reads every file back to front, rows + // within each row group included, so with each group's files + // reversed too the scan emits the declared orderings reversed. + // Re-validate against the reversed groups before keeping them; + // if the ordering does not hold, clear it and the caller falls + // back to `Inexact`, which keeps the `SortExec`. + if reverse_file_groups && is_exact { + new_config.output_ordering = self + .output_ordering + .iter() + .filter_map(|ordering| { + LexOrdering::new(ordering.iter().map(|expr| expr.reverse())) + }) + .collect(); + if !new_config + .eq_properties() + .ordering_satisfy(order.iter().cloned())? + { + new_config.output_ordering = vec![]; + } + return Ok(new_config); + } + // Sort files within groups by statistics when not reversing let all_non_overlapping = if !reverse_file_groups { if let Some(sort_order) = LexOrdering::new(order.iter().cloned()) { @@ -120,8 +143,8 @@ impl FileScanConfig { } else { // When reversing, files are already reversed above. We skip // statistics-based sorting here because it would undo the reversal. - // Note: reverse path is always Inexact, so all_non_overlapping - // is not used (is_exact is false). + // The exact reverse path returned above, so this is `Inexact` + // and `all_non_overlapping` is not used. false }; @@ -161,8 +184,14 @@ impl FileScanConfig { // stream isn't ordered, so `output_ordering` must be dropped. (false, _) => false, // Source already had validated ordering and the post-sort - // files still don't overlap — Exact carries through. - (true, true) => true, + // files still don't overlap — Exact carries through, provided + // the forward ordering really satisfies the request. A source + // that claims `Exact` by reading backwards relies on the file + // groups being reversed above; if they were not, the declared + // ordering does not describe its output, so do not keep it. + (true, true) => new_config + .eq_properties() + .ordering_satisfy(order.iter().cloned())?, // Source returned `Inexact`; re-validate against the // reordered file groups to decide whether to upgrade. // diff --git a/datafusion/proto-common/proto/datafusion_common.proto b/datafusion/proto-common/proto/datafusion_common.proto index 49070439c3d8f..1cd523a38cbed 100644 --- a/datafusion/proto-common/proto/datafusion_common.proto +++ b/datafusion/proto-common/proto/datafusion_common.proto @@ -554,6 +554,7 @@ message ParquetOptions { bool pushdown_filters = 5; // default = false bool reorder_filters = 6; // default = false bool force_filter_selections = 34; // default = false + bool enable_exact_reverse_scan = 42; // default = false uint64 data_pagesize_limit = 7; // default = 1024 * 1024 uint64 write_batch_size = 8; // default = 1024 string writer_version = 9; // default = "1.0" diff --git a/datafusion/proto-common/src/from_proto/mod.rs b/datafusion/proto-common/src/from_proto/mod.rs index fd94fc38d5901..5e858f95b2f83 100644 --- a/datafusion/proto-common/src/from_proto/mod.rs +++ b/datafusion/proto-common/src/from_proto/mod.rs @@ -1190,6 +1190,7 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { pushdown_filters: value.pushdown_filters, reorder_filters: value.reorder_filters, force_filter_selections: value.force_filter_selections, + enable_exact_reverse_scan: value.enable_exact_reverse_scan, data_pagesize_limit: to_usize( value.data_pagesize_limit, "data_pagesize_limit", diff --git a/datafusion/proto-common/src/generated/pbjson.rs b/datafusion/proto-common/src/generated/pbjson.rs index 09386a01eb556..26f91ef294186 100644 --- a/datafusion/proto-common/src/generated/pbjson.rs +++ b/datafusion/proto-common/src/generated/pbjson.rs @@ -6463,6 +6463,9 @@ impl serde::Serialize for ParquetOptions { if self.force_filter_selections { len += 1; } + if self.enable_exact_reverse_scan { + len += 1; + } if self.data_pagesize_limit != 0 { len += 1; } @@ -6581,6 +6584,9 @@ impl serde::Serialize for ParquetOptions { if self.force_filter_selections { struct_ser.serialize_field("forceFilterSelections", &self.force_filter_selections)?; } + if self.enable_exact_reverse_scan { + struct_ser.serialize_field("enableExactReverseScan", &self.enable_exact_reverse_scan)?; + } if self.data_pagesize_limit != 0 { #[allow(clippy::needless_borrow)] #[allow(clippy::needless_borrows_for_generic_args)] @@ -6781,6 +6787,8 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "reorderFilters", "force_filter_selections", "forceFilterSelections", + "enable_exact_reverse_scan", + "enableExactReverseScan", "data_pagesize_limit", "dataPagesizeLimit", "write_batch_size", @@ -6855,6 +6863,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { PushdownFilters, ReorderFilters, ForceFilterSelections, + EnableExactReverseScan, DataPagesizeLimit, WriteBatchSize, WriterVersion, @@ -6915,6 +6924,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "pushdownFilters" | "pushdown_filters" => Ok(GeneratedField::PushdownFilters), "reorderFilters" | "reorder_filters" => Ok(GeneratedField::ReorderFilters), "forceFilterSelections" | "force_filter_selections" => Ok(GeneratedField::ForceFilterSelections), + "enableExactReverseScan" | "enable_exact_reverse_scan" => Ok(GeneratedField::EnableExactReverseScan), "dataPagesizeLimit" | "data_pagesize_limit" => Ok(GeneratedField::DataPagesizeLimit), "writeBatchSize" | "write_batch_size" => Ok(GeneratedField::WriteBatchSize), "writerVersion" | "writer_version" => Ok(GeneratedField::WriterVersion), @@ -6973,6 +6983,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { let mut pushdown_filters__ = None; let mut reorder_filters__ = None; let mut force_filter_selections__ = None; + let mut enable_exact_reverse_scan__ = None; let mut data_pagesize_limit__ = None; let mut write_batch_size__ = None; let mut writer_version__ = None; @@ -7044,6 +7055,12 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { } force_filter_selections__ = Some(map_.next_value()?); } + GeneratedField::EnableExactReverseScan => { + if enable_exact_reverse_scan__.is_some() { + return Err(serde::de::Error::duplicate_field("enableExactReverseScan")); + } + enable_exact_reverse_scan__ = Some(map_.next_value()?); + } GeneratedField::DataPagesizeLimit => { if data_pagesize_limit__.is_some() { return Err(serde::de::Error::duplicate_field("dataPagesizeLimit")); @@ -7267,6 +7284,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { pushdown_filters: pushdown_filters__.unwrap_or_default(), reorder_filters: reorder_filters__.unwrap_or_default(), force_filter_selections: force_filter_selections__.unwrap_or_default(), + enable_exact_reverse_scan: enable_exact_reverse_scan__.unwrap_or_default(), data_pagesize_limit: data_pagesize_limit__.unwrap_or_default(), write_batch_size: write_batch_size__.unwrap_or_default(), writer_version: writer_version__.unwrap_or_default(), diff --git a/datafusion/proto-common/src/generated/prost.rs b/datafusion/proto-common/src/generated/prost.rs index ee586fa75ee37..68ccbc6e58ad6 100644 --- a/datafusion/proto-common/src/generated/prost.rs +++ b/datafusion/proto-common/src/generated/prost.rs @@ -834,6 +834,9 @@ pub struct ParquetOptions { /// default = false #[prost(bool, tag = "34")] pub force_filter_selections: bool, + /// default = false + #[prost(bool, tag = "42")] + pub enable_exact_reverse_scan: bool, /// default = 1024 * 1024 #[prost(uint64, tag = "7")] pub data_pagesize_limit: u64, diff --git a/datafusion/proto-common/src/to_proto/mod.rs b/datafusion/proto-common/src/to_proto/mod.rs index 80a9dc662db5e..e935a712b6df2 100644 --- a/datafusion/proto-common/src/to_proto/mod.rs +++ b/datafusion/proto-common/src/to_proto/mod.rs @@ -928,6 +928,7 @@ impl TryFrom<&ParquetOptions> for protobuf::ParquetOptions { pushdown_filters: value.pushdown_filters, reorder_filters: value.reorder_filters, force_filter_selections: value.force_filter_selections, + enable_exact_reverse_scan: value.enable_exact_reverse_scan, data_pagesize_limit: value.data_pagesize_limit as u64, write_batch_size: value.write_batch_size as u64, writer_version: value.writer_version.to_string(), diff --git a/datafusion/proto-models/proto/datafusion.proto b/datafusion/proto-models/proto/datafusion.proto index e16efc9999419..6507cf0bb8a95 100644 --- a/datafusion/proto-models/proto/datafusion.proto +++ b/datafusion/proto-models/proto/datafusion.proto @@ -1382,6 +1382,10 @@ message ParquetScanExecNode { // Source-specific footer prefetch size. Absent means no hint. optional uint64 metadata_size_hint = 7; + + // Reverse the rows within each row group too, making a reversed scan + // exactly ordered. Only meaningful with `reverse_row_groups`. + bool reverse_rows = 8; } message CsvScanExecNode { diff --git a/datafusion/proto-models/src/from_proto.rs b/datafusion/proto-models/src/from_proto.rs index 5ec8939f8b530..ded42ebe5c632 100644 --- a/datafusion/proto-models/src/from_proto.rs +++ b/datafusion/proto-models/src/from_proto.rs @@ -362,6 +362,7 @@ impl TryFrom<&ParquetOptionsProto> for ParquetOptions { pushdown_filters: proto.pushdown_filters, reorder_filters: proto.reorder_filters, force_filter_selections: proto.force_filter_selections, + enable_exact_reverse_scan: proto.enable_exact_reverse_scan, data_pagesize_limit: to_usize( proto.data_pagesize_limit, "data_pagesize_limit", diff --git a/datafusion/proto-models/src/generated/datafusion_proto_common.rs b/datafusion/proto-models/src/generated/datafusion_proto_common.rs index ee586fa75ee37..68ccbc6e58ad6 100644 --- a/datafusion/proto-models/src/generated/datafusion_proto_common.rs +++ b/datafusion/proto-models/src/generated/datafusion_proto_common.rs @@ -834,6 +834,9 @@ pub struct ParquetOptions { /// default = false #[prost(bool, tag = "34")] pub force_filter_selections: bool, + /// default = false + #[prost(bool, tag = "42")] + pub enable_exact_reverse_scan: bool, /// default = 1024 * 1024 #[prost(uint64, tag = "7")] pub data_pagesize_limit: u64, diff --git a/datafusion/proto-models/src/generated/pbjson.rs b/datafusion/proto-models/src/generated/pbjson.rs index 725cceeb75970..68cb151e41943 100644 --- a/datafusion/proto-models/src/generated/pbjson.rs +++ b/datafusion/proto-models/src/generated/pbjson.rs @@ -16926,6 +16926,9 @@ impl serde::Serialize for ParquetScanExecNode { if self.metadata_size_hint.is_some() { len += 1; } + if self.reverse_rows { + len += 1; + } let mut struct_ser = serializer.serialize_struct("datafusion.ParquetScanExecNode", len)?; if let Some(v) = self.base_conf.as_ref() { struct_ser.serialize_field("baseConf", v)?; @@ -16947,6 +16950,9 @@ impl serde::Serialize for ParquetScanExecNode { #[allow(clippy::needless_borrows_for_generic_args)] struct_ser.serialize_field("metadataSizeHint", ToString::to_string(&v).as_str())?; } + if self.reverse_rows { + struct_ser.serialize_field("reverseRows", &self.reverse_rows)?; + } struct_ser.end() } } @@ -16968,6 +16974,8 @@ impl<'de> serde::Deserialize<'de> for ParquetScanExecNode { "reverseRowGroups", "metadata_size_hint", "metadataSizeHint", + "reverse_rows", + "reverseRows", ]; #[allow(clippy::enum_variant_names)] @@ -16978,6 +16986,7 @@ impl<'de> serde::Deserialize<'de> for ParquetScanExecNode { SortOrderForReorder, ReverseRowGroups, MetadataSizeHint, + ReverseRows, } impl<'de> serde::Deserialize<'de> for GeneratedField { fn deserialize(deserializer: D) -> std::result::Result @@ -17005,6 +17014,7 @@ impl<'de> serde::Deserialize<'de> for ParquetScanExecNode { "sortOrderForReorder" | "sort_order_for_reorder" => Ok(GeneratedField::SortOrderForReorder), "reverseRowGroups" | "reverse_row_groups" => Ok(GeneratedField::ReverseRowGroups), "metadataSizeHint" | "metadata_size_hint" => Ok(GeneratedField::MetadataSizeHint), + "reverseRows" | "reverse_rows" => Ok(GeneratedField::ReverseRows), _ => Err(serde::de::Error::unknown_field(value, FIELDS)), } } @@ -17030,6 +17040,7 @@ impl<'de> serde::Deserialize<'de> for ParquetScanExecNode { let mut sort_order_for_reorder__ = None; let mut reverse_row_groups__ = None; let mut metadata_size_hint__ = None; + let mut reverse_rows__ = None; while let Some(k) = map_.next_key()? { match k { GeneratedField::BaseConf => { @@ -17070,6 +17081,12 @@ impl<'de> serde::Deserialize<'de> for ParquetScanExecNode { map_.next_value::<::std::option::Option<::pbjson::private::NumberDeserialize<_>>>()?.map(|x| x.0) ; } + GeneratedField::ReverseRows => { + if reverse_rows__.is_some() { + return Err(serde::de::Error::duplicate_field("reverseRows")); + } + reverse_rows__ = Some(map_.next_value()?); + } } } Ok(ParquetScanExecNode { @@ -17079,6 +17096,7 @@ impl<'de> serde::Deserialize<'de> for ParquetScanExecNode { sort_order_for_reorder: sort_order_for_reorder__, reverse_row_groups: reverse_row_groups__.unwrap_or_default(), metadata_size_hint: metadata_size_hint__, + reverse_rows: reverse_rows__.unwrap_or_default(), }) } } diff --git a/datafusion/proto-models/src/generated/prost.rs b/datafusion/proto-models/src/generated/prost.rs index 1034d0443dbce..0bcfed6904773 100644 --- a/datafusion/proto-models/src/generated/prost.rs +++ b/datafusion/proto-models/src/generated/prost.rs @@ -2126,6 +2126,10 @@ pub struct ParquetScanExecNode { /// Source-specific footer prefetch size. Absent means no hint. #[prost(uint64, optional, tag = "7")] pub metadata_size_hint: ::core::option::Option, + /// Reverse the rows within each row group too, making a reversed scan + /// exactly ordered. Only meaningful with `reverse_row_groups`. + #[prost(bool, tag = "8")] + pub reverse_rows: bool, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct CsvScanExecNode { diff --git a/datafusion/proto/tests/cases/plans/sources.rs b/datafusion/proto/tests/cases/plans/sources.rs index 32909afddf83b..9ccc35dfc7828 100644 --- a/datafusion/proto/tests/cases/plans/sources.rs +++ b/datafusion/proto/tests/cases/plans/sources.rs @@ -35,7 +35,7 @@ use datafusion::physical_plan::{ ExecutionPlan, Partitioning, PhysicalExpr, RangePartitioning, SplitPoint, Statistics, displayable, }; -use datafusion::prelude::SessionContext; +use datafusion::prelude::{ParquetReadOptions, SessionConfig, SessionContext}; use datafusion::scalar::ScalarValue; use datafusion_common::config::TableParquetOptions; use datafusion_common::stats::Precision; @@ -179,6 +179,51 @@ async fn roundtrip_parquet_exec_with_sort_pushdown() -> Result<()> { Ok(()) } +/// An exact reverse scan carries `reverse_rows` alongside +/// `reverse_row_groups`; losing it in the round trip would make the decoded +/// scan emit rows in on-disk order under a plan with no `SortExec` left. +#[tokio::test] +async fn roundtrip_parquet_exec_with_exact_reverse_scan() -> Result<()> { + let mut config = SessionConfig::new(); + config + .options_mut() + .execution + .parquet + .enable_exact_reverse_scan = true; + let ctx = SessionContext::new_with_config(config); + let testdata = datafusion::test_util::parquet_test_data(); + ctx.register_parquet( + "alltypes_plain", + &format!("{testdata}/alltypes_plain.parquet"), + ParquetReadOptions { + file_sort_order: vec![vec![datafusion::prelude::col("id").sort(true, false)]], + ..Default::default() + }, + ) + .await?; + let plan = ctx + .sql("SELECT id FROM alltypes_plain ORDER BY id DESC NULLS FIRST LIMIT 5") + .await? + .create_physical_plan() + .await?; + let before = displayable(plan.as_ref()).indent(true).to_string(); + assert!( + before.contains("reverse_row_groups=true, reverse_rows=true") + && !before.contains("SortExec"), + "expected an exact reverse scan in plan:\n{before}" + ); + + let roundtripped = roundtrip_test_and_return( + plan, + &ctx, + &DefaultPhysicalExtensionCodec {}, + &DefaultPhysicalProtoConverter {}, + )?; + let after = displayable(roundtripped.as_ref()).indent(true).to_string(); + pretty_assertions::assert_eq!(before, after); + Ok(()) +} + #[test] fn file_scan_rejects_zero_batch_size() -> Result<()> { let schema = Arc::new(Schema::empty()); diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index 4355051ef0254..78466602ca2b9 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -242,6 +242,7 @@ datafusion.execution.parquet.data_page_row_count_limit 20000 datafusion.execution.parquet.data_pagesize_limit 1048576 datafusion.execution.parquet.dictionary_enabled true datafusion.execution.parquet.dictionary_page_size_limit 1048576 +datafusion.execution.parquet.enable_exact_reverse_scan false datafusion.execution.parquet.enable_page_index true datafusion.execution.parquet.enable_rle_to_dictionary false datafusion.execution.parquet.encoding NULL @@ -409,6 +410,7 @@ datafusion.execution.parquet.data_page_row_count_limit 20000 (writing) Sets best datafusion.execution.parquet.data_pagesize_limit 1048576 (writing) Sets best effort maximum size of data page in bytes datafusion.execution.parquet.dictionary_enabled true (writing) Sets if dictionary encoding is enabled. If NULL, uses default parquet writer setting datafusion.execution.parquet.dictionary_page_size_limit 1048576 (writing) Sets best effort maximum dictionary page size, in bytes +datafusion.execution.parquet.enable_exact_reverse_scan false (reading) If true, a scan whose requested ordering is the reverse of the files' declared ordering reads each row group into memory and emits its rows in reverse, so the scan is exactly ordered and the `SortExec` above it can be removed. Peak memory per scanned file is one decoded row group (after projection and filtering). If false, only the row group order is reversed and the sort is kept. datafusion.execution.parquet.enable_page_index true (reading) If true, reads the Parquet data page level metadata (the Page Index), if present, to reduce the I/O and number of rows decoded. datafusion.execution.parquet.enable_rle_to_dictionary false (reading) If true, top-level string and binary Parquet columns with dictionary pages are inferred and scanned as `Dictionary` / `Dictionary` instead of their plain value type. This applies only when DataFusion infers the table schema. Tables with a user-supplied schema are not promoted because the Parquet footer is not read at DDL time, so dictionary pages cannot be detected per column. See datafusion.execution.parquet.encoding NULL (writing) Sets default encoding for any column. Valid values are: plain, plain_dictionary, rle, bit_packed, delta_binary_packed, delta_length_byte_array, delta_byte_array, rle_dictionary, byte_stream_split, and alp. These values are not case sensitive. If NULL, uses default parquet writer setting diff --git a/datafusion/sqllogictest/test_files/sort_pushdown_exact_reverse.slt b/datafusion/sqllogictest/test_files/sort_pushdown_exact_reverse.slt new file mode 100644 index 0000000000000..dc0c52ff5b74b --- /dev/null +++ b/datafusion/sqllogictest/test_files/sort_pushdown_exact_reverse.slt @@ -0,0 +1,366 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership +# SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements + +# Tests for `datafusion.execution.parquet.enable_exact_reverse_scan`. +# +# With the option on, a scan whose requested ordering is the reverse of the +# declared file ordering reads each file back to front, rows within each row +# group included, so the scan is exactly ordered and the `SortExec` above it +# is removed. With it off, only the row group order is reversed and the sort +# is kept (see sort_pushdown.slt). + +statement ok +SET datafusion.execution.target_partitions = 1; + +statement ok +SET datafusion.optimizer.enable_sort_pushdown = true; + +statement ok +SET datafusion.execution.parquet.enable_exact_reverse_scan = true; + +# Small row groups so every file has several of them. +statement ok +SET datafusion.execution.parquet.max_row_group_size = 2; + +statement ok +CREATE TABLE src(id INT, category VARCHAR) AS VALUES +(1, 'a'), (2, 'b'), (3, 'a'), (4, 'b'), (5, 'a'), +(6, 'b'), (7, 'a'), (8, 'b'), (9, 'a'), (10, 'b'); + +query I +COPY (SELECT * FROM src ORDER BY id ASC) +TO 'test_files/scratch/sort_pushdown_exact_reverse/single/data.parquet'; +---- +10 + +statement ok +CREATE EXTERNAL TABLE single_file(id INT, category VARCHAR) +STORED AS PARQUET +LOCATION 'test_files/scratch/sort_pushdown_exact_reverse/single/' +WITH ORDER (id ASC); + +# Test 1: DESC LIMIT on an ASC file: no SortExec, the limit is pushed into +# the scan, and the scan declares the reversed ordering. +query TT +EXPLAIN SELECT * FROM single_file ORDER BY id DESC LIMIT 3; +---- +logical_plan +01)Sort: single_file.id DESC NULLS FIRST, fetch=3 +02)--TableScan: single_file projection=[id, category] +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/single/data.parquet]]}, projection=[id, category], limit=3, output_ordering=[id@0 DESC], file_type=parquet, reverse_row_groups=true, reverse_rows=true + +query IT +SELECT * FROM single_file ORDER BY id DESC LIMIT 3; +---- +10 b +9 a +8 b + +# Test 2: the full reversed scan, across all five row groups. +query IT +SELECT * FROM single_file ORDER BY id DESC; +---- +10 b +9 a +8 b +7 a +6 b +5 a +4 b +3 a +2 b +1 a + +# Test 3: a pushed-down filter drops rows before they are reversed. +statement ok +SET datafusion.execution.parquet.pushdown_filters = true; + +query TT +EXPLAIN SELECT * FROM single_file WHERE category = 'a' ORDER BY id DESC LIMIT 4; +---- +logical_plan +01)Sort: single_file.id DESC NULLS FIRST, fetch=4 +02)--Filter: single_file.category = Utf8View("a") +03)----TableScan: single_file projection=[id, category], partial_filters=[single_file.category = Utf8View("a")] +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/single/data.parquet]]}, projection=[id, category], limit=4, output_ordering=[id@0 DESC], file_type=parquet, predicate=category@1 = a, reverse_row_groups=true, reverse_rows=true, pruning_predicate=category_null_count@2 != row_count@3 AND category_min@0 <= a AND a <= category_max@1, required_guarantees=[category in (a)] + +query IT +SELECT * FROM single_file WHERE category = 'a' ORDER BY id DESC LIMIT 4; +---- +9 a +7 a +5 a +3 a + +statement ok +SET datafusion.execution.parquet.pushdown_filters = false; + +# Test 4: several non-overlapping files in one group are read in reverse +# file order. +query I +COPY (SELECT * FROM src WHERE id <= 5 ORDER BY id ASC) +TO 'test_files/scratch/sort_pushdown_exact_reverse/multi/a.parquet'; +---- +5 + +query I +COPY (SELECT * FROM src WHERE id > 5 ORDER BY id ASC) +TO 'test_files/scratch/sort_pushdown_exact_reverse/multi/b.parquet'; +---- +5 + +statement ok +CREATE EXTERNAL TABLE multi_file(id INT, category VARCHAR) +STORED AS PARQUET +LOCATION 'test_files/scratch/sort_pushdown_exact_reverse/multi/' +WITH ORDER (id ASC); + +query TT +EXPLAIN SELECT * FROM multi_file ORDER BY id DESC LIMIT 7; +---- +logical_plan +01)Sort: multi_file.id DESC NULLS FIRST, fetch=7 +02)--TableScan: multi_file projection=[id, category] +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/multi/b.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/multi/a.parquet]]}, projection=[id, category], limit=7, output_ordering=[id@0 DESC], file_type=parquet, reverse_row_groups=true, reverse_rows=true + +query IT +SELECT * FROM multi_file ORDER BY id DESC LIMIT 7; +---- +10 b +9 a +8 b +7 a +6 b +5 a +4 b + +# Test 5: overlapping files cannot be read back to front as one ordered +# stream, so the pushdown falls back to `Inexact` and the sort stays. +query I +COPY (SELECT * FROM src WHERE id % 2 = 1 ORDER BY id ASC) +TO 'test_files/scratch/sort_pushdown_exact_reverse/overlap/odd.parquet'; +---- +5 + +query I +COPY (SELECT * FROM src WHERE id % 2 = 0 ORDER BY id ASC) +TO 'test_files/scratch/sort_pushdown_exact_reverse/overlap/even.parquet'; +---- +5 + +statement ok +CREATE EXTERNAL TABLE overlapping(id INT, category VARCHAR) +STORED AS PARQUET +LOCATION 'test_files/scratch/sort_pushdown_exact_reverse/overlap/' +WITH ORDER (id ASC); + +query TT +EXPLAIN SELECT * FROM overlapping ORDER BY id DESC LIMIT 3; +---- +logical_plan +01)Sort: overlapping.id DESC NULLS FIRST, fetch=3 +02)--TableScan: overlapping projection=[id, category] +physical_plan +01)SortExec: TopK(fetch=3), expr=[id@0 DESC], preserve_partitioning=[false] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/overlap/odd.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/overlap/even.parquet]]}, projection=[id, category], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible + +query IT +SELECT * FROM overlapping ORDER BY id DESC LIMIT 3; +---- +10 b +9 a +8 b + +# Test 6: a multi-column ordering reversed serves a prefix of its reverse +# and the full reverse. +statement ok +CREATE TABLE pairs(a INT, b INT) AS VALUES +(1, 3), (1, 2), (1, 1), (2, 3), (2, 2), (2, 1), (3, 2), (3, 1); + +query I +COPY (SELECT * FROM pairs ORDER BY a ASC, b DESC) +TO 'test_files/scratch/sort_pushdown_exact_reverse/pairs/data.parquet'; +---- +8 + +statement ok +CREATE EXTERNAL TABLE pairs_file(a INT, b INT) +STORED AS PARQUET +LOCATION 'test_files/scratch/sort_pushdown_exact_reverse/pairs/' +WITH ORDER (a ASC, b DESC); + +query TT +EXPLAIN SELECT * FROM pairs_file ORDER BY a DESC, b ASC; +---- +logical_plan +01)Sort: pairs_file.a DESC NULLS FIRST, pairs_file.b ASC NULLS LAST +02)--TableScan: pairs_file projection=[a, b] +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/pairs/data.parquet]]}, projection=[a, b], output_ordering=[a@0 DESC, b@1 ASC NULLS LAST], file_type=parquet, reverse_row_groups=true, reverse_rows=true + +query II +SELECT * FROM pairs_file ORDER BY a DESC, b ASC; +---- +3 1 +3 2 +2 1 +2 2 +2 3 +1 1 +1 2 +1 3 + +query II +SELECT * FROM pairs_file ORDER BY a DESC LIMIT 4; +---- +3 1 +3 2 +2 1 +2 2 + +# Test 7: NULLs written last under ASC NULLS LAST come out first when read +# back to front, which is what DESC NULLS FIRST asks for. +statement ok +CREATE TABLE with_nulls(id INT) AS VALUES (1), (2), (3), (NULL), (NULL); + +query I +COPY (SELECT * FROM with_nulls ORDER BY id ASC NULLS LAST) +TO 'test_files/scratch/sort_pushdown_exact_reverse/nulls/data.parquet'; +---- +5 + +statement ok +CREATE EXTERNAL TABLE nulls_file(id INT) +STORED AS PARQUET +LOCATION 'test_files/scratch/sort_pushdown_exact_reverse/nulls/' +WITH ORDER (id ASC NULLS LAST); + +query TT +EXPLAIN SELECT * FROM nulls_file ORDER BY id DESC NULLS FIRST LIMIT 3; +---- +logical_plan +01)Sort: nulls_file.id DESC NULLS FIRST, fetch=3 +02)--TableScan: nulls_file projection=[id] +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/nulls/data.parquet]]}, projection=[id], limit=3, output_ordering=[id@0 DESC], file_type=parquet, reverse_row_groups=true, reverse_rows=true + +query I +SELECT * FROM nulls_file ORDER BY id DESC NULLS FIRST LIMIT 3; +---- +NULL +NULL +3 + +# Test 8: on a NOT NULL column, `DESC NULLS LAST` is equivalent to the +# reverse of `ASC NULLS LAST`, but not literally its reverse. The file scan +# compares orderings literally, so it sorts the files by statistics instead +# of taking its reverse path, the declared ordering no longer describes the +# scan, and the exact claim is dropped: the sort stays and results are +# still correct. +statement ok +CREATE EXTERNAL TABLE multi_file_not_null(id INT NOT NULL, category VARCHAR) +STORED AS PARQUET +LOCATION 'test_files/scratch/sort_pushdown_exact_reverse/multi/' +WITH ORDER (id ASC NULLS LAST); + +query TT +EXPLAIN SELECT * FROM multi_file_not_null ORDER BY id DESC NULLS LAST LIMIT 7; +---- +logical_plan +01)Sort: multi_file_not_null.id DESC NULLS LAST, fetch=7 +02)--TableScan: multi_file_not_null projection=[id, category] +physical_plan +01)SortExec: TopK(fetch=7), expr=[id@0 DESC NULLS LAST], preserve_partitioning=[false] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/multi/b.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/multi/a.parquet]]}, projection=[id, category], file_type=parquet, predicate=DynamicFilter [ empty ], reverse_row_groups=true, reverse_rows=true, dynamic_rg_pruning=eligible + +query IT +SELECT * FROM multi_file_not_null ORDER BY id DESC NULLS LAST LIMIT 7; +---- +10 b +9 a +8 b +7 a +6 b +5 a +4 b + +# Test 9: two partitions, one file each. Each partition reads its file back +# to front and the merge above combines them. +statement ok +SET datafusion.execution.target_partitions = 2; + +statement ok +CREATE EXTERNAL TABLE multi_file_two_partitions(id INT, category VARCHAR) +STORED AS PARQUET +LOCATION 'test_files/scratch/sort_pushdown_exact_reverse/multi/' +WITH ORDER (id ASC); + +query TT +EXPLAIN SELECT * FROM multi_file_two_partitions ORDER BY id DESC LIMIT 7; +---- +logical_plan +01)Sort: multi_file_two_partitions.id DESC NULLS FIRST, fetch=7 +02)--TableScan: multi_file_two_partitions projection=[id, category] +physical_plan +01)SortPreservingMergeExec: [id@0 DESC], fetch=7 +02)--BufferExec: capacity=1073741824 +03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/multi/a.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/multi/b.parquet]]}, projection=[id, category], limit=7, output_ordering=[id@0 DESC], file_type=parquet, reverse_row_groups=true, reverse_rows=true + +query IT +SELECT * FROM multi_file_two_partitions ORDER BY id DESC LIMIT 7; +---- +10 b +9 a +8 b +7 a +6 b +5 a +4 b + +statement ok +SET datafusion.execution.target_partitions = 1; + +# Test 10: with the option off, the reverse scan stays `Inexact`. The parquet +# read options are captured when the table is created, so the table is +# created again after the option changes. +statement ok +SET datafusion.execution.parquet.enable_exact_reverse_scan = false; + +statement ok +CREATE EXTERNAL TABLE single_file_inexact(id INT, category VARCHAR) +STORED AS PARQUET +LOCATION 'test_files/scratch/sort_pushdown_exact_reverse/single/' +WITH ORDER (id ASC); + +query TT +EXPLAIN SELECT * FROM single_file_inexact ORDER BY id DESC LIMIT 3; +---- +logical_plan +01)Sort: single_file_inexact.id DESC NULLS FIRST, fetch=3 +02)--TableScan: single_file_inexact projection=[id, category] +physical_plan +01)SortExec: TopK(fetch=3), expr=[id@0 DESC], preserve_partitioning=[false] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/single/data.parquet]]}, projection=[id, category], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible + +query IT +SELECT * FROM single_file_inexact ORDER BY id DESC LIMIT 3; +---- +10 b +9 a +8 b + +# Cleanup +statement ok +RESET datafusion.execution.parquet.enable_exact_reverse_scan; + +statement ok +RESET datafusion.execution.parquet.max_row_group_size; + +statement ok +RESET datafusion.execution.parquet.pushdown_filters; + +statement ok +RESET datafusion.optimizer.enable_sort_pushdown; + +# The test harness runs files with 4 target partitions. +statement ok +SET datafusion.execution.target_partitions = 4; diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 445c63a05c93f..a167504883445 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -75,6 +75,7 @@ The following configuration settings are available: | datafusion.execution.parquet.pushdown_filters | false | (reading) If true, filter expressions are be applied during the parquet decoding operation to reduce the number of rows decoded. This optimization is sometimes called "late materialization". | | datafusion.execution.parquet.reorder_filters | false | (reading) If true, filter expressions evaluated during the parquet decoding operation will be reordered heuristically to minimize the cost of evaluation. If false, the filters are applied in the same order as written in the query | | datafusion.execution.parquet.force_filter_selections | false | (reading) Force the use of RowSelections for filter results, when pushdown_filters is enabled. If false, the reader will automatically choose between a RowSelection and a Bitmap based on the number and pattern of selected rows. | +| datafusion.execution.parquet.enable_exact_reverse_scan | false | (reading) If true, a scan whose requested ordering is the reverse of the files' declared ordering reads each row group into memory and emits its rows in reverse, so the scan is exactly ordered and the `SortExec` above it can be removed. Peak memory per scanned file is one decoded row group (after projection and filtering). If false, only the row group order is reversed and the sort is kept. | | datafusion.execution.parquet.schema_force_view_types | true | (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`, and `Binary/BinaryLarge` with `BinaryView`. The parquet reader is optimized for reading `Utf8View` and `BinaryView`, so such queries are significantly faster than reading `Utf8`/`Binary` and then casting to the view types. | | datafusion.execution.parquet.binary_as_string | false | (reading) If true, parquet reader will read columns of `Binary/LargeBinary` with `Utf8`, and `BinaryView` with `Utf8View`. Parquet files generated by some legacy writers do not correctly set the UTF8 flag for strings, causing string columns to be loaded as BLOB instead. The parquet reader has special optimizations for `Utf8` validation, so reading such columns as strings is significantly faster than reading them as binary and then casting to string. | | datafusion.execution.parquet.coerce_int96 | NULL | (reading) If set, parquet reader will read columns of physical type int96 as originating from this resolution instead of nanosecond. Valid values are: ns, us, ms, and s. This is useful for reading data from systems like Spark which store microsecond resolution timestamps in an int96, allowing them to write values with a larger date range than 64-bit timestamps with nanosecond resolution. | From cbd041da1d758bd7bc54c9bd6e6f4e74aa4df15e Mon Sep 17 00:00:00 2001 From: Qi Zhu Date: Thu, 5 Nov 2026 13:19:43 +0800 Subject: [PATCH 2/5] Clarify which row group a reversed scan reads first when loading the offset index --- datafusion/datasource-parquet/src/opener/mod.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index dfe3bb45d0ce8..b4ae43dc3f870 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -1440,7 +1440,10 @@ impl RowGroupsPrunedParquetOpen { // row group it reaches when that row group's output is known up // front (see `PreparedAccessPlan::select_tail_for_limit`). The // offset index lets the decoder skip the pages ahead of that tail - // instead of fetching the whole column chunk. + // instead of fetching the whole column chunk. `row_groups` is still + // in file order here (the reversal happens when the stream is + // built), so the row group a reversed scan reaches first is the + // last one. let prepared = &self.prepared.loaded.prepared; if prepared.reverse_rows && let Some(limit) = prepared.limit From 7e20cfb844e5d2d70ccd24c6ea955369a4d1af0b Mon Sep 17 00:00:00 2001 From: Qi Zhu Date: Thu, 5 Nov 2026 13:25:19 +0800 Subject: [PATCH 3/5] Test a limit that spans a row group boundary in an exact reverse scan --- .../datasource-parquet/src/opener/mod.rs | 27 +++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index b4ae43dc3f870..dabbb38b994e6 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -4602,6 +4602,33 @@ mod test { assert_eq!(collect_int32_values(stream).await, [12, 11, 8, 7, 4, 3]); } + /// A limit larger than the first row group read spans the row group + /// boundary: row group `9..=12` is taken whole, the next one (`5..=8`) + /// is cut to the 2 rows still needed (`7, 8`), and the walk stops, so + /// row group `1..=4` is read as is and only its last `limit` rows are + /// kept. The file stream stops after the first 6 rows, which are exactly + /// `12..=7`. + #[tokio::test] + async fn test_exact_reverse_scan_limit_spans_row_groups() { + let store = Arc::new(InMemory::new()) as Arc; + let (schema, file) = write_twelve_rows_in_three_row_groups(&store).await; + + let opener = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(schema) + .with_projection_indices(&[0]) + .with_limit(6) + .with_batch_size(3) + .with_reverse_row_groups(true) + .with_reverse_rows(true) + .build(); + let stream = open_file(&opener, file).await.unwrap(); + let values = collect_int32_values(stream).await; + + assert_eq!(values[..6], [12, 11, 10, 9, 8, 7]); + assert_eq!(values, [12, 11, 10, 9, 8, 7, 4, 3, 2, 1]); + } + /// Under a pushed-down filter a row group's output is known only when /// statistics prove every row passes. Row group `9..=12` is fully /// matched by `a > 5` and is cut to its last 3 rows up front; `5..=8` is From ec8e932c66d34ac3c001cd9ad2f5e58f86764f82 Mon Sep 17 00:00:00 2001 From: Qi Zhu Date: Sat, 7 Nov 2026 00:01:18 +0800 Subject: [PATCH 4/5] Keep limited scans and equivalent reverse requests correct in exact reverse scans - Decline filter pushdown into a scan that holds a limit, since the filter would run before the limit. - Do not push a sort into a scan that holds a limit, since reordering it changes which rows the limit keeps. - Reverse files and orderings when the reversed ordering satisfies the request under the scan's equivalences, not only when it is the literal reverse, so these requests become exact instead of leaving a row reversing scan under a TopK. - When an exact claim cannot be kept, continue with the original source so no row reversal is left under the sort. - Toggle the reverse flags when a reverse is pushed onto a reversed scan. - Group the row reversal state into RowGroupReverser and yield between the batches of a buffered row group so the query stays cancellable. - Document that the option is captured when a table is created. --- datafusion/common/src/config.rs | 4 +- .../datasource-parquet/src/opener/mod.rs | 16 +- .../datasource-parquet/src/push_decoder.rs | 187 ++++++++++-------- datafusion/datasource-parquet/src/source.rs | 44 ++++- .../datasource/src/file_scan_config/mod.rs | 148 ++++++++------ .../src/file_scan_config/sort_pushdown.rs | 112 +++++++++-- .../test_files/information_schema.slt | 2 +- .../sort_pushdown_exact_reverse.slt | 152 +++++++++++++- docs/source/user-guide/configs.md | 2 +- 9 files changed, 489 insertions(+), 178 deletions(-) diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index fd19f7e70e776..aa8815092ec1d 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1359,7 +1359,9 @@ config_namespace! { /// emits its rows in reverse, so the scan is exactly ordered and the /// `SortExec` above it can be removed. Peak memory per scanned file is /// one decoded row group (after projection and filtering). If false, - /// only the row group order is reversed and the sort is kept. + /// only the row group order is reversed and the sort is kept. The value + /// is captured when the table is created, so a later `SET` does not + /// affect existing tables. pub enable_exact_reverse_scan: bool, default = false /// (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`, diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index dabbb38b994e6..cee1618379fb3 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -17,7 +17,7 @@ use crate::metrics::{ByteProgress, RowFilterSkippedFullyMatchedMetric}; use crate::page_filter::PagePruningAccessPlanFilter; use crate::push_decoder::{ DecoderBuilderConfig, InitialDecoderState, PushDecoderStreamState, RgPlanEntry, - RowFilterContext, RowGroupPruner, + RowFilterContext, RowGroupPruner, RowGroupReverser, }; use crate::row_group_filter::{RowGroupAccessPlanFilter, row_group_in_range}; use crate::{ @@ -1962,20 +1962,16 @@ impl RowGroupsPrunedParquetOpen { filter_installed, row_filter_skipped_fully_matched, byte_progress, - reverse_rows: prepared.reverse_rows, - reverse_limit: prepared.limit.filter(|_| prepared.reverse_rows), - reversed_batches: Vec::new(), - reversed_reservation: prepared - .memory_pool - .as_ref() - .filter(|_| prepared.reverse_rows) - .map(|pool| { + reverser: prepared.reverse_rows.then(|| { + let reservation = prepared.memory_pool.as_ref().map(|pool| { MemoryConsumer::new(format!( "ParquetReverseRowGroup[{}]", prepared.partition_index )) .register(pool) - }), + }); + RowGroupReverser::new(prepared.limit, reservation) + }), } .into_stream(); diff --git a/datafusion/datasource-parquet/src/push_decoder.rs b/datafusion/datasource-parquet/src/push_decoder.rs index 22b34b953fa5e..8469036f88558 100644 --- a/datafusion/datasource-parquet/src/push_decoder.rs +++ b/datafusion/datasource-parquet/src/push_decoder.rs @@ -309,22 +309,10 @@ pub(crate) struct PushDecoderStreamState { /// group at a time as they are decoded or skipped, and topped up to the /// full range when the stream is dropped. pub(crate) byte_progress: ByteProgress, - /// Emit each row group's rows in reverse order. The active reader is - /// drained into [`Self::reversed_batches`] before anything is returned, - /// so one decoded row group is held in memory at a time. - pub(crate) reverse_rows: bool, - /// The scan's row limit, if any, when `reverse_rows` is set. No more than - /// this many rows of a row group can reach the output, so only its last - /// `limit` rows are kept. - pub(crate) reverse_limit: Option, - /// The current row group's batches, each already row-reversed and kept - /// in decode order, so popping from the end yields the row group back to - /// front. - pub(crate) reversed_batches: Vec, - /// Charges the memory pool for the batches held while a row group is - /// reversed. `None` when the scan does not reverse rows or was opened - /// without a pool. - pub(crate) reversed_reservation: Option, + /// Set when the scan emits each row group's rows in reverse order: the + /// row group is decoded in full before any of it is returned, so one + /// decoded row group is held in memory at a time. + pub(crate) reverser: Option, } /// A reusable, `Arc`-shared list of prebuilt row-filter candidates. @@ -450,32 +438,30 @@ impl PushDecoderStreamState { let elapsed_compute = self.baseline_metrics.elapsed_compute().clone(); let mut timer = elapsed_compute.timer(); loop { - // Step 1a: a reversed row group still being emitted. - if let Some(batch) = self.reversed_batches.pop() { - if let Some(reservation) = &self.reversed_reservation { - reservation.shrink(batch.get_array_memory_size()); - } + // Step 0: a reversed row group still being emitted. + if let Some(batch) = self.reverser.as_mut().and_then(RowGroupReverser::pop) { return Some((Ok(batch), self)); } - // Step 1b: in reverse mode, drain the whole row group, reverse it - // into `reversed_batches`, and emit it from step 1a. - if self.reverse_rows - && let Some(reader) = self.active_reader.take() - { - if let Err(e) = self.buffer_reversed_row_group(reader) { - return Some((Err(e), self)); - } - continue; - } - // Step 1: drain a batch from the active reader if any. if let Some(reader) = self.active_reader.as_mut() { match reader.next() { Some(Ok(batch)) => { self.copy_arrow_reader_metrics(); let result = self.project_batch(&batch); - return Some((result, self)); + let Some(reverser) = self.reverser.as_mut() else { + return Some((result, self)); + }; + // Reverse mode holds the row group until it is fully + // decoded. Yield between its batches so that decoding + // a large row group does not block cancellation. + if let Err(e) = result.and_then(|batch| reverser.push(batch)) { + return Some((Err(e), self)); + } + timer.stop(); + tokio::task::yield_now().await; + timer.restart(); + continue; } Some(Err(e)) => { return Some((Err(DataFusionError::from(e)), self)); @@ -484,6 +470,14 @@ impl PushDecoderStreamState { // Reader exhausted: drop and fall through to per-RG // boundary handling, then try_next_reader. self.active_reader = None; + // In reverse mode, emit the row group from step 0 + // first. + if let Some(reverser) = self.reverser.as_mut() { + if let Err(e) = reverser.finish() { + return Some((Err(e), self)); + } + continue; + } } } } @@ -743,63 +737,98 @@ impl PushDecoderStreamState { fn project_batch(&self, batch: &RecordBatch) -> Result { self.decoder_projection.map(batch) } +} - /// Decodes every remaining batch of `reader`'s row group, projects it, - /// reverses its rows, and stores the results in - /// [`Self::reversed_batches`]. Batches stay in decode order there, so - /// popping them yields the row group's rows last to first. - /// - /// Under [`Self::reverse_limit`] only the row group's last `limit` rows - /// are kept: batches that fall entirely before them are dropped as soon - /// as later batches cover the limit, so at most `limit` rows plus one - /// batch are held while decoding. - fn buffer_reversed_row_group( - &mut self, - reader: ParquetRecordBatchReader, - ) -> Result<()> { - let mut kept: VecDeque = VecDeque::new(); - let mut kept_rows = 0; - for batch in reader { - let batch = batch.map_err(DataFusionError::from)?; - self.copy_arrow_reader_metrics(); - let batch = self.project_batch(&batch)?; - if batch.num_rows() == 0 { - continue; - } - if let Some(reservation) = &self.reversed_reservation { - reservation.try_grow(batch.get_array_memory_size())?; - } - kept_rows += batch.num_rows(); - kept.push_back(batch); - if let Some(limit) = self.reverse_limit { - while let Some(front) = kept.front() - && kept_rows - front.num_rows() >= limit +/// Buffers one row group's decoded batches and returns them with the row +/// group's rows in reverse order. +pub(crate) struct RowGroupReverser { + /// The scan's row limit, if any. No more than this many rows of a row + /// group can reach the output, so only its last `limit` rows are kept. + limit: Option, + /// The row group's batches as decoded so far. + pending: VecDeque, + /// Rows in [`Self::pending`]. + pending_rows: usize, + /// The finished row group's batches, each already row-reversed and kept + /// in decode order, so popping from the end yields the row group back to + /// front. + reversed: Vec, + /// Charges the memory pool for the batches held. `None` when the scan was + /// opened without a pool. + reservation: Option, +} + +impl RowGroupReverser { + pub(crate) fn new( + limit: Option, + reservation: Option, + ) -> Self { + Self { + limit, + pending: VecDeque::new(), + pending_rows: 0, + reversed: Vec::new(), + reservation, + } + } + + /// Returns the next batch of the finished row group, last rows first. + fn pop(&mut self) -> Option { + let batch = self.reversed.pop()?; + if let Some(reservation) = &self.reservation { + reservation.shrink(batch.get_array_memory_size()); + } + Some(batch) + } + + /// Adds the next decoded batch of the current row group. Under + /// [`Self::limit`], batches that fall entirely before the row group's last + /// `limit` rows are dropped as soon as later batches cover the limit, so + /// at most `limit` rows plus one batch are held. + fn push(&mut self, batch: RecordBatch) -> Result<()> { + if batch.num_rows() == 0 { + return Ok(()); + } + if let Some(reservation) = &self.reservation { + reservation.try_grow(batch.get_array_memory_size())?; + } + self.pending_rows += batch.num_rows(); + self.pending.push_back(batch); + if let Some(limit) = self.limit { + while let Some(front) = self.pending.front() + && self.pending_rows - front.num_rows() >= limit + { + self.pending_rows -= front.num_rows(); + if let Some(dropped) = self.pending.pop_front() + && let Some(reservation) = &self.reservation { - kept_rows -= front.num_rows(); - if let Some(dropped) = kept.pop_front() - && let Some(reservation) = &self.reversed_reservation - { - reservation.shrink(dropped.get_array_memory_size()); - } + reservation.shrink(dropped.get_array_memory_size()); } } } - if let Some(limit) = self.reverse_limit - && kept_rows > limit - && let Some(front) = kept.front_mut() + Ok(()) + } + + /// Ends the current row group: trims it to its last `limit` rows and + /// reverses it, ready for [`Self::pop`]. + fn finish(&mut self) -> Result<()> { + if let Some(limit) = self.limit + && self.pending_rows > limit + && let Some(front) = self.pending.front_mut() { - let excess = kept_rows - limit; + let excess = self.pending_rows - limit; *front = front.slice(excess, front.num_rows() - excess); } + self.pending_rows = 0; // The reversed copies replace the decoded batches, so the - // reservation is resized to exactly what is stored, which is what - // step 1a releases batch by batch. - for batch in kept { - self.reversed_batches.push(reverse_batch_rows(&batch)?); + // reservation is resized to exactly what is stored, which `pop` + // releases batch by batch. + for batch in self.pending.drain(..) { + self.reversed.push(reverse_batch_rows(&batch)?); } - if let Some(reservation) = &self.reversed_reservation { + if let Some(reservation) = &self.reservation { let stored = self - .reversed_batches + .reversed .iter() .map(RecordBatch::get_array_memory_size) .sum(); diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index d89b16f358a87..c818cb3512f66 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -1078,10 +1078,12 @@ impl FileSource for ParquetSource { if reversed_satisfies && self.table_parquet_options.global.enable_exact_reverse_scan { + // Toggled, not set: pushing a reverse onto a scan that already + // reads back to front turns it forward again. let mut new_source = self.clone(); new_source.sort_order_for_reorder = None; - new_source.reverse_row_groups = true; - new_source.reverse_rows = true; + new_source.reverse_row_groups = !self.reverse_rows; + new_source.reverse_rows = !self.reverse_rows; return Ok(SortOrderPushdownResult::Exact { inner: Arc::new(new_source) as Arc, }); @@ -1803,6 +1805,44 @@ mod tests { assert!(inner_parquet.sort_order_for_reorder.is_none()); } + /// Pushing a reverse onto a scan that already reads back to front turns + /// it forward again, rather than leaving it reversed under an ordering + /// that has been flipped back. + #[test] + fn try_pushdown_sort_exact_reverse_twice_reads_forward() { + let (schema, eq, order) = exact_reverse_case(); + let mut options = TableParquetOptions::default(); + options.global.enable_exact_reverse_scan = true; + let source = ParquetSource::new(schema).with_table_parquet_options(options); + let SortOrderPushdownResult::Exact { inner: reversed } = + source.try_pushdown_sort(&order, &eq).unwrap() + else { + panic!("expected Exact for the first pushdown"); + }; + let reversed = reversed + .downcast_ref::() + .expect("inner is ParquetSource"); + assert!(reversed.reverse_rows()); + + // The reversed scan declares the reversed ordering; ask for the + // original one. + let mut reversed_eq = eq.clone(); + reversed_eq.clear_orderings(); + reversed_eq.add_ordering(order.clone()); + let forward_order: Vec<_> = order.iter().map(|expr| expr.reverse()).collect(); + let SortOrderPushdownResult::Exact { inner } = reversed + .try_pushdown_sort(&forward_order, &reversed_eq) + .unwrap() + else { + panic!("expected Exact for the second pushdown"); + }; + let forward = inner + .downcast_ref::() + .expect("inner is ParquetSource"); + assert!(!forward.reverse_row_groups()); + assert!(!forward.reverse_rows()); + } + /// Without the option, the same request stays `Inexact` and rows keep /// their on-disk order within each row group. #[test] diff --git a/datafusion/datasource/src/file_scan_config/mod.rs b/datafusion/datasource/src/file_scan_config/mod.rs index 7805a440d4f7f..48c9a6c38907f 100644 --- a/datafusion/datasource/src/file_scan_config/mod.rs +++ b/datafusion/datasource/src/file_scan_config/mod.rs @@ -51,7 +51,7 @@ use datafusion_physical_plan::execution_plan::SchedulingType; use datafusion_physical_plan::{ DisplayAs, DisplayFormatType, display::{ProjectSchemaDisplay, display_orderings}, - filter_pushdown::FilterPushdownPropagation, + filter_pushdown::{FilterPushdownPropagation, PushedDown}, metrics::ExecutionPlanMetricsSet, }; use log::{debug, warn}; @@ -1077,6 +1077,15 @@ impl DataSource for FileScanConfig { filters: Vec>, config: &ConfigOptions, ) -> Result>> { + // The scan applies its limit after any filter it holds, so a filter + // from above the limit would let through rows the limit should have + // cut. + if self.limit.is_some() { + return Ok(FilterPushdownPropagation::with_parent_pushdown_result( + vec![PushedDown::No; filters.len()], + )); + } + // Remap filter Column indices to match the table schema (file + partition columns). // This is necessary because filters refer to the output schema of this `DataSource` // (e.g., after projection pushdown has been applied) and need to be remapped to the table schema @@ -1144,6 +1153,8 @@ impl DataSource for FileScanConfig { /// │ (natural ordering satisfies request) /// │ → rebuild_with_source: sort files by stats, verify non-overlapping /// │ → SortExec removed, fetch (LIMIT) pushed to DataSourceExec + /// │ → if the ordering cannot be kept, continue as for Inexact + /// │ with the original source /// │ /// ├─► FileSource returns Inexact /// │ (e.g. column_in_file_schema: opener will reorder RGs at runtime) @@ -1166,6 +1177,12 @@ impl DataSource for FileScanConfig { &self, order: &[PhysicalSortExpr], ) -> Result>> { + // Reordering the files or rows of a scan changes which rows its limit + // keeps. + if self.limit.is_some() { + return Ok(SortOrderPushdownResult::Unsupported); + } + let pushdown_result = self .file_source .try_pushdown_sort(order, &self.eq_properties())?; @@ -1176,54 +1193,19 @@ impl DataSource for FileScanConfig { // rebuild_with_source keeps output_ordering only when all groups // are non-overlapping. If output_ordering was cleared, files // overlap despite within-file ordering → downgrade to Inexact. - if config.output_ordering.is_empty() { - Ok(SortOrderPushdownResult::Inexact { + if !config.output_ordering.is_empty() { + return Ok(SortOrderPushdownResult::Exact { inner: Arc::new(config), - }) - } else { - Ok(SortOrderPushdownResult::Exact { - inner: Arc::new(config), - }) + }); } + // An exact source may read rows back to front, which only pays + // off when the scan replaces the sort. With the sort kept, fall + // back to the source as it was, so no buffering is left under + // it. + self.inexact_pushdown(Arc::clone(&self.file_source), order) } SortOrderPushdownResult::Inexact { inner } => { - let mut config = self.rebuild_with_source(inner, false, order)?; - // `rebuild_with_source` reorders files by stats; if the - // post-sort files are non-overlapping AND the request now - // validates against the new file groups, `output_ordering` - // is preserved and we can upgrade back to Exact. This - // restores the sort-elimination behaviour that lived in - // the `Unsupported` → `try_sort_file_groups_by_statistics` - // path before #21956 routed `column_in_file_schema` cases - // here. - if config.output_ordering.is_empty() { - return Ok(SortOrderPushdownResult::Inexact { - inner: Arc::new(config), - }); - } - // Upgrading to Exact: the post-sort file groups are - // non-overlapping and each file's declared ordering - // re-validates, so reading the files in their natural - // (declared-sorted) order already yields the requested - // ordering — exactly like the `Unsupported` → Exact path, - // which reads files in natural order too. - // - // Drop the runtime row-group reorder hints the Inexact - // source carried (`sort_order_for_reorder` / - // `reverse_row_groups`) by restoring the original, - // hint-free source. With the `SortExec` removed those - // hints are not just redundant but unsafe: for a DESC - // request the opener sorts row groups ASC-by-min and then - // reverses them, which reorders two row groups within a - // single file that share the same `min` incorrectly - // (e.g. a file `[10,8,8,8]` whose row groups are - // `[10,8]` and `[8,8]` would stream as `8,8,10,8`). - // The `SortExec` used to mask this; once it is gone the - // reordered stream is the final, wrong answer. - config.file_source = Arc::clone(&self.file_source); - Ok(SortOrderPushdownResult::Exact { - inner: Arc::new(config), - }) + self.inexact_pushdown(inner, order) } SortOrderPushdownResult::Unsupported => { self.try_sort_file_groups_by_statistics(order) @@ -3551,6 +3533,8 @@ mod tests { struct ExactSortPushdownSource { metrics: ExecutionPlanMetricsSet, table_schema: TableSchema, + /// Whether this is the source returned by `try_pushdown_sort`. + pushed: bool, } impl ExactSortPushdownSource { @@ -3558,6 +3542,7 @@ mod tests { Self { metrics: ExecutionPlanMetricsSet::new(), table_schema, + pushed: false, } } } @@ -3585,7 +3570,11 @@ mod tests { } fn file_type(&self) -> &str { - "mock_exact" + if self.pushed { + "mock_exact_pushed" + } else { + "mock_exact" + } } fn try_pushdown_sort( @@ -3593,8 +3582,10 @@ mod tests { _order: &[PhysicalSortExpr], _eq_properties: &EquivalenceProperties, ) -> Result>> { + let mut pushed = self.clone(); + pushed.pushed = true; Ok(SortOrderPushdownResult::Exact { - inner: Arc::new(self.clone()) as Arc, + inner: Arc::new(pushed) as Arc, }) } @@ -3812,18 +3803,19 @@ mod tests { .downcast_ref::() .expect("Expected FileScanConfig"); assert!(pushed_config.output_ordering.is_empty()); + // The sort stays, so the scan keeps the source it had rather than the + // one that would read every file back to front under it. + assert_eq!(pushed_config.file_source.file_type(), "mock_exact"); Ok(()) } - /// A source can be `Exact` for a request that only its equivalence - /// reasoning recognises as the reverse of the declared ordering, here + /// A source can be `Exact` for a request that only equivalence reasoning + /// recognises as the reverse of the declared ordering, here /// `a DESC NULLS LAST` on a non-nullable `a` declared `a ASC NULLS LAST`. - /// `FileScanConfig` compares the orderings literally, so it sorts the - /// files by statistics instead of taking its reverse path, and the - /// forward ordering it would keep no longer describes the scan: the - /// claim is downgraded to `Inexact` and the sort is kept. + /// The scan reverses its files and orderings for it just as for a literal + /// reverse. #[test] - fn sort_pushdown_exact_without_file_reversal_downgraded_to_inexact() -> Result<()> { + fn sort_pushdown_exact_reverse_by_equivalence() -> Result<()> { let file_schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); let table_schema = TableSchema::from(&file_schema); @@ -3841,7 +3833,7 @@ mod tests { FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), file_source) .with_file_groups(file_groups) .with_output_ordering(vec![ - LexOrdering::new(vec![asc_nulls_last]).unwrap(), + LexOrdering::new(vec![asc_nulls_last.clone()]).unwrap(), ]) .build(); let desc_nulls_last = @@ -3849,13 +3841,53 @@ mod tests { let result = config.try_pushdown_sort(&[desc_nulls_last])?; - let SortOrderPushdownResult::Inexact { inner } = result else { - panic!("Expected Inexact result, got {result:?}"); + let SortOrderPushdownResult::Exact { inner } = result else { + panic!("Expected Exact result, got {result:?}"); }; let pushed_config = inner .downcast_ref::() .expect("Expected FileScanConfig"); - assert!(pushed_config.output_ordering.is_empty()); + let names: Vec<_> = pushed_config.file_groups[0] + .files() + .iter() + .map(|f| f.object_meta.location.as_ref()) + .collect(); + assert_eq!(names, ["file2", "file1"]); + assert_eq!( + pushed_config.output_ordering, + vec![LexOrdering::new(vec![asc_nulls_last.reverse()]).unwrap()] + ); + assert_eq!(pushed_config.file_source.file_type(), "mock_exact_pushed"); + Ok(()) + } + + /// A scan that holds a limit is not reordered: reading it in another order + /// would change which rows the limit keeps. + #[test] + fn sort_pushdown_with_limit_unsupported() -> Result<()> { + let file_schema = + Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); + let table_schema = TableSchema::from(&file_schema); + let file_source = Arc::new(ExactSortPushdownSource::new(table_schema)); + let sort_expr = PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0))); + let config = + FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), file_source) + .with_file_groups(vec![FileGroup::new(vec![ + make_file_with_stats("file1", 0.0, 9.0), + make_file_with_stats("file2", 10.0, 19.0), + ])]) + .with_output_ordering(vec![ + LexOrdering::new(vec![sort_expr.clone()]).unwrap(), + ]) + .with_limit(Some(5)) + .build(); + + let result = config.try_pushdown_sort(&[sort_expr.reverse()])?; + + assert!( + matches!(result, SortOrderPushdownResult::Unsupported), + "Expected Unsupported result, got {result:?}" + ); Ok(()) } diff --git a/datafusion/datasource/src/file_scan_config/sort_pushdown.rs b/datafusion/datasource/src/file_scan_config/sort_pushdown.rs index 6aa90e805e80a..c7b09849df754 100644 --- a/datafusion/datasource/src/file_scan_config/sort_pushdown.rs +++ b/datafusion/datasource/src/file_scan_config/sort_pushdown.rs @@ -70,7 +70,11 @@ impl FileScanConfig { let mut new_config = self.clone(); // Reverse file order (within each group) if the caller is requesting a reversal of this - // scan's declared output ordering. + // scan's declared output ordering. An exact source reads every file + // back to front, so for it the request only has to be satisfied by + // the reversed ordering under the scan's equivalences (constants from + // filters, monotonic functions, non-nullable columns), not be its + // literal reverse. let reverse_file_groups = if self.output_ordering.is_empty() { false } else if let Some(requested) = LexOrdering::new(order.iter().cloned()) { @@ -79,20 +83,13 @@ impl FileScanConfig { orderings .iter() .any(|ordering| ordering.is_reverse(&requested)) + || (is_exact && self.reverse_scan_satisfies(order)?) } else { false }; if reverse_file_groups { - new_config.file_groups = new_config - .file_groups - .into_iter() - .map(|group| { - let mut files = group.into_inner(); - files.reverse(); - files.into() - }) - .collect(); + new_config.file_groups = reversed_file_groups(new_config.file_groups); } new_config.file_source = new_file_source; @@ -104,13 +101,7 @@ impl FileScanConfig { // if the ordering does not hold, clear it and the caller falls // back to `Inexact`, which keeps the `SortExec`. if reverse_file_groups && is_exact { - new_config.output_ordering = self - .output_ordering - .iter() - .filter_map(|ordering| { - LexOrdering::new(ordering.iter().map(|expr| expr.reverse())) - }) - .collect(); + new_config.output_ordering = reversed_orderings(&self.output_ordering); if !new_config .eq_properties() .ordering_satisfy(order.iter().cloned())? @@ -230,6 +221,71 @@ impl FileScanConfig { Ok(new_config) } + /// Finishes a pushdown whose source result is `Inexact`, or an `Exact` + /// one the scan could not keep: reorders files by statistics and upgrades + /// to `Exact` when the reordered files make the declared ordering hold. + pub(crate) fn inexact_pushdown( + &self, + new_file_source: Arc, + order: &[PhysicalSortExpr], + ) -> Result>> { + let mut config = self.rebuild_with_source(new_file_source, false, order)?; + + // `rebuild_with_source` reorders files by stats; if the + // post-sort files are non-overlapping AND the request now + // validates against the new file groups, `output_ordering` + // is preserved and we can upgrade back to Exact. This + // restores the sort-elimination behaviour that lived in + // the `Unsupported` → `try_sort_file_groups_by_statistics` + // path before #21956 routed `column_in_file_schema` cases + // here. + if config.output_ordering.is_empty() { + return Ok(SortOrderPushdownResult::Inexact { + inner: Arc::new(config), + }); + } + // Upgrading to Exact: the post-sort file groups are + // non-overlapping and each file's declared ordering + // re-validates, so reading the files in their natural + // (declared-sorted) order already yields the requested + // ordering — exactly like the `Unsupported` → Exact path, + // which reads files in natural order too. + // + // Drop the runtime row-group reorder hints the Inexact + // source carried (`sort_order_for_reorder` / + // `reverse_row_groups`) by restoring the original, + // hint-free source. With the `SortExec` removed those + // hints are not just redundant but unsafe: for a DESC + // request the opener sorts row groups ASC-by-min and then + // reverses them, which reorders two row groups within a + // single file that share the same `min` incorrectly + // (e.g. a file `[10,8,8,8]` whose row groups are + // `[10,8]` and `[8,8]` would stream as `8,8,10,8`). + // The `SortExec` used to mask this; once it is gone the + // reordered stream is the final, wrong answer. + config.file_source = Arc::clone(&self.file_source); + Ok(SortOrderPushdownResult::Exact { + inner: Arc::new(config), + }) + } + + /// Whether only a back-to-front read of this scan satisfies `order`: the + /// declared orderings do not, but their reverse does. + fn reverse_scan_satisfies(&self, order: &[PhysicalSortExpr]) -> Result { + if self + .eq_properties() + .ordering_satisfy(order.iter().cloned())? + { + return Ok(false); + } + let mut reversed = self.clone(); + reversed.file_groups = reversed_file_groups(reversed.file_groups); + reversed.output_ordering = reversed_orderings(&self.output_ordering); + reversed + .eq_properties() + .ordering_satisfy(order.iter().cloned()) + } + /// Last-resort optimization when FileSource returns `Unsupported`. /// /// FileSource may return `Unsupported` because `eq_properties` had no @@ -316,6 +372,28 @@ impl FileScanConfig { } } +/// Returns `file_groups` with the files of each group in reverse order. +fn reversed_file_groups(file_groups: Vec) -> Vec { + file_groups + .into_iter() + .map(|group| { + let mut files = group.into_inner(); + files.reverse(); + files.into() + }) + .collect() +} + +/// Returns `orderings` with every sort expression reversed. +fn reversed_orderings(orderings: &[LexOrdering]) -> Vec { + orderings + .iter() + .filter_map(|ordering| { + LexOrdering::new(ordering.iter().map(|expr| expr.reverse())) + }) + .collect() +} + /// Sort files within each file group by their min/max statistics. /// /// No files are moved between groups — parallelism and group composition diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index 78466602ca2b9..899b2d16b2ee9 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -410,7 +410,7 @@ datafusion.execution.parquet.data_page_row_count_limit 20000 (writing) Sets best datafusion.execution.parquet.data_pagesize_limit 1048576 (writing) Sets best effort maximum size of data page in bytes datafusion.execution.parquet.dictionary_enabled true (writing) Sets if dictionary encoding is enabled. If NULL, uses default parquet writer setting datafusion.execution.parquet.dictionary_page_size_limit 1048576 (writing) Sets best effort maximum dictionary page size, in bytes -datafusion.execution.parquet.enable_exact_reverse_scan false (reading) If true, a scan whose requested ordering is the reverse of the files' declared ordering reads each row group into memory and emits its rows in reverse, so the scan is exactly ordered and the `SortExec` above it can be removed. Peak memory per scanned file is one decoded row group (after projection and filtering). If false, only the row group order is reversed and the sort is kept. +datafusion.execution.parquet.enable_exact_reverse_scan false (reading) If true, a scan whose requested ordering is the reverse of the files' declared ordering reads each row group into memory and emits its rows in reverse, so the scan is exactly ordered and the `SortExec` above it can be removed. Peak memory per scanned file is one decoded row group (after projection and filtering). If false, only the row group order is reversed and the sort is kept. The value is captured when the table is created, so a later `SET` does not affect existing tables. datafusion.execution.parquet.enable_page_index true (reading) If true, reads the Parquet data page level metadata (the Page Index), if present, to reduce the I/O and number of rows decoded. datafusion.execution.parquet.enable_rle_to_dictionary false (reading) If true, top-level string and binary Parquet columns with dictionary pages are inferred and scanned as `Dictionary` / `Dictionary` instead of their plain value type. This applies only when DataFusion infers the table schema. Tables with a user-supplied schema are not promoted because the Parquet footer is not read at DDL time, so dictionary pages cannot be detected per column. See datafusion.execution.parquet.encoding NULL (writing) Sets default encoding for any column. Valid values are: plain, plain_dictionary, rle, bit_packed, delta_binary_packed, delta_length_byte_array, delta_byte_array, rle_dictionary, byte_stream_split, and alp. These values are not case sensitive. If NULL, uses default parquet writer setting diff --git a/datafusion/sqllogictest/test_files/sort_pushdown_exact_reverse.slt b/datafusion/sqllogictest/test_files/sort_pushdown_exact_reverse.slt index dc0c52ff5b74b..203319a3821a0 100644 --- a/datafusion/sqllogictest/test_files/sort_pushdown_exact_reverse.slt +++ b/datafusion/sqllogictest/test_files/sort_pushdown_exact_reverse.slt @@ -250,12 +250,9 @@ NULL NULL 3 -# Test 8: on a NOT NULL column, `DESC NULLS LAST` is equivalent to the -# reverse of `ASC NULLS LAST`, but not literally its reverse. The file scan -# compares orderings literally, so it sorts the files by statistics instead -# of taking its reverse path, the declared ordering no longer describes the -# scan, and the exact claim is dropped: the sort stays and results are -# still correct. +# Test 8: on a NOT NULL column, `DESC NULLS LAST` is not literally the +# reverse of `ASC NULLS LAST`, but is equivalent to it, so the scan is still +# reversed exactly. statement ok CREATE EXTERNAL TABLE multi_file_not_null(id INT NOT NULL, category VARCHAR) STORED AS PARQUET @@ -268,9 +265,7 @@ EXPLAIN SELECT * FROM multi_file_not_null ORDER BY id DESC NULLS LAST LIMIT 7; logical_plan 01)Sort: multi_file_not_null.id DESC NULLS LAST, fetch=7 02)--TableScan: multi_file_not_null projection=[id, category] -physical_plan -01)SortExec: TopK(fetch=7), expr=[id@0 DESC NULLS LAST], preserve_partitioning=[false] -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/multi/b.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/multi/a.parquet]]}, projection=[id, category], file_type=parquet, predicate=DynamicFilter [ empty ], reverse_row_groups=true, reverse_rows=true, dynamic_rg_pruning=eligible +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/multi/b.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/multi/a.parquet]]}, projection=[id, category], limit=7, output_ordering=[id@0 DESC], file_type=parquet, reverse_row_groups=true, reverse_rows=true query IT SELECT * FROM multi_file_not_null ORDER BY id DESC NULLS LAST LIMIT 7; @@ -348,6 +343,145 @@ SELECT * FROM single_file_inexact ORDER BY id DESC LIMIT 3; 9 a 8 b +statement ok +SET datafusion.execution.parquet.enable_exact_reverse_scan = true; + +# Tests 11-13 use a scan that ends up holding a limit, with filters pushed +# into the parquet reader and four rows per row group. +statement ok +SET datafusion.execution.parquet.pushdown_filters = true; + +statement ok +SET datafusion.execution.parquet.max_row_group_size = 4; + +query I +COPY ( + SELECT value AS id, CASE WHEN value % 2 = 0 THEN 'a' ELSE 'b' END AS category + FROM generate_series(1, 40) ORDER BY id +) +TO 'test_files/scratch/sort_pushdown_exact_reverse/limited/data.parquet'; +---- +40 + +query I +COPY (SELECT * FROM (VALUES (1), (2), (39)) v(id)) +TO 'test_files/scratch/sort_pushdown_exact_reverse/probe/data.parquet'; +---- +3 + +statement ok +CREATE EXTERNAL TABLE limited(id BIGINT, category VARCHAR) +STORED AS PARQUET +LOCATION 'test_files/scratch/sort_pushdown_exact_reverse/limited/' +WITH ORDER (id ASC); + +statement ok +CREATE EXTERNAL TABLE probe(id BIGINT) +STORED AS PARQUET +LOCATION 'test_files/scratch/sort_pushdown_exact_reverse/probe/'; + +# Test 11: the reversed scan holds `LIMIT 5`, so a join's dynamic filter +# must not be pushed below it: the filter would run before the limit and +# let rows outside the top 5 through. +query I +SELECT p.id FROM probe p +JOIN (SELECT * FROM limited ORDER BY id DESC LIMIT 5) x ON p.id = x.id +ORDER BY 1; +---- +39 + +# Test 12: the same for the dynamic filter of a `TopK` above the limit. +query I +SELECT id FROM (SELECT * FROM limited ORDER BY id DESC LIMIT 5) +ORDER BY category DESC, id LIMIT 1; +---- +37 + +# Test 13: a scan that already holds a limit is not reversed for a sort +# above it, since reading it back to front would keep the other end of the +# file. +query I +SELECT id FROM (SELECT * FROM limited ORDER BY id ASC LIMIT 5) ORDER BY id DESC; +---- +5 +4 +3 +2 +1 + +statement ok +SET datafusion.execution.parquet.max_row_group_size = 2; + +# Test 14: a filter that fixes the leading column makes the reverse of +# `(a ASC, b ASC)` satisfy `b DESC`, so the scan is reversed exactly. The +# filter has to be pushed into the scan for the scan to see it. +query I +COPY ( + SELECT a, b FROM generate_series(0, 2) t1(a), generate_series(1, 20) t2(b) + ORDER BY a, b +) +TO 'test_files/scratch/sort_pushdown_exact_reverse/ab/data.parquet'; +---- +60 + +statement ok +CREATE EXTERNAL TABLE ab(a BIGINT, b BIGINT) +STORED AS PARQUET +LOCATION 'test_files/scratch/sort_pushdown_exact_reverse/ab/' +WITH ORDER (a ASC, b ASC); + +query TT +EXPLAIN SELECT * FROM ab WHERE a = 1 ORDER BY b DESC LIMIT 3; +---- +logical_plan +01)Sort: ab.b DESC NULLS FIRST, fetch=3 +02)--Filter: ab.a = Int64(1) +03)----TableScan: ab projection=[a, b], partial_filters=[ab.a = Int64(1)] +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/ab/data.parquet]]}, projection=[a, b], limit=3, output_ordering=[a@0 DESC, b@1 DESC], file_type=parquet, predicate=a@0 = 1, reverse_row_groups=true, reverse_rows=true, pruning_predicate=a_null_count@2 != row_count@3 AND a_min@0 <= 1 AND 1 <= a_max@1, required_guarantees=[a in (1)] + +query II +SELECT * FROM ab WHERE a = 1 ORDER BY b DESC LIMIT 3; +---- +1 20 +1 19 +1 18 + +statement ok +RESET datafusion.execution.parquet.pushdown_filters; + +# Test 15: a monotonic function of the declared column is served by the +# reversed scan too. +query I +COPY ( + SELECT to_timestamp_seconds(value * 21600) AS ts FROM generate_series(0, 11) + ORDER BY ts +) +TO 'test_files/scratch/sort_pushdown_exact_reverse/days/data.parquet'; +---- +12 + +statement ok +CREATE EXTERNAL TABLE days(ts TIMESTAMP) +STORED AS PARQUET +LOCATION 'test_files/scratch/sort_pushdown_exact_reverse/days/' +WITH ORDER (ts ASC); + +query TT +EXPLAIN SELECT ts FROM days ORDER BY date_trunc('day', ts) DESC LIMIT 4; +---- +logical_plan +01)Sort: date_trunc(Utf8("day"), days.ts) DESC NULLS FIRST, fetch=4 +02)--TableScan: days projection=[ts] +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown_exact_reverse/days/data.parquet]]}, projection=[ts], limit=4, output_ordering=[ts@0 DESC], file_type=parquet, reverse_row_groups=true, reverse_rows=true + +query P rowsort +SELECT ts FROM days ORDER BY date_trunc('day', ts) DESC LIMIT 4; +---- +1970-01-03T00:00:00 +1970-01-03T06:00:00 +1970-01-03T12:00:00 +1970-01-03T18:00:00 + # Cleanup statement ok RESET datafusion.execution.parquet.enable_exact_reverse_scan; diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index a167504883445..bbf7fd1b56954 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -75,7 +75,7 @@ The following configuration settings are available: | datafusion.execution.parquet.pushdown_filters | false | (reading) If true, filter expressions are be applied during the parquet decoding operation to reduce the number of rows decoded. This optimization is sometimes called "late materialization". | | datafusion.execution.parquet.reorder_filters | false | (reading) If true, filter expressions evaluated during the parquet decoding operation will be reordered heuristically to minimize the cost of evaluation. If false, the filters are applied in the same order as written in the query | | datafusion.execution.parquet.force_filter_selections | false | (reading) Force the use of RowSelections for filter results, when pushdown_filters is enabled. If false, the reader will automatically choose between a RowSelection and a Bitmap based on the number and pattern of selected rows. | -| datafusion.execution.parquet.enable_exact_reverse_scan | false | (reading) If true, a scan whose requested ordering is the reverse of the files' declared ordering reads each row group into memory and emits its rows in reverse, so the scan is exactly ordered and the `SortExec` above it can be removed. Peak memory per scanned file is one decoded row group (after projection and filtering). If false, only the row group order is reversed and the sort is kept. | +| datafusion.execution.parquet.enable_exact_reverse_scan | false | (reading) If true, a scan whose requested ordering is the reverse of the files' declared ordering reads each row group into memory and emits its rows in reverse, so the scan is exactly ordered and the `SortExec` above it can be removed. Peak memory per scanned file is one decoded row group (after projection and filtering). If false, only the row group order is reversed and the sort is kept. The value is captured when the table is created, so a later `SET` does not affect existing tables. | | datafusion.execution.parquet.schema_force_view_types | true | (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`, and `Binary/BinaryLarge` with `BinaryView`. The parquet reader is optimized for reading `Utf8View` and `BinaryView`, so such queries are significantly faster than reading `Utf8`/`Binary` and then casting to the view types. | | datafusion.execution.parquet.binary_as_string | false | (reading) If true, parquet reader will read columns of `Binary/LargeBinary` with `Utf8`, and `BinaryView` with `Utf8View`. Parquet files generated by some legacy writers do not correctly set the UTF8 flag for strings, causing string columns to be loaded as BLOB instead. The parquet reader has special optimizations for `Utf8` validation, so reading such columns as strings is significantly faster than reading them as binary and then casting to string. | | datafusion.execution.parquet.coerce_int96 | NULL | (reading) If set, parquet reader will read columns of physical type int96 as originating from this resolution instead of nanosecond. Valid values are: ns, us, ms, and s. This is useful for reading data from systems like Spark which store microsecond resolution timestamps in an int96, allowing them to write values with a larger date range than 64-bit timestamps with nanosecond resolution. | From 71beb85baa6fe18bc940820c1b91b7bd959be6dd Mon Sep 17 00:00:00 2001 From: Qi Zhu Date: Tue, 10 Nov 2026 10:07:49 +0800 Subject: [PATCH 5/5] Treat a fetch as a barrier in the PushdownSort and FilterPushdown rules Replace the scan-level `limit` checks with one check per rule, as suggested in review: a node with a fetch keeps the first rows in its input's current order, so `PushdownSort` must not reorder the input below it and `FilterPushdown` must not move a filter from above into or below it (the node's own filters still push down). This also fixes #27194 at the default configuration. The two `filter_pushdown.rs` snapshots that merged an outer filter into a `FilterExec` with a fetch now keep the filters separate, which keeps the rows the inner fetch picks. --- .../physical_optimizer/filter_pushdown.rs | 22 +++++---- .../datasource/src/file_scan_config/mod.rs | 47 +------------------ .../physical-optimizer/src/filter_pushdown.rs | 10 ++++ .../physical-optimizer/src/pushdown_sort.rs | 12 +++++ .../sort_pushdown_exact_reverse.slt | 46 ++++++++++++++++++ 5 files changed, 82 insertions(+), 55 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index 6b4f47227c427..987dca9fdf581 100644 --- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs @@ -440,8 +440,10 @@ fn test_filter_collapse_outer_fetch_preserved() { } #[test] -fn test_filter_collapse_inner_fetch_preserved() { - // When the inner filter has fetch and outer does not, the merged filter should preserve fetch +fn test_filter_not_collapsed_into_inner_fetch() { + // The inner filter keeps its first 5 matching rows before the outer filter + // applies, so merging the outer predicate into it would change which rows + // are kept. The filters stay separate. let scan = TestScanBuilder::new(schema()).with_support(false).build(); let predicate1 = col_lit_predicate("a", "foo", &schema()); let filter1 = Arc::new( @@ -463,16 +465,17 @@ fn test_filter_collapse_inner_fetch_preserved() { - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=false output: Ok: - - FilterExec: b@1 = bar AND a@0 = foo, fetch=5 - - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=false + - FilterExec: b@1 = bar + - FilterExec: a@0 = foo, fetch=5 + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=false " ); } #[test] -fn test_filter_collapse_both_fetch_uses_minimum() { - // When both filters have fetch, the merged filter should use the smaller (tighter) fetch. - // Inner fetch=5 is tighter than outer fetch=10, so the result should be fetch=5. +fn test_filter_with_fetch_not_collapsed_into_inner_fetch() { + // Same as above with a fetch on the outer filter too: the inner filter's + // fetch still picks its rows first, so the filters stay separate. let scan = TestScanBuilder::new(schema()).with_support(false).build(); let predicate1 = col_lit_predicate("a", "foo", &schema()); let filter1 = Arc::new( @@ -499,8 +502,9 @@ fn test_filter_collapse_both_fetch_uses_minimum() { - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=false output: Ok: - - FilterExec: b@1 = bar AND a@0 = foo, fetch=5 - - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=false + - FilterExec: b@1 = bar, fetch=10 + - FilterExec: a@0 = foo, fetch=5 + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=false " ); } diff --git a/datafusion/datasource/src/file_scan_config/mod.rs b/datafusion/datasource/src/file_scan_config/mod.rs index 48c9a6c38907f..f370c584fff90 100644 --- a/datafusion/datasource/src/file_scan_config/mod.rs +++ b/datafusion/datasource/src/file_scan_config/mod.rs @@ -51,7 +51,7 @@ use datafusion_physical_plan::execution_plan::SchedulingType; use datafusion_physical_plan::{ DisplayAs, DisplayFormatType, display::{ProjectSchemaDisplay, display_orderings}, - filter_pushdown::{FilterPushdownPropagation, PushedDown}, + filter_pushdown::FilterPushdownPropagation, metrics::ExecutionPlanMetricsSet, }; use log::{debug, warn}; @@ -1077,15 +1077,6 @@ impl DataSource for FileScanConfig { filters: Vec>, config: &ConfigOptions, ) -> Result>> { - // The scan applies its limit after any filter it holds, so a filter - // from above the limit would let through rows the limit should have - // cut. - if self.limit.is_some() { - return Ok(FilterPushdownPropagation::with_parent_pushdown_result( - vec![PushedDown::No; filters.len()], - )); - } - // Remap filter Column indices to match the table schema (file + partition columns). // This is necessary because filters refer to the output schema of this `DataSource` // (e.g., after projection pushdown has been applied) and need to be remapped to the table schema @@ -1177,12 +1168,6 @@ impl DataSource for FileScanConfig { &self, order: &[PhysicalSortExpr], ) -> Result>> { - // Reordering the files or rows of a scan changes which rows its limit - // keeps. - if self.limit.is_some() { - return Ok(SortOrderPushdownResult::Unsupported); - } - let pushdown_result = self .file_source .try_pushdown_sort(order, &self.eq_properties())?; @@ -3861,36 +3846,6 @@ mod tests { Ok(()) } - /// A scan that holds a limit is not reordered: reading it in another order - /// would change which rows the limit keeps. - #[test] - fn sort_pushdown_with_limit_unsupported() -> Result<()> { - let file_schema = - Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); - let table_schema = TableSchema::from(&file_schema); - let file_source = Arc::new(ExactSortPushdownSource::new(table_schema)); - let sort_expr = PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0))); - let config = - FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), file_source) - .with_file_groups(vec![FileGroup::new(vec![ - make_file_with_stats("file1", 0.0, 9.0), - make_file_with_stats("file2", 10.0, 19.0), - ])]) - .with_output_ordering(vec![ - LexOrdering::new(vec![sort_expr.clone()]).unwrap(), - ]) - .with_limit(Some(5)) - .build(); - - let result = config.try_pushdown_sort(&[sort_expr.reverse()])?; - - assert!( - matches!(result, SortOrderPushdownResult::Unsupported), - "Expected Unsupported result, got {result:?}" - ); - Ok(()) - } - #[test] fn sort_pushdown_exact_source_overlapping_downgraded_to_inexact() -> Result<()> { let file_schema = diff --git a/datafusion/physical-optimizer/src/filter_pushdown.rs b/datafusion/physical-optimizer/src/filter_pushdown.rs index 1a2bf3d7c46d9..4f3bb23e88ace 100644 --- a/datafusion/physical-optimizer/src/filter_pushdown.rs +++ b/datafusion/physical-optimizer/src/filter_pushdown.rs @@ -430,6 +430,16 @@ fn push_down_filters( config: &ConfigOptions, phase: FilterPushdownPhase, ) -> Result>> { + // A node with a fetch picks its rows before a filter from above applies, so + // that filter can't move below or into it. The node's own filters still can. + if node.fetch().is_some() && !parent_predicates.is_empty() { + let result = push_down_filters(node, vec![], config, phase)?; + return Ok(FilterPushdownPropagation { + filters: vec![PushedDown::No; parent_predicates.len()], + updated_node: result.updated_node, + }); + } + let mut parent_filter_pushdown_supports: Vec> = vec![vec![]; parent_predicates.len()]; diff --git a/datafusion/physical-optimizer/src/pushdown_sort.rs b/datafusion/physical-optimizer/src/pushdown_sort.rs index 7e3268dd19c49..556cde584d7ff 100644 --- a/datafusion/physical-optimizer/src/pushdown_sort.rs +++ b/datafusion/physical-optimizer/src/pushdown_sort.rs @@ -91,6 +91,9 @@ impl PhysicalOptimizerRule for PushdownSort { && sort_child.preserve_partitioning() { let sort_input = Arc::clone(sort_child.input()); + if input_has_fetch(&sort_input)? { + return Ok(Transformed::no(plan)); + } let required_ordering = sort_child.expr(); match sort_input.try_pushdown_sort(required_ordering)? { SortOrderPushdownResult::Exact { inner } => { @@ -144,6 +147,9 @@ impl PhysicalOptimizerRule for PushdownSort { }; let sort_input = Arc::clone(sort_exec.input()); + if input_has_fetch(&sort_input)? { + return Ok(Transformed::no(plan)); + } let required_ordering = sort_exec.expr(); // Try to push the sort requirement down through the plan tree @@ -216,3 +222,9 @@ impl PhysicalOptimizerRule for PushdownSort { true } } + +/// Whether a node below this sort has a fetch. A fetch keeps the first rows in the +/// input's current order, so changing that order would change which rows it keeps. +fn input_has_fetch(input: &Arc) -> Result { + input.exists(|node| Ok(node.fetch().is_some())) +} diff --git a/datafusion/sqllogictest/test_files/sort_pushdown_exact_reverse.slt b/datafusion/sqllogictest/test_files/sort_pushdown_exact_reverse.slt index 203319a3821a0..e9b6f1090c84b 100644 --- a/datafusion/sqllogictest/test_files/sort_pushdown_exact_reverse.slt +++ b/datafusion/sqllogictest/test_files/sort_pushdown_exact_reverse.slt @@ -482,6 +482,52 @@ SELECT ts FROM days ORDER BY date_trunc('day', ts) DESC LIMIT 4; 1970-01-03T12:00:00 1970-01-03T18:00:00 +# Test 16: #27194, with the option off (the default). A subquery's +# `ORDER BY ... LIMIT` over a sorted file leaves the scan holding the limit; +# neither a reversing sort pushdown nor a pushed-down filter may change which +# rows that limit keeps. +statement ok +SET datafusion.execution.parquet.enable_exact_reverse_scan = false; + +query I +COPY (SELECT value AS s, value * 10 AS x FROM generate_series(1, 12) ORDER BY s) +TO 'test_files/scratch/sort_pushdown_exact_reverse/issue_27194/t.parquet' +STORED AS PARQUET OPTIONS ('format.max_row_group_size' '3'); +---- +12 + +query I +COPY (SELECT value AS k FROM generate_series(7, 8)) +TO 'test_files/scratch/sort_pushdown_exact_reverse/issue_27194/b.parquet' +STORED AS PARQUET; +---- +2 + +statement ok +CREATE EXTERNAL TABLE issue_27194_t STORED AS PARQUET +LOCATION 'test_files/scratch/sort_pushdown_exact_reverse/issue_27194/t.parquet'; + +statement ok +CREATE EXTERNAL TABLE issue_27194_b STORED AS PARQUET +LOCATION 'test_files/scratch/sort_pushdown_exact_reverse/issue_27194/b.parquet'; + +query I +SELECT s FROM (SELECT * FROM issue_27194_t ORDER BY s LIMIT 3) ORDER BY s DESC; +---- +3 +2 +1 + +query I +SELECT s FROM (SELECT * FROM issue_27194_t ORDER BY s LIMIT 3) ORDER BY x DESC LIMIT 1; +---- +3 + +query I +SELECT i.s FROM issue_27194_b JOIN (SELECT s FROM issue_27194_t ORDER BY s LIMIT 3) i +ON issue_27194_b.k = i.s; +---- + # Cleanup statement ok RESET datafusion.execution.parquet.enable_exact_reverse_scan;