feat: add MP-1/MP-2/MP-3 runtime invariant checks for merge operations

Add three merge-policy invariants to the shared verification layer
(quickwit-dst) with check_invariant! enforcement in
ParquetMergeOperation::new():

- MP-1: all splits share the same num_merge_ops level
- MP-2: merge op has at least 2 input splits
- MP-3: all splits share the same compaction scope (sort_fields + window)

Shared pure functions in quickwit_dst::invariants::merge_policy are the
single source of truth, usable by Stateright models and production code.
Debug builds panic on violation; all builds emit metrics via the
invariant recorder.

Tests written first (4 should_panic tests failed before adding checks,
pass after). Plus 1 positive test and 3 unit tests for shared functions.

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

Author: g-talbot

quickwit/quickwit-dst/src/invariants/merge_policy.rs

@@ -0,0 +1,113 @@
+// Copyright 2021-Present Datadog, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Shared merge policy invariant checks.
+//!
+//! These pure functions are the single source of truth for merge operation
+//! validity, used by both Stateright models and production code.
+
+/// MP-1: all splits in a merge operation must share the same `num_merge_ops`.
+///
+/// If splits from different levels are merged, the output gets stamped with
+/// `max(levels) + 1`, prematurely maturing lower-level data and breaking
+/// the bounded write amplification guarantee.
+pub fn all_same_merge_level(num_merge_ops: &[u32]) -> bool {
+    match num_merge_ops.first() {
+        None => true,
+        Some(&first) => num_merge_ops.iter().all(|&n| n == first),
+    }
+}
+
+/// MP-2: every merge operation must have at least 2 input splits.
+///
+/// Merging a single split is a no-op that wastes I/O. Merging zero splits
+/// is nonsensical.
+pub fn has_minimum_splits(count: usize) -> bool {
+    count >= 2
+}
+
+/// MP-3: all splits in a merge operation must share the same compaction scope.
+///
+/// The scope is defined by `(sort_fields, window_start, window_duration)`.
+/// The merge engine validates that all inputs agree on these; a policy bug
+/// that groups incompatible splits will cause the merge to fail.
+///
+/// `index_uid` and `partition_id` are also part of the scope but are
+/// typically enforced by the grouping layer before the policy runs.
+pub fn all_same_compaction_scope(
+    sort_fields: &[&str],
+    windows: &[(i64, i64)], // (start, duration) pairs
+) -> bool {
+    let same_sort = match sort_fields.first() {
+        None => true,
+        Some(&first) => sort_fields.iter().all(|&s| s == first),
+    };
+    let same_window = match windows.first() {
+        None => true,
+        Some(&first) => windows.iter().all(|w| *w == first),
+    };
+    same_sort && same_window
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_all_same_merge_level() {
+        assert!(all_same_merge_level(&[]));
+        assert!(all_same_merge_level(&[0]));
+        assert!(all_same_merge_level(&[2, 2, 2]));
+        assert!(!all_same_merge_level(&[0, 1]));
+        assert!(!all_same_merge_level(&[0, 0, 1]));
+    }
+
+    #[test]
+    fn test_has_minimum_splits() {
+        assert!(!has_minimum_splits(0));
+        assert!(!has_minimum_splits(1));
+        assert!(has_minimum_splits(2));
+        assert!(has_minimum_splits(100));
+    }
+
+    #[test]
+    fn test_all_same_compaction_scope() {
+        // Empty is vacuously true.
+        assert!(all_same_compaction_scope(&[], &[]));
+
+        // Same scope.
+        assert!(all_same_compaction_scope(
+            &["a|b|ts/V2", "a|b|ts/V2"],
+            &[(0, 3600), (0, 3600)],
+        ));
+
+        // Different sort fields.
+        assert!(!all_same_compaction_scope(
+            &["a|b|ts/V2", "a|ts/V2"],
+            &[(0, 3600), (0, 3600)],
+        ));
+
+        // Same start, different duration.
+        assert!(!all_same_compaction_scope(
+            &["a|b|ts/V2", "a|b|ts/V2"],
+            &[(0, 900), (0, 1800)],
+        ));
+
+        // Different start.
+        assert!(!all_same_compaction_scope(
+            &["a|b|ts/V2", "a|b|ts/V2"],
+            &[(0, 3600), (3600, 3600)],
+        ));
+    }
+}

