fix(merge): consumer honors SS-3 (move F12 forward from #6426 to #6425)

Previously the F12 fix — "consumer side honors SS-3 missing prefix
columns" — lived in the hardening PR (#6426). At the #6425
isolation level, the legacy adapter records `None` for a prefix
column absent from the parquet schema and stamps
`rg_partition_prefix_len = target_prefix_len` on the output, but
the reader's `find_prefix_parquet_col_indices` bails on any missing
column. So #6425 + #6424 alone would produce a legacy-adapter file
that the streaming-merge reader rejects mid-merge — i.e. a known-
incoherent intermediate stack state.

Move F12 into this PR so the adapter and reader agree at the same
slice:

- `find_prefix_parquet_col_indices` now returns
  `Result<Vec<Option<PrefixColumn>>>`. `Some(_)` when the column
  is present in the parquet schema; `None` per SS-3 when the
  column is named in `qh.sort_fields` but absent from the schema.
- `extract_rg_composite_prefix_key` skips `None` slots entirely
  (no ordinal byte, no value bytes for that column). The trailing
  `u8(prefix_len)` sentinel introduced in the storekey refactor
  keeps the resulting key well-formed across present/absent
  columns.
- Callers that index into `prefix_cols` updated to use
  `.as_ref().expect(…)` where they assume presence.

Existing SS-3 test
`test_missing_prefix_col_treated_as_null_satisfies_alignment` in
`legacy_adapter.rs` gets an `assert_unique_rg_prefix_keys` call
verifying the adapter's output is consumable by the reader — pins
the "stack-coherent at #6425" property the F12 hop establishes.

Also incidental nightly-fmt cleanups in `sorted_series::append_prefix_col_to_key`
and the two-input fixture in `test_all_null_prefix_rg_groups_into_separate_region_sorted_last`.

The hardening PR (#6426) will be re-cascaded to drop the now-
duplicated F12 hunks (keeping its F8 adapter-rejects-unsorted +
F2 verifier-strength changes intact).

485 lib tests pass on this slice; workspace clippy + nightly fmt
clean.

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

Author: g-talbot

quickwit/quickwit-parquet-engine/src/merge/streaming.rs

@@ -2789,7 +2789,10 @@ mod tests {
                 .expect("resolve");
         // Sanity: the second prefix column must be flagged DESC.
         assert!(
-            prefix_cols[1].descending,
+            prefix_cols[1]
+                .as_ref()
+                .expect("env present in this fixture")
+                .descending,
             "env must be parsed as DESC from sort schema",
         );
 

quickwit/quickwit-parquet-engine/src/merge/streaming/region_grouping.rs

@@ -118,14 +118,25 @@ pub(crate) struct PrefixColumn {
 
 /// Resolve the first `prefix_len` sort columns to parquet leaf
 /// indices. Honours the legacy `timestamp` → `timestamp_secs` alias.
-/// Errors if the sort schema has fewer columns than `prefix_len` or
-/// if any column is missing from the parquet schema.
+///
+/// Returns one entry per requested prefix column. `Some(PrefixColumn)`
+/// when the column is present in the parquet schema; `None` when the
+/// column is named in `sort_fields_str` but absent from the parquet
+/// schema. Per SS-3 the missing column is treated as constant null at
+/// every row of the file — [`extract_rg_composite_prefix_key`]
+/// synthesizes a fixed byte sequence in that slot so ordering is
+/// driven entirely by the present columns.
+///
+/// Errors only when the sort schema declares fewer columns than
+/// requested — that means we don't have a *name* for one of the
+/// prefix columns and can't claim alignment on something we can't
+/// identify.
 pub(crate) fn find_prefix_parquet_col_indices(
     metadata: &ParquetMetaData,
     sort_fields_str: &str,
     prefix_len: usize,
-    input_idx: usize,
-) -> Result<Vec<PrefixColumn>> {
+    _input_idx: usize,
+) -> Result<Vec<Option<PrefixColumn>>> {
     let sort_field_schema = parse_sort_fields(sort_fields_str)?;
     if sort_field_schema.column.len() < prefix_len {
         bail!(
@@ -145,34 +156,34 @@ pub(crate) fn find_prefix_parquet_col_indices(
         } else {
             sort_col.name.as_str()
         };
+        let descending = sort_col.sort_direction
+            == quickwit_proto::sortschema::SortColumnDirection::SortDirectionDescending as i32;
         let mut found = None;
         for (col_idx, col) in parquet_schema.columns().iter().enumerate() {
             if col.path().parts()[0] == resolved {
                 found = Some(col_idx);
                 break;
             }
         }
-        let parquet_col_idx = found.ok_or_else(|| {
-            anyhow!(
-                "input {input_idx} parquet schema is missing prefix sort column '{}' (position \
-                 {pos})",
-                sort_col.name,
-            )
-        })?;
-        let descending = sort_col.sort_direction
-            == quickwit_proto::sortschema::SortColumnDirection::SortDirectionDescending as i32;
+        // SS-3: missing column → `None`. The composite-key extractor
+        // skips this slot entirely (no ordinal byte, no value bytes);
+        // the trailing prefix-length sentinel in
+        // `extract_rg_composite_prefix_key` ensures the resulting key
+        // still sorts cleanly relative to RGs with present values
+        // (and matches sorted_series's row-level null-skip).
+        //
         // Ordinal matches the column's position in `qh.sort_fields`.
         // For prefix cols (always the first `prefix_len` entries of
         // the sort schema) the ordinal equals the iteration index
         // `pos`, which is also the ordinal `sorted_series` would
         // assign — so the per-RG prefix key composes as a literal
         // byte prefix of every sorted_series key.
-        prefix_cols.push(PrefixColumn {
+        prefix_cols.push(found.map(|parquet_col_idx| PrefixColumn {
             name: sort_col.name.clone(),
             parquet_col_idx,
             descending,
             ordinal: pos as u8,
-        });
+        }));
     }
     Ok(prefix_cols)
 }
@@ -197,23 +208,34 @@ fn parquet_has_column(
 /// in this RG.
 ///
 /// Null handling:
-/// - **All-null RG on a prefix column**: the column is skipped entirely (the next column's higher
-///   ordinal byte appears in its place), so the RG sorts after any RG carrying a non-null value for
-///   this column. This mirrors the row-level convention in `sorted_series` and gives nulls-last
-///   ordering for free.
+/// - **Column absent from schema (`None` in `prefix_cols`)**: SS-3 case. Every row of the file has
+///   a constant null in this slot, so the contribution to the composite is empty (column skipped).
+///   The trailing prefix-length sentinel keeps the resulting key well-formed.
+/// - **All-null RG on a present prefix column**: column skipped for this RG (the next column's
+///   higher ordinal byte — or the trailing sentinel — appears in its place), so the RG sorts after
+///   any RG carrying a non-null value for this column. Mirrors the row-level convention in
+///   `sorted_series` and gives nulls-last ordering for free.
 /// - **Mixed null + non-null in one RG**: rows in the RG would encode to two distinct prefix keys
 ///   (the non-null value's key and the column-skipped key), breaking the
 ///   at-most-one-prefix-value-per-RG invariant (PA-1). Reject.
 /// - **No nulls**: standard `min == max` check on stats, then encode that single value.
 pub(crate) fn extract_rg_composite_prefix_key(
     metadata: &ParquetMetaData,
     rg_idx: usize,
-    prefix_cols: &[PrefixColumn],
+    prefix_cols: &[Option<PrefixColumn>],
     input_idx: usize,
 ) -> Result<Vec<u8>> {
     let rg_meta = metadata.row_group(rg_idx);
     let mut key = Vec::new();
-    for col in prefix_cols {
+    for col_opt in prefix_cols {
+        let Some(col) = col_opt else {
+            // SS-3 implicit null: column absent from schema, so every
+            // row's value is null. Skip the slot entirely — the
+            // trailing prefix-length sentinel will keep this from
+            // colliding with present-value keys, and sorted_series
+            // applies the same "skip null cols" rule at the row level.
+            continue;
+        };
         let chunk = rg_meta.column(col.parquet_col_idx);
         let stats = chunk.statistics().ok_or_else(|| {
             anyhow!(
@@ -575,10 +597,22 @@ pub(crate) fn extract_regions_from_metadata(
         .collect())
 }
 
-/// Post-write check: verify the parquet file at `metadata` has no two
-/// row groups sharing the same composite prefix key, for the first
-/// `prefix_len` sort columns. Returns `Ok(())` immediately if
-/// `prefix_len == 0` (no alignment claim).
+/// Post-write check: verify every row group in `metadata` satisfies
+/// the prefix-alignment claim declared by `prefix_len`.
+///
+/// Enforces both halves of the prefix-alignment contract in one pass:
+/// - **PA-1 (intra-RG constancy):** within each RG, each of the first `prefix_len` sort columns has
+///   `min == max` (the column is constant across the RG). This is checked transitively by
+///   [`extract_rg_composite_prefix_key`] — it returns an error when any prefix column's chunk stats
+///   show `min != max`.
+/// - **PA-3 (inter-RG uniqueness):** no two RGs share the same composite prefix value. The
+///   streaming engine pairs at most one input RG per region per prefix value, so a duplicate would
+///   silently drop rows or corrupt the body-col / sort-col mapping.
+///
+/// Returns `Ok(())` immediately when `prefix_len == 0` (no claim to
+/// verify) or `num_rgs == 0` (no RGs to check). Single-RG files are
+/// NOT short-circuited — they still go through PA-1 because an
+/// unsorted single-RG file CAN have `min != max` on a prefix column.
 ///
 /// This is the writer-side mirror of the read-side check in
 /// `extract_regions_from_metadata` — both indexing and the compaction
@@ -600,8 +634,8 @@ pub(crate) fn assert_unique_rg_prefix_keys(
         return Ok(());
     }
     let num_rgs = metadata.num_row_groups();
-    if num_rgs <= 1 {
-        // Single-RG (or zero-RG) files vacuously satisfy the invariant.
+    if num_rgs == 0 {
+        // Zero-RG files vacuously satisfy both halves of the claim.
         return Ok(());
     }
     let prefix_cols =

quickwit/quickwit-parquet-engine/src/storage/legacy_adapter.rs

@@ -1586,6 +1586,22 @@ mod tests {
             Some("2"),
             "stamped prefix_len must match caller's request even when one col is implicitly null",
         );
+
+        // SS-3 consumer-side verification: the file the adapter just
+        // produced must be consumable by the merge engine's
+        // `extract_rg_composite_prefix_key` reader. With `env` absent
+        // from the parquet schema, `find_prefix_parquet_col_indices`
+        // returns `None` in that slot and the composite-key extractor
+        // skips it. PA-1 + PA-3 still hold: each RG's metric_name
+        // min == max, and skipping the constant-null `env` slot makes
+        // the RG composite keys differ only by metric_name.
+        crate::merge::streaming::region_grouping::assert_unique_rg_prefix_keys(
+            adapter.metadata(),
+            "metric_name|env|-timestamp_secs/V2",
+            2,
+            "test_missing_prefix_col_treated_as_null_satisfies_alignment adapter output",
+        )
+        .expect("SS-3 null col must satisfy PA-1 + PA-3 (null is constant across all RGs)");
     }
 
     /// Composite-prefix fixture: rows grouped by `(metric, service)`