review: parquet_file singular, proto doc link, fix metastore accessor

Author: g-talbot

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

@@ -981,7 +981,7 @@ fn metrics_split_matches_query(split: &StoredMetricsSplit, query: &ListMetricsSp
 
     // Filter by compaction scope
     if let Some(ws) = query.window_start
-        && split.metadata.window_start != Some(ws)
+        && split.metadata.window_start() != Some(ws)
     {
         return false;
     }

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

@@ -2441,45 +2441,80 @@ impl MetastoreService for PostgresqlMetastore {
         );
 
         // Only delete splits that are marked for deletion
+        // Match the non-metrics delete_splits pattern: distinguish
+        // "not found" (warn + succeed) from "not deletable" (FailedPrecondition).
         const DELETE_SPLITS_QUERY: &str = r#"
-            DELETE FROM metrics_splits
-            WHERE
-                index_uid = $1
-                AND split_id = ANY($2)
-                AND split_state = 'MarkedForDeletion'
-            RETURNING split_id
+            WITH input_splits AS (
+                SELECT input_splits.split_id, metrics_splits.split_state
+                FROM UNNEST($2::text[]) AS input_splits(split_id)
+                LEFT JOIN metrics_splits
+                    ON metrics_splits.index_uid = $1
+                    AND metrics_splits.split_id = input_splits.split_id
+            ),
+            deleted AS (
+                DELETE FROM metrics_splits
+                USING input_splits
+                WHERE
+                    metrics_splits.index_uid = $1
+                    AND metrics_splits.split_id = input_splits.split_id
+                    AND NOT EXISTS (
+                        SELECT 1 FROM input_splits
+                        WHERE split_state IN ('Staged', 'Published')
+                    )
+                RETURNING metrics_splits.split_id
+            )
+            SELECT
+                (SELECT COUNT(*) FROM input_splits WHERE split_state IS NOT NULL) as num_found,
+                (SELECT COUNT(*) FROM deleted) as num_deleted,
+                COALESCE(
+                    (SELECT ARRAY_AGG(split_id) FROM input_splits
+                     WHERE split_state IN ('Staged', 'Published')),
+                    ARRAY[]::text[]
+                ) as not_deletable,
+                COALESCE(
+                    (SELECT ARRAY_AGG(split_id) FROM input_splits
+                     WHERE split_state IS NULL),
+                    ARRAY[]::text[]
+                ) as not_found
         "#;
 
-        let deleted_split_ids: Vec<String> = sqlx::query_scalar(DELETE_SPLITS_QUERY)
+        let (num_found, num_deleted, not_deletable_ids, not_found_ids): (
+            i64,
+            i64,
+            Vec<String>,
+            Vec<String>,
+        ) = sqlx::query_as(DELETE_SPLITS_QUERY)
             .bind(request.index_uid())
             .bind(&request.split_ids)
-            .fetch_all(&self.connection_pool)
+            .fetch_one(&self.connection_pool)
             .await
             .map_err(|sqlx_error| convert_sqlx_err(&request.index_uid().index_id, sqlx_error))?;
 
-        // Log if some splits were not deleted (either non-existent or not
-        // in MarkedForDeletion state). Delete is idempotent — we don't error
-        // for missing splits.
-        if deleted_split_ids.len() != request.split_ids.len() {
-            let not_deleted: Vec<String> = request
-                .split_ids
-                .iter()
-                .filter(|id| !deleted_split_ids.contains(id))
-                .cloned()
-                .collect();
+        if !not_deletable_ids.is_empty() {
+            let message = format!(
+                "splits `{}` are not deletable",
+                not_deletable_ids.join(", ")
+            );
+            let entity = EntityKind::Splits {
+                split_ids: not_deletable_ids,
+            };
+            return Err(MetastoreError::FailedPrecondition { entity, message });
+        }
 
-            if !not_deleted.is_empty() {
-                warn!(
-                    index_uid = %request.index_uid(),
-                    not_deleted = ?not_deleted,
-                    "some metrics splits were not deleted (non-existent or not marked for deletion)"
-                );
-            }
+        if !not_found_ids.is_empty() {
+            warn!(
+                index_uid = %request.index_uid(),
+                not_found = ?not_found_ids,
+                "{} metrics splits were not found and could not be deleted",
+                not_found_ids.len()
+            );
         }
 
+        let _ = (num_found, num_deleted); // used by the CTE logic
+
         info!(
             index_uid = %request.index_uid(),
-            deleted_count = deleted_split_ids.len(),
+            num_deleted,
             "deleted metrics splits successfully"
         );
         Ok(EmptyResponse {})

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

@@ -160,8 +160,8 @@ pub struct MetricsSplitMetadata {
     /// When this split was created.
     pub created_at: SystemTime,
 
-    /// Parquet file path(s) relative to storage root.
-    pub parquet_files: Vec<String>,
+    /// Parquet file path relative to storage root.
+    pub parquet_file: String,
 
     /// Time window as `[start, start + duration)` in epoch seconds.
     /// None for pre-Phase-31 splits (backward compat).
@@ -175,7 +175,8 @@ pub struct MetricsSplitMetadata {
     /// 0 for newly ingested splits.
     pub num_merge_ops: u32,
 
-    /// RowKeys (sort-key min/max boundaries) as proto bytes.
+    /// RowKeys (sort-key min/max boundaries) as serialized proto bytes
+    /// ([`sortschema::RowKeys`](../../quickwit-proto/protos/event_store_sortschema/event_store_sortschema.proto)).
     /// None for pre-Phase-31 splits or splits without sort schema.
     pub row_keys_proto: Option<Vec<u8>>,
 
@@ -198,7 +199,7 @@ struct MetricsSplitMetadataSerde {
     low_cardinality_tags: HashMap<String, HashSet<String>>,
     high_cardinality_tag_keys: HashSet<String>,
     created_at: SystemTime,
-    parquet_files: Vec<String>,
+    parquet_file: String,
 
     #[serde(default, skip_serializing_if = "Option::is_none")]
     window_start: Option<i64>,
@@ -235,7 +236,7 @@ impl From<MetricsSplitMetadataSerde> for MetricsSplitMetadata {
             low_cardinality_tags: s.low_cardinality_tags,
             high_cardinality_tag_keys: s.high_cardinality_tag_keys,
             created_at: s.created_at,
-            parquet_files: s.parquet_files,
+            parquet_file: s.parquet_file,
             window,
             sort_fields: s.sort_fields,
             num_merge_ops: s.num_merge_ops,
@@ -261,7 +262,7 @@ impl From<MetricsSplitMetadata> for MetricsSplitMetadataSerde {
             low_cardinality_tags: m.low_cardinality_tags,
             high_cardinality_tag_keys: m.high_cardinality_tag_keys,
             created_at: m.created_at,
-            parquet_files: m.parquet_files,
+            parquet_file: m.parquet_file,
             window_start,
             window_duration_secs,
             sort_fields: m.sort_fields,
@@ -351,7 +352,7 @@ pub struct MetricsSplitMetadataBuilder {
     metric_names: HashSet<String>,
     low_cardinality_tags: HashMap<String, HashSet<String>>,
     high_cardinality_tag_keys: HashSet<String>,
-    parquet_files: Vec<String>,
+    parquet_file: String,
     window_start: Option<i64>,
     window_duration_secs: u32,
     sort_fields: String,
@@ -425,8 +426,8 @@ impl MetricsSplitMetadataBuilder {
         self
     }
 
-    pub fn add_parquet_file(mut self, path: impl Into<String>) -> Self {
-        self.parquet_files.push(path.into());
+    pub fn parquet_file(mut self, path: impl Into<String>) -> Self {
+        self.parquet_file = path.into();
         self
     }
 
@@ -500,7 +501,7 @@ impl MetricsSplitMetadataBuilder {
             low_cardinality_tags: self.low_cardinality_tags,
             high_cardinality_tag_keys: self.high_cardinality_tag_keys,
             created_at: SystemTime::now(),
-            parquet_files: self.parquet_files,
+            parquet_file: self.parquet_file,
             window,
             sort_fields: self.sort_fields,
             num_merge_ops: self.num_merge_ops,
@@ -626,7 +627,7 @@ mod tests {
             "low_cardinality_tags": {},
             "high_cardinality_tag_keys": [],
             "created_at": {"secs_since_epoch": 1700000000, "nanos_since_epoch": 0},
-            "parquet_files": ["split1.parquet"]
+            "parquet_file": "split1.parquet"
         }"#;
 
         let metadata: MetricsSplitMetadata =

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

@@ -122,7 +122,7 @@ impl ParquetSplitWriter {
             .size_bytes(0)
             .sort_fields(self.writer.sort_fields_string())
             .window_duration_secs(window_duration)
-            .add_parquet_file(filename);
+            .parquet_file(filename);
 
         if let Some(ws) = window_start_secs {
             builder = builder.window_start_secs(ws);