Merge branch 'main' into implement_groups_accumulator_count_distinct_primitive_types

Author: coderfender

Cargo.lock

@@ -100,6 +100,7 @@ arrow-avro = { version = "58.1.0", default-features = false, features = [
     "xz",
 ] }
 arrow-buffer = { version = "58.1.0", default-features = false }
+arrow-data = { version = "58.1.0", default-features = false }
 arrow-flight = { version = "58.1.0", features = [
     "flight-sql-experimental",
 ] }
@@ -192,7 +193,7 @@ strum = "0.28.0"
 strum_macros = "0.28.0"
 tempfile = "3"
 testcontainers-modules = { version = "0.15" }
-tokio = { version = "1.51", features = ["macros", "rt", "sync"] }
+tokio = { version = "1.52", features = ["macros", "rt", "sync"] }
 tokio-stream = "0.1"
 tokio-util = "0.7"
 url = "2.5.7"

Cargo.toml

@@ -311,6 +311,15 @@ config_namespace! {
         /// By default, `nulls_max` is used to follow Postgres's behavior.
         /// postgres rule: <https://www.postgresql.org/docs/current/queries-order.html>
         pub default_null_ordering: String, default = "nulls_max".to_string()
+
+        /// When set to true, DataFusion may remove `ORDER BY` clauses from
+        /// subqueries or CTEs during SQL planning when their ordering cannot
+        /// affect the result, such as when no `LIMIT` or other
+        /// order-sensitive operator depends on them.
+        ///
+        /// Disable this option to preserve explicit subquery ordering in the
+        /// planned query.
+        pub enable_subquery_sort_elimination: bool, default = true
     }
 }
 

datafusion/common/src/config.rs

@@ -1276,7 +1276,14 @@ fn make_staggered_batches_i32(len: usize, with_extra_column: bool) -> Vec<Record
     input12.sort_unstable();
     let input1 = Int32Array::from_iter_values(input12.clone().into_iter().map(|k| k.0));
     let input2 = Int32Array::from_iter_values(input12.clone().into_iter().map(|k| k.1));
-    let input3 = Int32Array::from_iter_values(input3);
+    let input3 = Int32Array::from_iter(input3.into_iter().map(|v| {
+        // ~10% NULLs in filter column to exercise NULL filter handling
+        if rng.random_range(0..10) == 0 {
+            None
+        } else {
+            Some(v)
+        }
+    }));
     let input4 = Int32Array::from_iter_values(input4);
 
     let mut columns = vec![