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

@@ -2250,10 +2250,8 @@ mod tests {
             1,
         );
         let bytes_b = make_nullable_prefix_input_single_rg(&[None, None, None], 1);
-        let inputs: Vec<Box<dyn ColumnPageStream>> = vec![
-            open_stream(bytes_a).await,
-            open_stream(bytes_b).await,
-        ];
+        let inputs: Vec<Box<dyn ColumnPageStream>> =
+            vec![open_stream(bytes_a).await, open_stream(bytes_b).await];
 
         let tmp = TempDir::new().expect("tmpdir");
         let outputs = streaming_merge_sorted_parquet_files(inputs, tmp.path(), &merge_config(1))
@@ -2278,15 +2276,21 @@ mod tests {
         use arrow::array::StringArray;
         let combined = read_output_to_record_batch(&outputs[0].path);
         let mn_idx = combined.schema().index_of("metric_name").expect("mn col");
-        let arr = combined.column(mn_idx).as_any().downcast_ref::<StringArray>();
+        let arr = combined
+            .column(mn_idx)
+            .as_any()
+            .downcast_ref::<StringArray>();
         let arr = arr.expect("metric_name should decode as StringArray");
         // First 3 rows are the non-null region, last 3 are all-null.
         for i in 0..3 {
             assert!(arr.is_valid(i), "row {i} should be non-null");
             assert_eq!(arr.value(i), "cpu.usage");
         }
         for i in 3..6 {
-            assert!(arr.is_null(i), "row {i} should be null (all-null region sorts last)");
+            assert!(
+                arr.is_null(i),
+                "row {i} should be null (all-null region sorts last)"
+            );
         }
     }
 
@@ -2690,7 +2694,7 @@ mod tests {
         Bytes::from(buf)
     }
 
-/// End-to-end regression for DESC prefix columns. Three RGs with
+    /// End-to-end regression for DESC prefix columns. Three RGs with
     /// the same metric_name (ASC) and distinct `env` values; sort
     /// schema declares env DESC. The input file must itself be
     /// DESC-sorted on env (RGs in physical order staging → prod →
@@ -2785,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!(
@@ -245,9 +267,8 @@ pub(crate) fn extract_rg_composite_prefix_key(
             bail!(
                 "input {input_idx} rg {rg_idx} col '{}' is NOT prefix-aligned: contains \
                  {null_count} nulls plus {} non-null values. PA-1 requires each row group to \
-                 carry a single prefix value; rows with null on this column encode to a \
-                 different prefix key (with the column skipped) than rows with the non-null \
-                 value.",
+                 carry a single prefix value; rows with null on this column encode to a different \
+                 prefix key (with the column skipped) than rows with the non-null value.",
                 col.name,
                 num_values - null_count,
             );
@@ -259,22 +280,17 @@ pub(crate) fn extract_rg_composite_prefix_key(
     // Trailing prefix-length sentinel: an additional `u8(prefix_len)`
     // ordinal byte that does two things at once:
     //
-    // 1. **Forces nulls-last ordering across RGs.** For prefix_len=1
-    //    an all-null RG produces an empty per-column body and would
-    //    otherwise lex-sort *before* any non-null RG. With the
-    //    sentinel, the all-null key becomes `[prefix_len]` and the
-    //    non-null key becomes `[ord(0), storekey(value), ..., prefix_len]`.
-    //    The non-null key starts with `ord(0) = 0x00`, smaller than
-    //    `prefix_len >= 1`, so non-null RGs sort first — matching
-    //    `sorted_series`'s row-level nulls-last convention via the
-    //    same "the next ordinal byte appears in the skipped slot"
+    // 1. **Forces nulls-last ordering across RGs.** For prefix_len=1 an all-null RG produces an
+    //    empty per-column body and would otherwise lex-sort *before* any non-null RG. With the
+    //    sentinel, the all-null key becomes `[prefix_len]` and the non-null key becomes `[ord(0),
+    //    storekey(value), ..., prefix_len]`. The non-null key starts with `ord(0) = 0x00`, smaller
+    //    than `prefix_len >= 1`, so non-null RGs sort first — matching `sorted_series`'s row-level
+    //    nulls-last convention via the same "the next ordinal byte appears in the skipped slot"
     //    mechanism.
-    // 2. **Preserves the "literal prefix of sorted_series" property.**
-    //    The byte we append is exactly what `sorted_series` writes
-    //    right after the prefix columns: the ordinal of the next
-    //    sort-schema column (`u8(prefix_len)`). So the per-RG key
-    //    remains a byte-for-byte prefix of every row's
-    //    `sorted_series` value in that RG.
+    // 2. **Preserves the "literal prefix of sorted_series" property.** The byte we append is
+    //    exactly what `sorted_series` writes right after the prefix columns: the ordinal of the
+    //    next sort-schema column (`u8(prefix_len)`). So the per-RG key remains a byte-for-byte
+    //    prefix of every row's `sorted_series` value in that RG.
     storekey::encode(&mut key, &(prefix_cols.len() as u8))
         .map_err(|e| anyhow!("storekey encode prefix-length sentinel: {}", e))?;
 
@@ -581,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
@@ -606,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/sorted_series/mod.rs

@@ -286,9 +286,10 @@ pub(crate) fn append_prefix_col_to_key<T>(
     value: &T,
     descending: bool,
 ) -> Result<()>
-where T: ?Sized + storekey::Encode {
-    storekey::encode(&mut *buf, &ordinal)
-        .map_err(|e| anyhow!("storekey encode ordinal: {}", e))?;
+where
+    T: ?Sized + storekey::Encode,
+{
+    storekey::encode(&mut *buf, &ordinal).map_err(|e| anyhow!("storekey encode ordinal: {}", e))?;
     let value_start = buf.len();
     storekey::encode(&mut *buf, value).map_err(|e| anyhow!("storekey encode value: {}", e))?;
     if descending {

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)`