diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 63a7797e396ef..35a987b3d03ca 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1367,6 +1367,16 @@ 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. 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`, /// 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/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-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..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::{ @@ -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,32 @@ 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. `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 + && 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 +1667,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 +1699,7 @@ impl RowGroupsPrunedParquetOpen { if prepared.preserve_order { None } else { - prepared.limit + decode_order_limit }, ); access_plan = page_pruning_result.access_plan; @@ -1708,6 +1749,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 +1803,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 +1962,16 @@ impl RowGroupsPrunedParquetOpen { filter_installed, row_filter_skipped_fully_matched, byte_progress, + 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(); @@ -2156,6 +2218,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 +2231,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 +2292,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 +2514,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 +2625,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 +2740,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 +4526,267 @@ 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]); + } + + /// 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 + /// 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..8469036f88558 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,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, + /// 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. @@ -432,13 +438,30 @@ impl PushDecoderStreamState { let elapsed_compute = self.baseline_metrics.elapsed_compute().clone(); let mut timer = elapsed_compute.timer(); loop { + // 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 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)); @@ -447,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; + } } } } @@ -708,6 +739,117 @@ impl PushDecoderStreamState { } } +/// 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 + { + reservation.shrink(dropped.get_array_memory_size()); + } + } + } + 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 = 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 `pop` + // releases batch by batch. + for batch in self.pending.drain(..) { + self.reversed.push(reverse_batch_rows(&batch)?); + } + if let Some(reservation) = &self.reservation { + let stored = self + .reversed + .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)] mod tests { use super::*; diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index ac7e10f94e6f5..c818cb3512f66 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,29 @@ 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 + { + // 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 = !self.reverse_rows; + new_source.reverse_rows = !self.reverse_rows; + return Ok(SortOrderPushdownResult::Exact { + inner: Arc::new(new_source) as Arc, + }); + } + if !column_in_file_schema && !reversed_satisfies { return Ok(SortOrderPushdownResult::Unsupported); } @@ -1101,6 +1161,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 +1196,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 +1238,7 @@ impl ParquetSource { parquet_options, sort_order_for_reorder, reverse_row_groups, + reverse_rows, metadata_size_hint, } = scan; @@ -1254,6 +1319,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 +1762,106 @@ 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()); + } + + /// 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] + 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..f370c584fff90 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)?; @@ -1141,6 +1144,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) @@ -1173,54 +1178,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) @@ -3548,6 +3518,8 @@ mod tests { struct ExactSortPushdownSource { metrics: ExecutionPlanMetricsSet, table_schema: TableSchema, + /// Whether this is the source returned by `try_pushdown_sort`. + pushed: bool, } impl ExactSortPushdownSource { @@ -3555,6 +3527,7 @@ mod tests { Self { metrics: ExecutionPlanMetricsSet::new(), table_schema, + pushed: false, } } } @@ -3582,7 +3555,11 @@ mod tests { } fn file_type(&self) -> &str { - "mock_exact" + if self.pushed { + "mock_exact_pushed" + } else { + "mock_exact" + } } fn try_pushdown_sort( @@ -3590,8 +3567,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, }) } @@ -3734,6 +3713,139 @@ 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()); + // 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 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`. + /// The scan reverses its files and orderings for it just as for a literal + /// reverse. + #[test] + 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); + 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.clone()]).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::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, ["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(()) + } + #[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..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,24 +83,34 @@ 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; + // 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 = reversed_orderings(&self.output_ordering); + 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 +134,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 +175,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. // @@ -201,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 @@ -287,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/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/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 ee1d803bd2d60..5aaadaaa78b8e 100644 --- a/datafusion/proto-models/proto/datafusion.proto +++ b/datafusion/proto-models/proto/datafusion.proto @@ -1383,6 +1383,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 34f2a293ae816..1444b37de5251 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 06342881ecac3..0df3ba14ca72b 100644 --- a/datafusion/proto-models/src/generated/prost.rs +++ b/datafusion/proto-models/src/generated/prost.rs @@ -2127,6 +2127,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 c1aabc2946ea6..407f3fdd5f6c7 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -243,6 +243,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 @@ -412,6 +413,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. 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 new file mode 100644 index 0000000000000..e9b6f1090c84b --- /dev/null +++ b/datafusion/sqllogictest/test_files/sort_pushdown_exact_reverse.slt @@ -0,0 +1,546 @@ +# 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 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 +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 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; +---- +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 + +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 + +# 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; + +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 2c3ca887d4a38..c7a2fabe131a2 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. 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. |