Fix proto deserialization: add enable_exact_reverse_scan default

Author: zhuqi-lucas

datafusion/common/src/file_options/parquet_writer.rs

@@ -465,6 +465,7 @@ mod tests {
             skip_arrow_metadata: defaults.skip_arrow_metadata,
             coerce_int96: None,
             max_predicate_cache_size: defaults.max_predicate_cache_size,
+            enable_exact_reverse_scan: defaults.enable_exact_reverse_scan,
         }
     }
 
@@ -579,6 +580,8 @@ mod tests {
                 binary_as_string: global_options_defaults.binary_as_string,
                 skip_arrow_metadata: global_options_defaults.skip_arrow_metadata,
                 coerce_int96: None,
+                enable_exact_reverse_scan: global_options_defaults
+                    .enable_exact_reverse_scan,
             },
             column_specific_options,
             key_value_metadata,

datafusion/datasource-parquet/src/opener.rs

@@ -585,10 +585,10 @@ impl FileOpener for ParquetOpener {
             // reversal (in ReversedRowGroupStream), not at the parquet reader
             // level. Applying limit here would read the first N rows in forward
             // order and then reverse them, giving wrong results.
-            if let Some(limit) = limit {
-                if !reverse_rows {
-                    builder = builder.with_limit(limit)
-                }
+            if let Some(limit) = limit
+                && !reverse_rows
+            {
+                builder = builder.with_limit(limit)
             }
 
             if let Some(max_predicate_cache_size) = max_predicate_cache_size {

datafusion/proto-common/src/from_proto/mod.rs

@@ -1013,6 +1013,7 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions {
             max_predicate_cache_size: value.max_predicate_cache_size_opt.map(|opt| match opt {
                 protobuf::parquet_options::MaxPredicateCacheSizeOpt::MaxPredicateCacheSize(v) => Some(v as usize),
             }).unwrap_or(None),
+            enable_exact_reverse_scan: false,
         })
     }
 }

datafusion/proto/src/logical_plan/file_formats.rs

@@ -525,6 +525,7 @@ mod parquet {
             max_predicate_cache_size: proto.max_predicate_cache_size_opt.as_ref().map(|opt| match opt {
                 parquet_options::MaxPredicateCacheSizeOpt::MaxPredicateCacheSize(size) => *size as usize,
             }),
+            enable_exact_reverse_scan: false,
         }
         }
     }

datafusion/sqllogictest/test_files/information_schema.slt

@@ -243,6 +243,7 @@ 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_page_index true
+datafusion.execution.parquet.enable_exact_reverse_scan false
 datafusion.execution.parquet.encoding NULL
 datafusion.execution.parquet.force_filter_selections false
 datafusion.execution.parquet.max_predicate_cache_size NULL
@@ -378,6 +379,7 @@ datafusion.execution.parquet.data_pagesize_limit 1048576 (writing) Sets best eff
 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_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_exact_reverse_scan false (reading) If true, reverse scans produce exact descending order by reversing rows within each row group. This allows the Sort operator to be removed entirely and fetch/limit to be pushed down to the scan. If false (default), reverse scans only reverse row group order (inexact), keeping TopK above for final sorting.
 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, and byte_stream_split. These values are not case sensitive. If NULL, uses default parquet writer setting
 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.max_predicate_cache_size NULL (reading) The maximum predicate cache size, in bytes. When `pushdown_filters` is enabled, sets the maximum memory used to cache the results of predicate evaluation between filter evaluation and output generation. Decreasing this value will reduce memory usage, but may increase IO and CPU usage. None means use the default parquet reader setting. 0 means no caching.

docs/source/user-guide/configs.md

@@ -90,6 +90,7 @@ The following configuration settings are available:
 | datafusion.execution.parquet.coerce_int96                               | NULL                      | (reading) If true, parquet reader will read columns of physical type int96 as originating from a different resolution than nanosecond. This is useful for reading data from systems like Spark which stores microsecond resolution timestamps in an int96 allowing it to write values with a larger date range than 64-bit timestamps with nanosecond resolution.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                            |
 | datafusion.execution.parquet.bloom_filter_on_read                       | true                      | (reading) Use any available bloom filters when reading parquet files                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         |
 | datafusion.execution.parquet.max_predicate_cache_size                   | NULL                      | (reading) The maximum predicate cache size, in bytes. When `pushdown_filters` is enabled, sets the maximum memory used to cache the results of predicate evaluation between filter evaluation and output generation. Decreasing this value will reduce memory usage, but may increase IO and CPU usage. None means use the default parquet reader setting. 0 means no caching.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               |
+| datafusion.execution.parquet.enable_exact_reverse_scan                  | false                     | (reading) If true, reverse scans produce exact descending order by reversing rows within each row group. This allows the Sort operator to be removed entirely and fetch/limit to be pushed down to the scan. If false (default), reverse scans only reverse row group order (inexact), keeping TopK above for final sorting.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                 |
 | datafusion.execution.parquet.data_pagesize_limit                        | 1048576                   | (writing) Sets best effort maximum size of data page in bytes                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                |
 | datafusion.execution.parquet.write_batch_size                           | 1024                      | (writing) Sets write_batch_size in bytes                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     |
 | datafusion.execution.parquet.writer_version                             | 1.0                       | (writing) Sets parquet writer version valid values are "1.0" and "2.0"                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                       |