quickwit/quickwit-dst/src/invariants/mod.rs

@@ -21,6 +21,7 @@
 //! No external dependencies — only `std`.
 
 mod check;
+pub mod merge_policy;
 pub mod recorder;
 pub mod registry;
 pub mod sort;

quickwit/quickwit-dst/src/invariants/registry.rs

@@ -74,6 +74,13 @@ pub enum InvariantId {
     DM4,
     /// DM-5: timeseries_id persists through compaction without recomputation
     DM5,
+
+    /// MP-1: all splits in a merge operation have the same num_merge_ops level
+    MP1,
+    /// MP-2: every merge operation has at least 2 input splits
+    MP2,
+    /// MP-3: all splits in a merge operation share the same compaction scope
+    MP3,
 }
 
 impl InvariantId {
@@ -106,6 +113,10 @@ impl InvariantId {
             Self::DM3 => "DM-3",
             Self::DM4 => "DM-4",
             Self::DM5 => "DM-5",
+
+            Self::MP1 => "MP-1",
+            Self::MP2 => "MP-2",
+            Self::MP3 => "MP-3",
         }
     }
 
@@ -136,6 +147,10 @@ impl InvariantId {
             Self::DM3 => "no interpolation — only ingested points",
             Self::DM4 => "deterministic TSID from tags",
             Self::DM5 => "TSID persists through compaction",
+
+            Self::MP1 => "merge op splits share num_merge_ops level",
+            Self::MP2 => "merge op has at least 2 splits",
+            Self::MP3 => "merge op splits share compaction scope",
         }
     }
 }
@@ -157,6 +172,7 @@ mod tests {
         assert_eq!(InvariantId::CS3.to_string(), "CS-3");
         assert_eq!(InvariantId::MC4.to_string(), "MC-4");
         assert_eq!(InvariantId::DM5.to_string(), "DM-5");
+        assert_eq!(InvariantId::MP1.to_string(), "MP-1");
     }
 
     #[test]
