fix(31): close port gaps — split_writer metadata, compaction scope, publish validation

Close critical gaps identified during port review:

split_writer.rs:
- Store table_config on ParquetSplitWriter (not just pass-through)
- Compute window_start from batch time range using table_config.window_duration_secs
- Populate sort_fields, window_duration_secs, parquet_files on metadata before write
- Call write_to_file_with_metadata(Some(&metadata)) to embed KV metadata in Parquet
- Update size_bytes after write completes

metastore/mod.rs:
- Add window_start and sort_fields fields to ListMetricsSplitsQuery
- Add with_compaction_scope() builder method

metastore/postgres/metastore.rs:
- Add compaction scope filters (AND window_start = $N, AND sort_fields = $N) to list query
- Add replaced_split_ids count verification in publish_metrics_splits
- Bind compaction scope query parameters

ingest/config.rs:
- Add table_config: TableConfig field to ParquetIngestConfig

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

Author: g-talbot

quickwit/quickwit-metastore/src/metastore/mod.rs

@@ -75,6 +75,10 @@ pub struct ListMetricsSplitsQuery {
     pub tag_region: Option<String>,
     /// Host tag filter.
     pub tag_host: Option<String>,
+    /// Window start filter for compaction scope queries.
+    pub window_start: Option<i64>,
+    /// Sort fields filter for compaction scope queries.
+    pub sort_fields: Option<String>,
     /// Limit number of results.
     pub limit: Option<usize>,
 }
@@ -107,6 +111,17 @@ impl ListMetricsSplitsQuery {
         self.metric_names = names;
         self
     }
+
+    /// Filter by compaction scope (window_start + sort_fields).
+    pub fn with_compaction_scope(
+        mut self,
+        window_start: i64,
+        sort_fields: impl Into<String>,
+    ) -> Self {
+        self.window_start = Some(window_start);
+        self.sort_fields = Some(sort_fields.into());
+        self
+    }
 }
 
 /// Splits batch size returned by the stream splits API

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

@@ -2167,13 +2167,13 @@ impl MetastoreService for PostgresqlMetastore {
                 return Err(MetastoreError::FailedPrecondition { entity, message });
             }
 
