Disable prune_by_limit when output ordering is required

When a parquet scan has output ordering constraints (e.g. ORDER BY + LIMIT),
prune_by_limit must not discard partially-matched row groups. A partially-
matched group may contain rows that sort before any fully-matched group,
so skipping it returns incorrect results.

This matches the fix in upstream DataFusion (apache#21190) where
prune_by_limit is guarded by preserve_order.

Add preserve_order: bool to ParquetOpener. Set to true when output_ordering
is non-empty. Guard prune_by_limit with !preserve_order.

Author: zhuqi-lucas

datafusion/datasource-parquet/src/opener.rs

@@ -118,6 +118,10 @@ pub(super) struct ParquetOpener {
     pub max_predicate_cache_size: Option<usize>,
     /// Whether to read row groups in reverse order
     pub reverse_row_groups: bool,
+    /// When `true`, row ordering must be preserved — `prune_by_limit` must not
+    /// discard partially-matched row groups because they may contain rows that
+    /// sort before fully-matched groups.
+    pub preserve_order: bool,
 }
 
 /// Represents a prepared access plan with optional row selection
@@ -262,6 +266,7 @@ impl FileOpener for ParquetOpener {
         let enable_bloom_filter = self.enable_bloom_filter;
         let enable_row_group_stats_pruning = self.enable_row_group_stats_pruning;
         let limit = self.limit;
+        let preserve_order = self.preserve_order;
 
         let predicate_creation_errors = MetricBuilder::new(&self.metrics)
             .global_counter("num_predicate_creation_errors");
@@ -523,8 +528,11 @@ impl FileOpener for ParquetOpener {
                     .add_matched(n_remaining_row_groups);
             }
 
-            // Prune by limit
-            if let Some(limit) = limit {
+            // Prune by limit: only safe when order does not matter.
+            // With preserve_order=true, partially-matched row groups may
+            // contain rows that sort before fully-matched groups, so
+            // discarding them would return incorrect results.
+            if let (Some(limit), false) = (limit, preserve_order) {
                 row_groups.prune_by_limit(limit, rg_metadata, &file_metrics);
             }
 
@@ -1076,6 +1084,7 @@ mod test {
         coerce_int96: Option<arrow::datatypes::TimeUnit>,
         max_predicate_cache_size: Option<usize>,
         reverse_row_groups: bool,
+        preserve_order: bool,
     }
 
     impl ParquetOpenerBuilder {
@@ -1101,6 +1110,7 @@ mod test {
                 coerce_int96: None,
                 max_predicate_cache_size: None,
                 reverse_row_groups: false,
+                preserve_order: false,
             }
         }
 
@@ -1208,6 +1218,7 @@ mod test {
                 encryption_factory: None,
                 max_predicate_cache_size: self.max_predicate_cache_size,
                 reverse_row_groups: self.reverse_row_groups,
+                preserve_order: self.preserve_order,
             }
         }
     }

datafusion/datasource-parquet/src/source.rs

@@ -567,6 +567,7 @@ impl FileSource for ParquetSource {
             encryption_factory: self.get_encryption_factory_with_config(),
             max_predicate_cache_size: self.max_predicate_cache_size(),
             reverse_row_groups: self.reverse_row_groups,
+            preserve_order: !base_config.output_ordering.is_empty(),
         });
         Ok(opener)
     }