fix(31): correct postgres types for window_duration_secs and zonemap_regexes

Gap 1: Change window_duration_secs from i32 to Option<i32> in both
PgMetricsSplit and InsertableMetricsSplit. Pre-Phase-31 splits now
correctly map 0 → NULL in PostgreSQL, enabling Phase 32 compaction
queries to use `WHERE window_duration_secs IS NOT NULL` instead of
the fragile `WHERE window_duration_secs > 0`.

Gap 2: Change zonemap_regexes from String to serde_json::Value in
both structs. This maps directly to JSONB in sqlx, avoiding ambiguity
when PostgreSQL JSONB operators are used in Phase 34/35 zonemap pruning.

Gap 3: Add two missing tests:
- test_insertable_from_metadata_with_compaction_fields: verifies all 6
  compaction fields round-trip through InsertableMetricsSplit
- test_insertable_from_metadata_pre_phase31_defaults: verifies pre-Phase-31
  metadata produces window_duration_secs: None, zonemap_regexes: json!({})

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

Author: g-talbot

quickwit/quickwit-metastore/src/metastore/postgres/metastore.rs

@@ -1844,15 +1844,11 @@ impl MetastoreService for PostgresqlMetastore {
             size_bytes_list.push(insertable.size_bytes);
             split_metadata_jsons.push(insertable.split_metadata_json);
             window_starts.push(insertable.window_start);
-            window_duration_secs_list.push(if insertable.window_duration_secs == 0 {
-                None
-            } else {
-                Some(insertable.window_duration_secs)
-            });
+            window_duration_secs_list.push(insertable.window_duration_secs);
             sort_fields_list.push(insertable.sort_fields);
             num_merge_ops_list.push(insertable.num_merge_ops);
             row_keys_list.push(insertable.row_keys);
-            zonemap_regexes_json_list.push(insertable.zonemap_regexes);
+            zonemap_regexes_json_list.push(insertable.zonemap_regexes.to_string());
         }
 
         info!(
@@ -2384,11 +2380,11 @@ impl MetastoreService for PostgresqlMetastore {
                     // to_metadata() — the SQL columns are only used for
                     // filtering and SS-5 consistency checks.
                     window_start: None,
-                    window_duration_secs: 0,
+                    window_duration_secs: None,
                     sort_fields: String::new(),
                     num_merge_ops: 0,
                     row_keys: None,
-                    zonemap_regexes: "{}".to_string(),
+                    zonemap_regexes: serde_json::json!({}),
                 };
 
                 let state = pg_split.split_state().unwrap_or(MetricsSplitState::Staged);

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

@@ -72,11 +72,11 @@ pub struct PgMetricsSplit {
     pub split_metadata_json: String,
     pub update_timestamp: i64,
     pub window_start: Option<i64>,
-    pub window_duration_secs: i32,
+    pub window_duration_secs: Option<i32>,
     pub sort_fields: String,
     pub num_merge_ops: i32,
     pub row_keys: Option<Vec<u8>>,
-    pub zonemap_regexes: String,
+    pub zonemap_regexes: serde_json::Value,
 }
 
 /// Insertable row for metrics_splits table.
@@ -99,11 +99,11 @@ pub struct InsertableMetricsSplit {
     pub size_bytes: i64,
     pub split_metadata_json: String,
     pub window_start: Option<i64>,
-    pub window_duration_secs: i32,
+    pub window_duration_secs: Option<i32>,
     pub sort_fields: String,
     pub num_merge_ops: i32,
     pub row_keys: Option<Vec<u8>>,
-    pub zonemap_regexes: String,
+    pub zonemap_regexes: serde_json::Value,
 }
 
 impl InsertableMetricsSplit {
@@ -114,12 +114,6 @@ impl InsertableMetricsSplit {
     ) -> Result<Self, serde_json::Error> {
         let split_metadata_json = serde_json::to_string(metadata)?;
 
-        let zonemap_regexes_json = if metadata.zonemap_regexes.is_empty() {
-            "{}".to_string()
-        } else {
-            serde_json::to_string(&metadata.zonemap_regexes)?
-        };
-
         Ok(Self {
             split_id: metadata.split_id.as_str().to_string(),
             split_state: state.as_str().to_string(),
@@ -137,11 +131,16 @@ impl InsertableMetricsSplit {
             size_bytes: metadata.size_bytes as i64,
             split_metadata_json,
             window_start: metadata.window_start,
-            window_duration_secs: metadata.window_duration_secs as i32,
+            window_duration_secs: if metadata.window_duration_secs > 0 {
+                Some(metadata.window_duration_secs as i32)
+            } else {
+                None
+            },
             sort_fields: metadata.sort_fields.clone(),
             num_merge_ops: metadata.num_merge_ops as i32,
             row_keys: metadata.row_keys_proto.clone(),
-            zonemap_regexes: zonemap_regexes_json,
+            zonemap_regexes: serde_json::to_value(&metadata.zonemap_regexes)
+                .unwrap_or_else(|_| serde_json::json!({})),
         })
     }
 }
@@ -165,7 +164,7 @@ impl PgMetricsSplit {
         debug_assert_eq!(metadata.window_start, self.window_start);
         debug_assert_eq!(
             metadata.window_duration_secs,
-            self.window_duration_secs as u32
+            self.window_duration_secs.unwrap_or(0) as u32
         );
         debug_assert_eq!(metadata.sort_fields, self.sort_fields);
         debug_assert_eq!(metadata.num_merge_ops, self.num_merge_ops as u32);
@@ -257,6 +256,61 @@ mod tests {
         assert_eq!(insertable.size_bytes, 1024 * 1024);
     }
 
+    #[test]
+    fn test_insertable_from_metadata_with_compaction_fields() {
+        let metadata = MetricsSplitMetadata::builder()
+            .split_id(SplitId::new("compaction-test"))
+            .index_uid("test-index:00000000000000000000000000")
+            .time_range(TimeRange::new(1000, 2000))
+            .num_rows(100)
+            .size_bytes(500)
+            .window_start_secs(1700000000)
+            .window_duration_secs(3600)
+            .sort_fields("metric_name|host|timestamp/V2")
+            .num_merge_ops(2)
+            .row_keys_proto(vec![0x08, 0x01])
+            .add_zonemap_regex("metric_name", "cpu\\..*")
+            .build();
+
+        let insertable =
+            InsertableMetricsSplit::from_metadata(&metadata, MetricsSplitState::Published)
+                .expect("conversion should succeed");
+
+        assert_eq!(insertable.window_start, Some(1700000000));
+        assert_eq!(insertable.window_duration_secs, Some(3600));
+        assert_eq!(insertable.sort_fields, "metric_name|host|timestamp/V2");
+        assert_eq!(insertable.num_merge_ops, 2);
+        assert_eq!(insertable.row_keys, Some(vec![0x08, 0x01]));
+        assert!(insertable.zonemap_regexes.is_object());
+        assert_eq!(
+            insertable.zonemap_regexes["metric_name"],
+            serde_json::json!("cpu\\..*")
+        );
+    }
+
+    #[test]
+    fn test_insertable_from_metadata_pre_phase31_defaults() {
+        let metadata = MetricsSplitMetadata::builder()
+            .split_id(SplitId::new("pre-phase31"))
+            .index_uid("test-index:00000000000000000000000000")
+            .time_range(TimeRange::new(1000, 2000))
+            .build();
+
+        let insertable =
+            InsertableMetricsSplit::from_metadata(&metadata, MetricsSplitState::Staged)
+                .expect("conversion should succeed");
+
+        assert!(insertable.window_start.is_none());
+        assert!(
+            insertable.window_duration_secs.is_none(),
+            "pre-Phase-31 splits should have NULL window_duration_secs"
+        );
+        assert_eq!(insertable.sort_fields, "");
+        assert_eq!(insertable.num_merge_ops, 0);
+        assert!(insertable.row_keys.is_none());
+        assert_eq!(insertable.zonemap_regexes, serde_json::json!({}));
+    }
+
     #[test]
     fn test_pg_split_to_metadata_roundtrip() {
         let original = MetricsSplitMetadata::builder()