@@ -182,6 +198,9 @@ mod tests {
             InvariantId::DM3,
             InvariantId::DM4,
             InvariantId::DM5,
+            InvariantId::MP1,
+            InvariantId::MP2,
+            InvariantId::MP3,
         ];
         for id in all {
             assert!(!id.description().is_empty(), "{} has empty description", id);

quickwit/quickwit-parquet-engine/src/merge/policy/mod.rs

@@ -50,7 +50,49 @@ impl ParquetMergeOperation {
     ///
     /// Generates a fresh split ID for the merged output. The `kind` is inferred
     /// from the first split (all splits in a merge share the same kind).
+    ///
+    /// # Invariant checks (debug builds panic, all builds emit metrics)
+    ///
+    /// - **MP-1**: all splits share the same `num_merge_ops` level
+    /// - **MP-2**: at least 2 input splits
+    /// - **MP-3**: all splits share the same compaction scope (sort_fields + window)
     pub fn new(splits: Vec<ParquetSplitMetadata>) -> Self {
+        use quickwit_dst::check_invariant;
+        use quickwit_dst::invariants::InvariantId;
+        use quickwit_dst::invariants::merge_policy;
+
+        // MP-2: minimum split count.
+        check_invariant!(
+            InvariantId::MP2,
+            merge_policy::has_minimum_splits(splits.len()),
+            ": got {} splits",
+            splits.len()
+        );
+
+        // MP-1: level homogeneity.
+        let levels: Vec<u32> = splits.iter().map(|s| s.num_merge_ops).collect();
+        check_invariant!(
+            InvariantId::MP1,
+            merge_policy::all_same_merge_level(&levels),
+            ": levels={:?}",
+            levels
+        );
+
+        // MP-3: scope homogeneity (sort_fields + window).
+        let sort_fields_vec: Vec<&str> =
+            splits.iter().map(|s| s.sort_fields.as_str()).collect();
+        let windows: Vec<(i64, i64)> = splits
+            .iter()
+            .map(|s| match &s.window {
+                Some(w) => (w.start, w.end - w.start),
+                None => (0, 0),
+            })
+            .collect();
+        check_invariant!(
+            InvariantId::MP3,
+            merge_policy::all_same_compaction_scope(&sort_fields_vec, &windows)
+        );
+
         let kind = splits
             .first()
             .map(|s| s.kind)
@@ -123,3 +165,93 @@ pub trait ParquetMergePolicy: Send + Sync + fmt::Debug {
     ) {
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashSet;
+    use std::time::SystemTime;
+
+    use super::*;
+    use crate::split::{ParquetSplitId, ParquetSplitKind, TimeRange};
+
+    fn make_split(
+        split_id: &str,
+        num_merge_ops: u32,
+        sort_fields: &str,
+        window: Option<(i64, i64)>,
+    ) -> ParquetSplitMetadata {
+        ParquetSplitMetadata {
+            kind: ParquetSplitKind::Metrics,
+            split_id: ParquetSplitId::new(split_id),
+            index_uid: "test:001".to_string(),
+            partition_id: 0,
+            time_range: TimeRange::new(1000, 2000),
+            num_rows: 100,
+            size_bytes: 1_000_000,
+            metric_names: HashSet::new(),
+            low_cardinality_tags: Default::default(),
+            high_cardinality_tag_keys: Default::default(),
+            created_at: SystemTime::now(),
+            parquet_file: format!("{split_id}.parquet"),
+            window: window.map(|(start, dur)| start..start + dur),
+            sort_fields: sort_fields.to_string(),
+            num_merge_ops,
+            row_keys_proto: None,
+            zonemap_regexes: Default::default(),
+        }
+    }
+
+    #[test]
+    #[should_panic(expected = "MP-1 violated")]
+    fn test_mp1_mixed_merge_levels_panics() {
+        // MP-1: constructing a merge op with splits at different levels
+        // must panic in debug builds.
+        let splits = vec![
+            make_split("l0", 0, "a|ts/V2", Some((0, 3600))),
+            make_split("l1", 1, "a|ts/V2", Some((0, 3600))),
+        ];
+        ParquetMergeOperation::new(splits);
+    }
+
+    #[test]
+    #[should_panic(expected = "MP-2 violated")]
+    fn test_mp2_single_split_panics() {
+        // MP-2: constructing a merge op with < 2 splits must panic.
+        let splits = vec![make_split("s0", 0, "a|ts/V2", Some((0, 3600)))];
+        ParquetMergeOperation::new(splits);
+    }
+
+    #[test]
+    #[should_panic(expected = "MP-3 violated")]
+    fn test_mp3_mixed_sort_fields_panics() {
+        // MP-3: constructing a merge op with different sort_fields must panic.
+        let splits = vec![
+            make_split("s0", 0, "a|b|ts/V2", Some((0, 3600))),
+            make_split("s1", 0, "a|ts/V2", Some((0, 3600))),
+        ];
+        ParquetMergeOperation::new(splits);
+    }
+
+    #[test]
+    #[should_panic(expected = "MP-3 violated")]
+    fn test_mp3_mixed_window_duration_panics() {
+        // MP-3: same start but different duration must panic.
+        let splits = vec![
+            make_split("s0", 0, "a|ts/V2", Some((0, 900))),
+            make_split("s1", 0, "a|ts/V2", Some((0, 1800))),
+        ];
+        ParquetMergeOperation::new(splits);
+    }
+
+    #[test]
+    fn test_valid_merge_operation_succeeds() {
+        // A well-formed merge op should not panic.
+        let splits = vec![
+            make_split("s0", 0, "a|ts/V2", Some((0, 3600))),
+            make_split("s1", 0, "a|ts/V2", Some((0, 3600))),
+        ];
+        let op = ParquetMergeOperation::new(splits);
+        assert_eq!(op.splits.len(), 2);
+    }
+}
+