refactor(parquet sampling): address PR apache#22000 review feedback

Two changes responding to review on the parent commit:

1. Key sampling on a stable `file_index` instead of `file_name`
   (apache#22000 (comment)).

   Both `apply_row_group_sampling` and `apply_row_fraction_sampling`
   now take `file_index: usize` rather than `file_name: &str`. The
   parquet opener passes the execution `partition_index`. This makes
   sampling reproducible across environments (no dependency on the
   on-disk path), while still decorrelating files assigned to
   different partitions.

2. Extract the row-window selection into `build_row_window_selectors`
   and add fuzz coverage
   (apache#22000 (comment)).

   The previous inline arithmetic could produce overlapping windows
   when `target_rows` was close to `total_rows`: `window_size =
   ceil(target / n_windows)` could exceed `stride = total / n_windows`,
   so adjacent strides' windows would intersect. The extracted
   function caps `window_size` at `stride` (the construction that
   guarantees disjointness) and is covered by:

   * `row_window_selection_basic_layout` — hand-checked anchor case.
   * `row_window_selection_returns_none_on_invalid_input` — degenerate
     inputs return `None` cleanly.
   * `row_window_selection_full_target_no_overlap` — the previously
     buggy `target_rows == total_rows` case.
   * `row_window_selection_fuzz_invariants` — 5 000 randomized
     `(total_rows, target_rows, cluster_size, seed)` configurations,
     asserting full coverage, in-bounds positions, and no overlap.
   * `row_window_selection_fuzz_determinism` — 1 000 iterations
     verifying identical seeds produce identical layouts.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: adriangb

datafusion/datasource-parquet/src/opener.rs

@@ -893,24 +893,24 @@ impl FiltersPreparedParquetOpen {
             rg_metadata.len(),
         )?;
 
-        // SYSTEM-mode adaptive split: when the SamplePushdown rule
-        // hands us a residual fraction `remaining_p`, choose the
-        // row-group / row split based on the row-group count we just
-        // observed. With ≥ 2 row groups we split as `sqrt(remaining)`
-        // at both axes; with 1 row group we apply the full residual at
-        // the row level (the row-group axis can't reduce). This keeps
-        // the expected output close to `p × N_total` even for tiny
-        // scans where the cube-root math otherwise undershoots
-        // (single-file / single-row-group inputs would read
-        // `cbrt(p)` of the rows, ~46% for SYSTEM(10)).
+        // Apply optional row-group and row-range sampling now that we
+        // know the actual row-group count. Selection is deterministic
+        // per `(partition_index, row_group_index, fraction,
+        // cluster_size)` so re-runs match. The execution
+        // `partition_index` is the stable per-file id we plumb in: it
+        // makes sampling reproducible across environments without
+        // depending on object-store paths, and decorrelates files
+        // assigned to different partitions.
         if let Some(remaining) = prepared.sampling.system_target_remaining {
-            // SYSTEM-mode adaptive split: choose the row-group / row
-            // axes based on the row-group count we just observed. With
-            // ≥ 2 row groups split as `sqrt(remaining)` at both axes;
-            // with 1 row group skip row-group sampling and apply the
-            // full residual at the row level. Without this adaptation
-            // a single-file / single-row-group scan would only reach
-            // `cbrt(remaining)` of the rows (~46% for SYSTEM(10)).
+            // SYSTEM-mode adaptive split: when the SamplePushdown rule
+            // hands us a residual fraction `remaining`, choose the
+            // row-group / row split based on the row-group count we
+            // just observed. With ≥ 2 row groups split as
+            // `sqrt(remaining)` at both axes; with 1 row group skip
+            // row-group sampling and apply the full residual at the
+            // row level. Without this adaptation a single-file /
+            // single-row-group scan would only reach `cbrt(remaining)`
+            // of the rows (~46% for SYSTEM(10)).
             let n_rg = rg_metadata.len();
             let mut adapted = prepared.sampling.clone();
             if n_rg >= 2 {
@@ -924,25 +924,25 @@ impl FiltersPreparedParquetOpen {
             adapted.apply_row_group_sampling(
                 &mut initial_plan,
                 n_rg,
-                &prepared.file_name,
+                prepared.partition_index,
             );
             adapted.apply_row_fraction_sampling(
                 &mut initial_plan,
                 rg_metadata,
-                &prepared.file_name,
+                prepared.partition_index,
             );
         } else {
             // Legacy direct-builder path: each method is a no-op when
             // its corresponding fraction is `None`.
             prepared.sampling.apply_row_group_sampling(
                 &mut initial_plan,
                 rg_metadata.len(),
-                &prepared.file_name,
+                prepared.partition_index,
             );
             prepared.sampling.apply_row_fraction_sampling(
                 &mut initial_plan,
                 rg_metadata,
-                &prepared.file_name,
+                prepared.partition_index,
             );
         }