fix: SortExec.fetch was 0 when create_filter was called

create_filter() was called before new_sort.fetch was set, so
DynamicFilterPhysicalExpr.fetch was always 0 (or None from old self).
Fix by setting fetch before creating the filter.

This was the root cause of stats init and cumulative prune not
triggering on CI — fetch=0 meant "no rows needed" → skip.

Author: zhuqi-lucas

datafusion/physical-plan/src/sorts/sort.rs

@@ -962,14 +962,16 @@ impl SortExec {
         if fetch.is_some() && is_pipeline_friendly {
             cache = cache.with_boundedness(Boundedness::Bounded);
         }
-        let filter = fetch.is_some().then(|| {
-            // If we already have a filter, keep it. Otherwise, create a new one.
-            self.filter.clone().unwrap_or_else(|| self.create_filter())
-        });
         let mut new_sort = self.cloned();
         new_sort.fetch = fetch;
         new_sort.cache = cache.into();
-        new_sort.filter = filter;
+        new_sort.filter = fetch.is_some().then(|| {
+            // If we already have a filter, keep it. Otherwise, create a new one.
+            // Must be called after setting fetch so DynamicFilter gets the K value.
+            self.filter
+                .clone()
+                .unwrap_or_else(|| new_sort.create_filter())
+        });
         new_sort
     }