feat: wire shared invariant functions into all production check_invariant! calls (#6344)

* fix: SS-2 invariant — nulls always sort last, regardless of direction

The SS-2 invariant incorrectly specified that nulls sort before non-null
for descending columns. The actual production code uses nulls_first=false
everywhere — nulls always sort after non-null values, regardless of
direction. This is required for sorted_series key correctness: null
columns are omitted from the key, and their keys naturally sort after
keys that include those columns.

Fixed in all locations:
- TLA+ SortSchema.tla: CompareValues and SS2_NullOrdering
- Rust invariants/sort.rs: compare_with_null_ordering
- Rust models/sort_schema.rs: SS-2 property check and compare_values test
- Rust invariants/registry.rs: SS-2 description
- ADR-002: SS-2 invariant table
- Phase 1 design doc: null handling descriptions

Stateright exhaustive_sort_schema BFS passes with corrected model.

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

* test: verify model detects null ordering violations

Added two tests to the Stateright sort_schema model:

- ss2_detects_null_before_non_null_in_descending: constructs rows
  with null before non-null in a descending column, verifies both
  row_leq and is_sorted reject this ordering (nulls always last).

- ss2_accepts_non_null_before_null_in_descending: verifies the
  correct ordering (non-null then null) is accepted.

These confirm the model can catch the exact class of bug that the
SS-2 invariant fix addresses.

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

* feat: add SS-2 production invariant check using shared model function

Adds verify_ss2_null_ordering to the writer's sort verification path.
This calls quickwit_dst::invariants::sort::compare_with_null_ordering —
the same function the Stateright model uses for exhaustive BFS
verification — to confirm that:

1. The shared invariant function returns Greater for (null, non-null)
   regardless of sort direction (debug_assert at function entry)
2. No adjacent row pair in the sorted output has null before non-null
   in any sort column (when earlier columns are equal)

This bridges the gap where SS-2 was verified by the model but never
checked at runtime in production code. The check runs in debug builds
(via check_invariant! / debug_assert!) and reports to the invariant
recorder in all builds.

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

* refactor: use shared invariant functions for all TW-2 checks

All TW-2 (window_duration divides 3600) checks now call the shared
quickwit_dst::invariants::window::is_valid_window_duration instead
of inlining the `3600 % duration_secs == 0` logic. This ensures
production and model execute identical validation:

- sort_fields/window.rs: validate_window_duration + window_start
- storage/writer.rs: build_compaction_key_value_metadata
- split/metadata.rs: ParquetSplitMetadataBuilder::build

The shared functions used by production check_invariant! calls are
now the same code the Stateright models use:
- SS-2: compare_with_null_ordering (sort.rs)
- TW-2: is_valid_window_duration (window.rs)
- TW-1: window_start_secs (window.rs) — already shared

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

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: g-talbot

quickwit/quickwit-parquet-engine/src/sort_fields/window.rs

@@ -48,7 +48,7 @@ pub fn validate_window_duration(duration_secs: u32) -> Result<(), SortFieldsErro
             reason: "must be positive",
         });
     }
