Rename to preserve_order

xudong963 · xudong963 · commit 24ae747203d1 · 2025-11-27T17:36:02.000+08:00
diff --git a/datafusion/catalog-listing/src/table.rs b/datafusion/catalog-listing/src/table.rs
@@ -506,7 +506,7 @@ impl TableProvider for ListingTable {
                     .with_statistics(statistics)
                     .with_projection_indices(projection)
                     .with_limit(limit)
-                    .with_limit_order_sensitive(args.limit_order_sensitive())
+                    .with_preserve_order(args.preserve_order())
                     .with_output_ordering(output_ordering)
                     .with_expr_adapter(self.expr_adapter_factory.clone())
                     .build(),
diff --git a/datafusion/catalog/src/table.rs b/datafusion/catalog/src/table.rs
@@ -336,7 +336,7 @@ pub struct ScanArgs<'a> {
     filters: Option<&'a [Expr]>,
     projection: Option<&'a [usize]>,
     limit: Option<usize>,
-    limit_order_sensitive: bool,
+    preserve_order: bool,
 }
 
 impl<'a> ScanArgs<'a> {
@@ -399,23 +399,15 @@ impl<'a> ScanArgs<'a> {
         self.limit
     }
 
-    /// Set whether the scan's limit should be order-sensitive.
-    ///
-    /// If specified, the scan should return the limited rows in a specific order.
-    /// Or we can leverage limit pruning to optimize the scan.
-    ///
-    /// # Arguments
-    /// * `order_sensitive` - Whether the scan's limit should be order-sensitive
-    pub fn with_limit_order_sensitive(mut self, order_sensitive: bool) -> Self {
-        self.limit_order_sensitive = order_sensitive;
+    /// Set whether should keep the output rows in order
+    pub fn with_preserve_order(mut self, order_sensitive: bool) -> Self {
+        self.preserve_order = order_sensitive;
         self
     }
 
-    /// Get whether the scan's limit should be order-sensitive.
-    ///
-    /// Returns `true` if the scan's limit should be order-sensitive, or `false` if not.
-    pub fn limit_order_sensitive(&self) -> bool {
-        self.limit_order_sensitive
+    /// Get whether should keep the output rows in order
+    pub fn preserve_order(&self) -> bool {
+        self.preserve_order
     }
 }
 
diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs
@@ -457,7 +457,7 @@ impl DefaultPhysicalPlanner {
                 projection,
                 filters,
                 fetch,
-                fetch_order_sensitive,
+                preserve_order,
                 ..
             }) => {
                 let source = source_as_provider(source)?;
@@ -470,7 +470,7 @@ impl DefaultPhysicalPlanner {
                     .with_projection(projection.as_deref())
                     .with_filters(Some(&filters_vec))
                     .with_limit(*fetch)
-                    .with_limit_order_sensitive(*fetch_order_sensitive);
+                    .with_preserve_order(*preserve_order);
                 let res = source.scan_with_args(session_state, opts).await?;
                 Arc::clone(res.plan())
             }
diff --git a/datafusion/datasource-parquet/src/opener.rs b/datafusion/datasource-parquet/src/opener.rs
@@ -68,8 +68,8 @@ pub(super) struct ParquetOpener {
     pub batch_size: usize,
     /// Optional limit on the number of rows to read
     pub limit: Option<usize>,
-    /// limit order sensitivity
-    pub limit_order_sensitive: bool,
+    /// If should keep the output rows in order
+    pub preserve_order: bool,
     /// Optional predicate to apply during the scan
     pub predicate: Option<Arc<dyn PhysicalExpr>>,
     /// Schema of the output table without partition columns.
@@ -165,7 +165,7 @@ impl FileOpener for ParquetOpener {
         let encryption_context = self.get_encryption_context();
         let max_predicate_cache_size = self.max_predicate_cache_size;
 
-        let limit_order_sensitive = self.limit_order_sensitive;
+        let preserve_order = self.preserve_order;
 
         Ok(Box::pin(async move {
             #[cfg(feature = "parquet_encryption")]
@@ -412,7 +412,7 @@ impl FileOpener for ParquetOpener {
             }
 
             // Prune by limit if limit is set and limit order is not sensitive
-            if let (Some(limit), false) = (limit, limit_order_sensitive) {
+            if let (Some(limit), false) = (limit, preserve_order) {
                 row_groups.prune_by_limit(limit, rg_metadata, &file_metrics);
             }
 
@@ -885,7 +885,7 @@ mod test {
                 projection: Arc::new([0, 1]),
                 batch_size: 1024,
                 limit: None,
-                limit_order_sensitive: false,
+                preserve_order: false,
                 predicate: Some(predicate),
                 logical_file_schema: schema.clone(),
                 metadata_size_hint: None,
@@ -955,7 +955,7 @@ mod test {
                 projection: Arc::new([0]),
                 batch_size: 1024,
                 limit: None,
-                limit_order_sensitive: false,
+                preserve_order: false,
                 predicate: Some(predicate),
                 logical_file_schema: file_schema.clone(),
                 metadata_size_hint: None,
@@ -1045,7 +1045,7 @@ mod test {
                 projection: Arc::new([0]),
                 batch_size: 1024,
                 limit: None,
-                limit_order_sensitive: false,
+                preserve_order: false,
                 predicate: Some(predicate),
                 logical_file_schema: file_schema.clone(),
                 metadata_size_hint: None,
@@ -1138,7 +1138,7 @@ mod test {
                 projection: Arc::new([0]),
                 batch_size: 1024,
                 limit: None,
-                limit_order_sensitive: false,
+                preserve_order: false,
                 predicate: Some(predicate),
                 logical_file_schema: file_schema.clone(),
                 metadata_size_hint: None,
@@ -1231,7 +1231,7 @@ mod test {
                 projection: Arc::new([0]),
                 batch_size: 1024,
                 limit: None,
-                limit_order_sensitive: false,
+                preserve_order: false,
                 predicate: Some(predicate),
                 logical_file_schema: file_schema.clone(),
                 metadata_size_hint: None,
@@ -1386,7 +1386,7 @@ mod test {
             projection: Arc::new([0, 1]),
             batch_size: 1024,
             limit: None,
-            limit_order_sensitive: false,
+            preserve_order: false,
             predicate: Some(predicate),
             logical_file_schema: Arc::clone(&table_schema),
             metadata_size_hint: None,
diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs
@@ -583,7 +583,7 @@ impl FileSource for ParquetSource {
                 .batch_size
                 .expect("Batch size must set before creating ParquetOpener"),
             limit: base_config.limit,
-            limit_order_sensitive: base_config.limit_order_sensitive,
+            preserve_order: base_config.preserve_order,
             predicate: self.predicate.clone(),
             logical_file_schema: Arc::clone(base_config.file_schema()),
             partition_fields: base_config.table_partition_cols().clone(),
diff --git a/datafusion/datasource/src/file_scan_config.rs b/datafusion/datasource/src/file_scan_config.rs
@@ -176,7 +176,7 @@ pub struct FileScanConfig {
     /// all records after filtering are returned.
     pub limit: Option<usize>,
     /// Whether the scan's limit is order sensitive
-    pub limit_order_sensitive: bool,
+    pub preserve_order: bool,
     /// All equivalent lexicographical orderings that describe the schema.
     pub output_ordering: Vec<LexOrdering>,
     /// File compression type
@@ -258,7 +258,7 @@ pub struct FileScanConfigBuilder {
     object_store_url: ObjectStoreUrl,
     file_source: Arc<dyn FileSource>,
     limit: Option<usize>,
-    limit_order_sensitive: bool,
+    preserve_order: bool,
     projection_indices: Option<Vec<usize>>,
     constraints: Option<Constraints>,
     file_groups: Vec<FileGroup>,
@@ -290,7 +290,7 @@ impl FileScanConfigBuilder {
             file_compression_type: None,
             new_lines_in_values: None,
             limit: None,
-            limit_order_sensitive: false,
+            preserve_order: false,
             projection_indices: None,
             constraints: None,
             batch_size: None,
@@ -306,8 +306,8 @@ impl FileScanConfigBuilder {
     }
 
     /// Set whether the limit should be order-sensitive.
-    pub fn with_limit_order_sensitive(mut self, order_sensitive: bool) -> Self {
-        self.limit_order_sensitive = order_sensitive;
+    pub fn with_preserve_order(mut self, order_sensitive: bool) -> Self {
+        self.preserve_order = order_sensitive;
         self
     }
 
@@ -438,7 +438,7 @@ impl FileScanConfigBuilder {
             object_store_url,
             file_source,
             limit,
-            limit_order_sensitive,
+            preserve_order,
             projection_indices,
             constraints,
             file_groups,
@@ -472,7 +472,7 @@ impl FileScanConfigBuilder {
             object_store_url,
             file_source,
             limit,
-            limit_order_sensitive,
+            preserve_order,
             projection_exprs,
             constraints,
             file_groups,
@@ -497,7 +497,7 @@ impl From<FileScanConfig> for FileScanConfigBuilder {
             file_compression_type: Some(config.file_compression_type),
             new_lines_in_values: Some(config.new_lines_in_values),
             limit: config.limit,
-            limit_order_sensitive: config.limit_order_sensitive,
+            preserve_order: config.preserve_order,
             projection_indices: config
                 .projection_exprs
                 .map(|p| p.ordered_column_indices()),
diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs
@@ -2652,9 +2652,8 @@ pub struct TableScan {
     pub filters: Vec<Expr>,
     /// Optional number of rows to read
     pub fetch: Option<usize>,
-    /// If the fetch is order-sensitive, it'll be true.
-    /// And the limit pruning will be enabled.
-    pub fetch_order_sensitive: bool,
+    /// If should keep the output rows in order
+    pub preserve_order: bool,
 }
 
 impl Debug for TableScan {
@@ -2677,7 +2676,7 @@ impl PartialEq for TableScan {
             && self.projected_schema == other.projected_schema
             && self.filters == other.filters
             && self.fetch == other.fetch
-            && self.fetch_order_sensitive == other.fetch_order_sensitive
+            && self.preserve_order == other.preserve_order
     }
 }
 
@@ -2698,21 +2697,21 @@ impl PartialOrd for TableScan {
             /// Optional number of rows to read
             pub fetch: &'a Option<usize>,
             /// Whether the fetch is order-sensitive
-            pub fetch_order_sensitive: bool,
+            pub preserve_order: bool,
         }
         let comparable_self = ComparableTableScan {
             table_name: &self.table_name,
             projection: &self.projection,
             filters: &self.filters,
             fetch: &self.fetch,
-            fetch_order_sensitive: self.fetch_order_sensitive,
+            preserve_order: self.preserve_order,
         };
         let comparable_other = ComparableTableScan {
             table_name: &other.table_name,
             projection: &other.projection,
             filters: &other.filters,
             fetch: &other.fetch,
-            fetch_order_sensitive: other.fetch_order_sensitive,
+            preserve_order: other.preserve_order,
         };
         comparable_self
             .partial_cmp(&comparable_other)
@@ -2728,7 +2727,7 @@ impl Hash for TableScan {
         self.projected_schema.hash(state);
         self.filters.hash(state);
         self.fetch.hash(state);
-        self.fetch_order_sensitive.hash(state);
+        self.preserve_order.hash(state);
     }
 }
 
@@ -2782,7 +2781,7 @@ impl TableScan {
             projected_schema,
             filters,
             fetch,
-            fetch_order_sensitive: false,
+            preserve_order: false,
         })
     }
 }
@@ -4942,7 +4941,7 @@ mod tests {
             projected_schema: Arc::clone(&schema),
             filters: vec![],
             fetch: None,
-            fetch_order_sensitive: false,
+            preserve_order: false,
         }));
         let col = schema.field_names()[0].clone();
 
@@ -4973,7 +4972,7 @@ mod tests {
             projected_schema: Arc::clone(&unique_schema),
             filters: vec![],
             fetch: None,
-            fetch_order_sensitive: false,
+            preserve_order: false,
         }));
         let col = schema.field_names()[0].clone();
 
diff --git a/datafusion/expr/src/logical_plan/tree_node.rs b/datafusion/expr/src/logical_plan/tree_node.rs
@@ -599,7 +599,7 @@ impl LogicalPlan {
                 projected_schema,
                 filters,
                 fetch,
-                fetch_order_sensitive,
+                preserve_order,
             }) => filters.map_elements(f)?.update_data(|filters| {
                 LogicalPlan::TableScan(TableScan {
                     table_name,
@@ -608,7 +608,7 @@ impl LogicalPlan {
                     projected_schema,
                     filters,
                     fetch,
-                    fetch_order_sensitive,
+                    preserve_order,
                 })
             }),
             LogicalPlan::Distinct(Distinct::On(DistinctOn {
diff --git a/datafusion/optimizer/src/optimize_projections/mod.rs b/datafusion/optimizer/src/optimize_projections/mod.rs
@@ -260,7 +260,7 @@ fn optimize_projections(
                 projection,
                 filters,
                 fetch,
-                fetch_order_sensitive,
+                preserve_order,
                 projected_schema: _,
             } = table_scan;
 
@@ -272,7 +272,7 @@ fn optimize_projections(
             };
             let mut new_scan =
                 TableScan::try_new(table_name, source, Some(projection), filters, fetch)?;
-            new_scan.fetch_order_sensitive = fetch_order_sensitive;
+            new_scan.preserve_order = preserve_order;
 
             return Ok(Transformed::yes(LogicalPlan::TableScan(new_scan)));
         }
diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs
@@ -3119,7 +3119,7 @@ mod tests {
             projection,
             source: Arc::new(test_provider),
             fetch: None,
-            fetch_order_sensitive: false,
+            preserve_order: false,
         });
 
         Ok(LogicalPlanBuilder::from(table_scan))
diff --git a/datafusion/optimizer/src/push_down_limit.rs b/datafusion/optimizer/src/push_down_limit.rs
@@ -124,8 +124,7 @@ impl OptimizerRule for PushDownLimit {
                 })),
 
             LogicalPlan::Sort(mut sort) => {
-                let marked_input =
-                    mark_fetch_order_sensitive(Arc::unwrap_or_clone(sort.input))?;
+                let marked_input = mark_preserve_order(Arc::unwrap_or_clone(sort.input))?;
                 sort.input = Arc::new(marked_input);
                 let new_fetch = {
                     let sort_fetch = skip + fetch;
@@ -271,10 +270,10 @@ fn push_down_join(mut join: Join, limit: usize) -> Transformed<Join> {
     Transformed::yes(join)
 }
 
-fn mark_fetch_order_sensitive(plan: LogicalPlan) -> Result<LogicalPlan> {
+fn mark_preserve_order(plan: LogicalPlan) -> Result<LogicalPlan> {
     plan.transform_down(|node| match node {
         LogicalPlan::TableScan(mut scan) => {
-            scan.fetch_order_sensitive = true;
+            scan.preserve_order = true;
             Ok(Transformed::yes(LogicalPlan::TableScan(scan)))
         }
         _ => Ok(Transformed::no(node)),
@@ -1147,11 +1146,11 @@ mod test {
         )
     }
 
-    fn has_fetch_order_sensitive_scan(plan: &LogicalPlan) -> bool {
+    fn has_preserve_order_scan(plan: &LogicalPlan) -> bool {
         let mut found = false;
         plan.apply(|node| {
             if let LogicalPlan::TableScan(scan) = node {
-                if scan.fetch_order_sensitive {
+                if scan.preserve_order {
                     found = true;
                     return Ok(TreeNodeRecursion::Stop);
                 }
@@ -1177,7 +1176,7 @@ mod test {
         let optimized_plan =
             Optimizer::with_rules(rules).optimize(plan, &optimizer_ctx, |_, _| {})?;
 
-        assert!(has_fetch_order_sensitive_scan(&optimized_plan));
+        assert!(has_preserve_order_scan(&optimized_plan));
 
         Ok(())
     }
diff --git a/datafusion/proto/src/logical_plan/mod.rs b/datafusion/proto/src/logical_plan/mod.rs
