feat: enable stats init for ALL TopK queries + fix fuzz test tiebreaker

Stats init now fires for all TopK queries, not just sort pushdown path.
The null-aware filter (IS NULL OR col > threshold for NULLS FIRST)
ensures correctness when NULLs are present.

Fix fuzz test: add remaining columns as ASC NULLS LAST tiebreakers to
ORDER BY, making the sort fully deterministic. This is the correct
approach since SQL doesn't guarantee tie-breaking order, and any
optimization that changes RG read order may produce different but
equally valid results for tied rows.

Author: zhuqi-lucas

datafusion/core/tests/fuzz_cases/topk_filter_pushdown.rs

@@ -305,21 +305,36 @@ async fn test_fuzz_topk_filter_pushdown() {
     }
 
     let mut queries = vec![];
+    let all_columns = ["id", "name", "department"];
 
     for limit in [1, 10] {
         for num_order_by_columns in [1, 2, 3] {
-            for order_columns in ["id", "name", "department"]
-                .iter()
-                .combinations(num_order_by_columns)
-            {
+            for order_columns in all_columns.iter().combinations(num_order_by_columns) {
                 for orderings in order_columns
                     .iter()
                     .map(|col| orders.get(**col).unwrap())
                     .multi_cartesian_product()
                 {
+                    // Add remaining columns as ASC tiebreakers to make
+                    // the ordering fully deterministic. Without this,
+                    // optimizations that change RG read order (e.g.
+                    // statistics-based pruning) may produce different
+                    // but equally valid tie-breaking results.
+                    let used: Vec<&str> = order_columns.iter().map(|c| **c).collect();
+                    let tiebreakers: Vec<String> = all_columns
+                        .iter()
+                        .filter(|c| !used.contains(*c))
+                        .map(|c| format!("{c} ASC NULLS LAST"))
+                        .collect();
+                    let mut all_orderings: Vec<&str> =
+                        orderings.iter().map(|s| s.as_str()).collect();
+                    let tiebreaker_refs: Vec<&str> =
+                        tiebreakers.iter().map(|s| s.as_str()).collect();
+                    all_orderings.extend(tiebreaker_refs);
+
                     let query = format!(
                         "SELECT * FROM test_table ORDER BY {} LIMIT {}",
-                        orderings.into_iter().join(", "),
+                        all_orderings.join(", "),
                         limit
                     );
                     queries.push(query);

datafusion/datasource-parquet/src/opener.rs

@@ -838,11 +838,10 @@ impl MetadataLoadedParquetOpen {
         // BEFORE building the pruning predicate. The PruningPredicate compiles
         // the expression at build time, so the DynamicFilterPhysicalExpr must
         // already have the threshold set for pruning to be effective.
-        // Only when sort pushdown is active (sort_order_for_reorder set).
-        // For non-sort-pushdown TopK, pruning may change tie-breaking for
-        // equal values across RGs. Future: extend to all TopK once
-        // tie-breaking is handled or fuzz tests are updated.
-        if prepared.sort_order_for_reorder.is_some() {
+        // Works for ALL TopK queries. The filter is null-aware (NULLS FIRST
+        // adds `col IS NULL OR ...`) to avoid incorrectly pruning RGs with
+        // NULL values that belong in the result.
+        {
             let file_metadata = reader_metadata.metadata();
             let rg_metadata = file_metadata.row_groups();
             let topk_limit = prepared.limit.unwrap_or(1);