-    if 3600 % duration_secs != 0 {
+    if !quickwit_dst::invariants::window::is_valid_window_duration(duration_secs) {
         return Err(SortFieldsError::InvalidWindowDuration {
             duration_secs,
             reason: "must evenly divide 3600 (one hour)",
@@ -87,7 +87,8 @@ pub fn window_start(
     );
     check_invariant!(
         InvariantId::TW2,
-        3600 % duration_secs == 0,
+        duration_secs > 0
+            && quickwit_dst::invariants::window::is_valid_window_duration(duration_secs as u32),
         ": duration_secs={} does not divide 3600",
         duration_secs
     );

quickwit/quickwit-parquet-engine/src/split/metadata.rs

@@ -532,7 +532,10 @@ impl ParquetSplitMetadataBuilder {
         // Enforced at build time so no invalid metadata propagates to storage.
         quickwit_dst::check_invariant!(
             quickwit_dst::invariants::InvariantId::TW2,
-            self.window_duration_secs == 0 || 3600 % self.window_duration_secs == 0,
+            self.window_duration_secs == 0
+                || quickwit_dst::invariants::window::is_valid_window_duration(
+                    self.window_duration_secs
+                ),
             ": window_duration_secs={} does not divide 3600",
             self.window_duration_secs
         );

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

@@ -69,7 +69,10 @@ pub(crate) fn build_compaction_key_value_metadata(
     // but belt-and-suspenders at the serialization boundary).
     quickwit_dst::check_invariant!(
         quickwit_dst::invariants::InvariantId::TW2,
-        metadata.window_duration_secs() == 0 || 3600 % metadata.window_duration_secs() == 0,
+        metadata.window_duration_secs() == 0
+            || quickwit_dst::invariants::window::is_valid_window_duration(
+                metadata.window_duration_secs()
+            ),
         " at Parquet write: window_duration_secs={} does not divide 3600",
         metadata.window_duration_secs()
     );
@@ -308,6 +311,17 @@ impl ParquetWriter {
                         i
                     );
                 }
+
+                // SS-2: verify nulls always sort after non-null values.
+                // Uses the shared compare_with_null_ordering from
+                // quickwit_dst::invariants::sort — the same function the
+                // Stateright model uses for exhaustive verification.
+                // For each adjacent row pair and each sort column, when
+                // earlier columns are equal, the comparison must not
+                // yield Greater (which would mean null came before non-null).
+                if sorted_batch.num_rows() > 1 {
+                    verify_ss2_null_ordering(&sorted_batch, &self.resolved_sort_fields);
+                }
             }
         }
 
@@ -552,6 +566,101 @@ fn resolve_sort_fields(sort_fields_str: &str) -> Result<Vec<ResolvedSortField>,
         .collect())
 }
 
+/// SS-2: Verify that nulls always sort after non-null values in the sorted batch.
+///
+/// Uses [`quickwit_dst::invariants::sort::compare_with_null_ordering`] — the
+/// same function the Stateright model (`models::sort_schema`) uses for
+/// exhaustive verification. This ensures production and model execute
+/// identical null-ordering logic.
+///
+/// For each pair of adjacent rows and each sort column, when all earlier
+/// sort columns have equal null/non-null status at both rows, we call
+/// `compare_with_null_ordering(None, Some(&()), ascending)` to ask the
+/// shared invariant: "does null come before non-null?" If the answer is
+/// Greater (null should sort after), but we see null at row i and non-null
+/// at row i+1, that's a violation.
+fn verify_ss2_null_ordering(batch: &RecordBatch, sort_fields: &[ResolvedSortField]) {
+    let schema = batch.schema();
+    let num_rows = batch.num_rows();
+    if num_rows <= 1 {
+        return;
+    }
+
+    // Resolve sort column indices.
+    let sort_cols: Vec<(usize, &str, bool)> = sort_fields
+        .iter()
+        .filter_map(|sf| {
+            schema
+                .index_of(sf.name.as_str())
+                .ok()
+                .map(|idx| (idx, sf.name.as_str(), !sf.descending))
+        })
+        .collect();
+
+    // Ask the shared invariant function: does null sort after non-null?
+    // This must return Greater (null > non-null) for SS-2 to hold.
+    // We check once and use the result — if the shared function is wrong,
+    // this test and the Stateright model will both fail.
+    let null_vs_nonnull = quickwit_dst::invariants::sort::compare_with_null_ordering(
+        None::<&u8>,
+        Some(&0u8),
+        true, // ascending — but SS-2 says direction doesn't matter
+    );
+    debug_assert_eq!(
+        null_vs_nonnull,
+        std::cmp::Ordering::Greater,
+        "SS-2: compare_with_null_ordering must return Greater for (null, non-null)"
+    );
+
+    // Also verify for descending — the shared function must give the same answer.
+    let null_vs_nonnull_desc = quickwit_dst::invariants::sort::compare_with_null_ordering(
+        None::<&u8>,
+        Some(&0u8),
+        false, // descending
+    );
+    debug_assert_eq!(
+        null_vs_nonnull_desc,
+        std::cmp::Ordering::Greater,
+        "SS-2: compare_with_null_ordering must return Greater for (null, non-null) in descending \
+         too"
+    );
+
+    // Now check the actual data: for each adjacent row pair, null must not
+    // appear before non-null when earlier columns are equal.
+    for i in 0..num_rows - 1 {
+        for (k, &(col_idx, col_name, _ascending)) in sort_cols.iter().enumerate() {
+            let col = batch.column(col_idx);
+
+            // Only check when earlier columns are equal at rows i and i+1.
+            // Two values are equal if both null, or both non-null with the
+            // same scalar value (checked via single-element slice comparison).
+            let earlier_equal = sort_cols[..k].iter().all(|&(earlier_idx, _, _)| {
+                let c = batch.column(earlier_idx);
+                let a_null = c.is_null(i);
+                let b_null = c.is_null(i + 1);
+                if a_null != b_null {
+                    return false;
+                }
+                if a_null {
+                    return true; // both null
+                }
+                // Both non-null: compare single-element slices.
+                c.slice(i, 1) == c.slice(i + 1, 1)
+            });
+
+            if earlier_equal && col.is_null(i) && !col.is_null(i + 1) {
+                quickwit_dst::check_invariant!(
+                    quickwit_dst::invariants::InvariantId::SS2,
+                    false,
+                    ": null before non-null in column '{}' at row {}",
+                    col_name,
+                    i
+                );
+            }
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;