-            // Verify all replaced splits were marked for deletion
+            // Verify all replaced splits were marked for deletion.
             if marked_count as usize != replaced_split_ids.len() {
                 let entity = EntityKind::Splits {
                     split_ids: replaced_split_ids.clone(),
                 };
                 let message = format!(
-                    "expected to mark {} splits for deletion, but only {} were in Published state",
+                    "expected to replace {} splits, but only {} were in Published state",
                     replaced_split_ids.len(),
                     marked_count
                 );
@@ -2273,6 +2273,16 @@ impl MetastoreService for PostgresqlMetastore {
             param_idx += 1;
         }
 
+        // Compaction scope filters
+        if query.window_start.is_some() {
+            sql.push_str(&format!(" AND window_start = ${}", param_idx));
+            param_idx += 1;
+        }
+        if query.sort_fields.is_some() {
+            sql.push_str(&format!(" AND sort_fields = ${}", param_idx));
+            param_idx += 1;
+        }
+
         sql.push_str(" ORDER BY time_range_start ASC");
 
         // Add limit
@@ -2332,6 +2342,12 @@ impl MetastoreService for PostgresqlMetastore {
         if let Some(ref host) = query.tag_host {
             query_builder = query_builder.bind(host);
         }
+        if let Some(ws) = query.window_start {
+            query_builder = query_builder.bind(ws);
+        }
+        if let Some(ref sf) = query.sort_fields {
+            query_builder = query_builder.bind(sf);
+        }
         if let Some(limit) = query.limit {
             query_builder = query_builder.bind(limit as i64);
         }

quickwit/quickwit-parquet-engine/src/index/config.rs

@@ -17,6 +17,7 @@
 use std::sync::OnceLock;
 
 use crate::storage::ParquetWriterConfig;
+use crate::table_config::TableConfig;
 
 /// Default maximum rows to accumulate before flushing to split.
 const DEFAULT_MAX_ROWS: usize = 1_000_000;
@@ -58,6 +59,8 @@ pub struct ParquetIndexingConfig {
     pub max_bytes: usize,
     /// Parquet writer configuration for split creation.
     pub writer_config: ParquetWriterConfig,
+    /// Table-level configuration (sort fields, window duration, product type).
+    pub table_config: TableConfig,
 }
 
 impl Default for ParquetIndexingConfig {
@@ -66,6 +69,7 @@ impl Default for ParquetIndexingConfig {
             max_rows: get_max_rows_from_env(),
             max_bytes: get_max_bytes_from_env(),
             writer_config: ParquetWriterConfig::default(),
+            table_config: TableConfig::default(),
         }
     }
 }

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

@@ -25,13 +25,16 @@ use tracing::{debug, info, instrument};
 
 use super::config::ParquetWriterConfig;
 use super::writer::{ParquetWriteError, ParquetWriter};
+use crate::sort_fields::window_start;
 use crate::split::{MetricsSplitMetadata, ParquetSplit, SplitId, TAG_SERVICE, TimeRange};
 use crate::table_config::TableConfig;
 
 /// Writer that produces complete ParquetSplit with metadata from RecordBatch data.
 pub struct ParquetSplitWriter {
     /// The underlying Parquet writer.
     writer: ParquetWriter,
+    /// Table configuration (sort fields, window duration, product type).
+    table_config: TableConfig,
     /// Base directory for split files.
     base_path: PathBuf,
 }
@@ -42,13 +45,15 @@ impl ParquetSplitWriter {
     /// # Arguments
     /// * `config` - Parquet writer configuration
     /// * `base_path` - Directory where split files will be written
+    /// * `table_config` - Table-level config (sort fields, window duration)
     pub fn new(
         config: ParquetWriterConfig,
         base_path: impl Into<PathBuf>,
         table_config: &TableConfig,
     ) -> Self {
         Self {
             writer: ParquetWriter::new(config, table_config),
+            table_config: table_config.clone(),
             base_path: base_path.into(),
         }
     }
@@ -60,6 +65,10 @@ impl ParquetSplitWriter {
 
     /// Write a RecordBatch to a Parquet file and return a ParquetSplit with metadata.
     ///
+    /// Builds metadata (including sort fields, window_start, window_duration from
+    /// table_config) before writing, then embeds compaction KV metadata into the
+    /// Parquet file. size_bytes is updated after the write completes.
+    ///
     /// # Arguments
     /// * `batch` - The RecordBatch to write
     /// * `index_uid` - The index unique identifier for the split metadata
@@ -72,57 +81,74 @@ impl ParquetSplitWriter {
         batch: &RecordBatch,
         index_uid: &str,
     ) -> Result<ParquetSplit, ParquetWriteError> {
-        // Generate unique split ID
         let split_id = SplitId::generate();
+        let filename = format!("{}.parquet", split_id);
+        let file_path = self.base_path.join(&filename);
 
-        let file_path = self.base_path.join(format!("{}.parquet", split_id));
-
-        // Ensure the base directory exists
         std::fs::create_dir_all(&self.base_path)?;
 
-        // Write batch to file
-        let size_bytes = self
-            .writer
-            .write_to_file_with_metadata(batch, &file_path, None)?;
-
-        // Extract time range from batch
+        // Extract batch-level metadata before writing.
         let time_range = extract_time_range(batch)?;
         debug!(
             start_secs = time_range.start_secs,
             end_secs = time_range.end_secs,
             "extracted time range from batch"
         );
 
-        // Extract distinct metric names from batch
         let metric_names = extract_metric_names(batch)?;
-
-        // Extract distinct service names from batch
         let service_names = extract_service_names(batch)?;
 
-        // Build metadata
-        let metadata = MetricsSplitMetadata::builder()
+        // Compute window_start from the earliest timestamp in the batch.
+        let window_duration = self.table_config.window_duration_secs;
+        let window_start_secs = if window_duration > 0 && time_range.start_secs > 0 {
+            match window_start(time_range.start_secs as i64, window_duration as i64) {
+                Ok(dt) => Some(dt.timestamp()),
+                Err(e) => {
+                    tracing::warn!(error = %e, "failed to compute window_start, omitting");
+                    None
+                }
+            }
+        } else {
+            None
+        };
+
+        // Build metadata with sort fields and window from table_config.
+        // size_bytes is set to 0 here and updated after write.
+        let mut builder = MetricsSplitMetadata::builder()
             .split_id(split_id.clone())
             .index_uid(index_uid)
             .time_range(time_range)
             .num_rows(batch.num_rows() as u64)
-            .size_bytes(size_bytes);
+            .size_bytes(0)
+            .sort_fields(self.writer.sort_fields_string())
+            .window_duration_secs(window_duration)
+            .add_parquet_file(filename);
+
+        if let Some(ws) = window_start_secs {
+            builder = builder.window_start_secs(ws);
+        }
+
+        for name in metric_names {
+            builder = builder.add_metric_name(name);
+        }
+        for name in service_names {
+            builder = builder.add_low_cardinality_tag(TAG_SERVICE, name);
+        }
 
-        // Add metric names
-        let metadata = metric_names
-            .into_iter()
-            .fold(metadata, |m, name| m.add_metric_name(name));
+        let mut metadata = builder.build();
 
-        // Add service names as low-cardinality tags
-        let metadata = service_names.into_iter().fold(metadata, |m, name| {
-            m.add_low_cardinality_tag(TAG_SERVICE, name)
-        });
+        // Write with compaction metadata embedded in Parquet KV metadata.
+        let size_bytes =
+            self.writer
+                .write_to_file_with_metadata(batch, &file_path, Some(&metadata))?;
 
-        let metadata = metadata.build();
+        metadata.size_bytes = size_bytes;
 
         info!(
             split_id = %split_id,
             file_path = %file_path.display(),
             size_bytes,
+            sort_fields = %self.writer.sort_fields_string(),
             "split file written successfully"
         );