appease linter

mattmkim · mattmkim · commit 3bbfb7ff2626 · 2026-04-14T16:27:10.000-07:00
diff --git a/quickwit/Cargo.lock b/quickwit/Cargo.lock
diff --git a/quickwit/quickwit-control-plane/src/indexing_scheduler/mod.rs b/quickwit/quickwit-control-plane/src/indexing_scheduler/mod.rs
@@ -23,7 +23,6 @@ use std::time::{Duration, Instant};
 
 use fnv::{FnvHashMap, FnvHashSet};
 use itertools::Itertools;
-use once_cell::sync::OnceCell;
 use quickwit_common::is_parquet_pipeline_index;
 use quickwit_common::pretty::PrettySample;
 use quickwit_config::{FileSourceParams, SourceParams, indexing_pipeline_params_fingerprint};
diff --git a/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs b/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs
@@ -658,8 +658,7 @@ impl IndexingPipeline {
             self.params.indexing_directory.path(),
             &table_config,
         );
-        let parquet_packager =
-            ParquetPackager::new(split_writer, parquet_uploader_mailbox);
+        let parquet_packager = ParquetPackager::new(split_writer, parquet_uploader_mailbox);
         let (parquet_packager_mailbox, parquet_packager_handle) = ctx
             .spawn_actor()
             .set_kill_switch(self.kill_switch.clone())
diff --git a/quickwit/quickwit-indexing/src/actors/parquet_doc_processor.rs b/quickwit/quickwit-indexing/src/actors/parquet_doc_processor.rs
@@ -570,7 +570,12 @@ mod tests {
         // Create ParquetPackager
         let writer_config = ParquetWriterConfig::default();
         let table_config = quickwit_parquet_engine::table_config::TableConfig::default();
-        let split_writer = ParquetSplitWriter::new(quickwit_parquet_engine::split::ParquetSplitKind::Metrics, writer_config, temp_dir.path(), &table_config);
+        let split_writer = ParquetSplitWriter::new(
+            quickwit_parquet_engine::split::ParquetSplitKind::Metrics,
+            writer_config,
+            temp_dir.path(),
+            &table_config,
+        );
         let packager = ParquetPackager::new(split_writer, uploader_mailbox);
         let (packager_mailbox, packager_handle) = universe.spawn_builder().spawn(packager);
 
diff --git a/quickwit/quickwit-indexing/src/actors/parquet_e2e_test.rs b/quickwit/quickwit-indexing/src/actors/parquet_e2e_test.rs
@@ -175,7 +175,12 @@ async fn test_metrics_pipeline_e2e() {
 
     let writer_config = ParquetWriterConfig::default();
     let table_config = quickwit_parquet_engine::table_config::TableConfig::default();
-    let split_writer = ParquetSplitWriter::new(ParquetSplitKind::Metrics, writer_config, temp_dir.path(), &table_config);
+    let split_writer = ParquetSplitWriter::new(
+        ParquetSplitKind::Metrics,
+        writer_config,
+        temp_dir.path(),
+        &table_config,
+    );
     let packager = ParquetPackager::new(split_writer, uploader_mailbox);
     let (packager_mailbox, packager_handle) = universe.spawn_builder().spawn(packager);
 
@@ -267,7 +272,7 @@ async fn test_file_backed_metastore_metrics_operations() {
     use quickwit_config::IndexConfig;
     use quickwit_metastore::{
         CreateIndexRequestExt, FileBackedMetastore, ListParquetSplitsQuery,
-        ListParquetSplitsRequestExt, ListParquetSplitsResponseExt, ParquetSplitRecord,
+        ListParquetSplitsRequestExt, ListParquetSplitsResponseExt, ParquetSplitRecord, SplitState,
         StageParquetSplitsRequestExt,
     };
     use quickwit_parquet_engine::split::{ParquetSplitMetadata, TimeRange};
@@ -314,7 +319,7 @@ async fn test_file_backed_metastore_metrics_operations() {
     // Verify staged
     let query = ListParquetSplitsQuery::for_index(index_uid.clone())
         .with_split_states([SplitState::Staged]);
-    let list_request = ListParquetSplitsQuery::try_from_query(index_uid.clone(), &query).unwrap();
+    let list_request = ListMetricsSplitsRequest::try_from_query(index_uid.clone(), &query).unwrap();
     let list_response = metastore.list_metrics_splits(list_request).await.unwrap();
     let staged: Vec<ParquetSplitRecord> = list_response.deserialize_splits().unwrap();
     assert_eq!(staged.len(), 2);
@@ -511,7 +516,7 @@ async fn test_sketch_pipeline_e2e() {
         ParquetSplitKind::Sketches,
         writer_config,
         temp_dir.path(),
-        &TableConfig::default(),
+        &quickwit_parquet_engine::table_config::TableConfig::default(),
     );
     let packager = ParquetPackager::new(split_writer, uploader_mailbox);
     let (packager_mailbox, packager_handle) = universe.spawn_builder().spawn(packager);
diff --git a/quickwit/quickwit-indexing/src/actors/parquet_indexer.rs b/quickwit/quickwit-indexing/src/actors/parquet_indexer.rs
@@ -608,7 +608,12 @@ mod tests {
     ) -> (Mailbox<ParquetPackager>, ActorHandle<ParquetPackager>) {
         let writer_config = ParquetWriterConfig::default();
         let table_config = quickwit_parquet_engine::table_config::TableConfig::default();
-        let split_writer = ParquetSplitWriter::new(quickwit_parquet_engine::split::ParquetSplitKind::Metrics, writer_config, temp_dir, &table_config);
+        let split_writer = ParquetSplitWriter::new(
+            quickwit_parquet_engine::split::ParquetSplitKind::Metrics,
+            writer_config,
+            temp_dir,
+            &table_config,
+        );
 
         let packager = ParquetPackager::new(split_writer, uploader_mailbox);
         universe.spawn_builder().spawn(packager)
diff --git a/quickwit/quickwit-indexing/src/actors/parquet_packager.rs b/quickwit/quickwit-indexing/src/actors/parquet_packager.rs
@@ -275,7 +275,12 @@ mod tests {
     ) -> (Mailbox<ParquetPackager>, ActorHandle<ParquetPackager>) {
         let writer_config = ParquetWriterConfig::default();
         let table_config = quickwit_parquet_engine::table_config::TableConfig::default();
-        let split_writer = ParquetSplitWriter::new(quickwit_parquet_engine::split::ParquetSplitKind::Metrics, writer_config, temp_dir, &table_config);
+        let split_writer = ParquetSplitWriter::new(
+            quickwit_parquet_engine::split::ParquetSplitKind::Metrics,
+            writer_config,
+            temp_dir,
+            &table_config,
+        );
 
         let packager = ParquetPackager::new(split_writer, uploader_mailbox);
         universe.spawn_builder().spawn(packager)
diff --git a/quickwit/quickwit-indexing/src/actors/parquet_uploader.rs b/quickwit/quickwit-indexing/src/actors/parquet_uploader.rs
@@ -391,16 +391,16 @@ mod tests {
     use crate::actors::Sequencer;
     use crate::models::PublishLock;
 
-    fn create_test_metrics_split(index_uid: &str, split_id: &str) -> ParquetSplit {
-        let metadata = ParquetSplitMetadata::metrics_builder()
+    fn create_test_metrics_split(index_uid: &str, split_id: &str) -> ParquetSplitMetadata {
+        ParquetSplitMetadata::metrics_builder()
             .index_uid(index_uid)
             .split_id(quickwit_parquet_engine::split::ParquetSplitId::new(
                 split_id,
             ))
             .time_range(TimeRange::new(1000, 2000))
             .num_rows(100)
             .size_bytes(1024)
-            .build();
+            .build()
     }
 
     /// Create placeholder parquet files in the temp directory for testing.
diff --git a/quickwit/quickwit-metastore/src/lib.rs b/quickwit/quickwit-metastore/src/lib.rs
@@ -47,8 +47,8 @@ pub use metastore::{
     IndexMetadataResponseExt, IndexesMetadataResponseExt, ListIndexesMetadataResponseExt,
     ListParquetSplitsQuery, ListParquetSplitsRequestExt, ListParquetSplitsResponseExt,
     ListSplitsQuery, ListSplitsRequestExt, ListSplitsResponseExt, MetastoreServiceExt,
-    MetastoreServiceStreamSplitsExt, PublishParquetSplitsRequestExt, PublishSplitsRequestExt,
-    ParquetSplitRecord, StageParquetSplitsRequestExt, StageSplitsRequestExt,
+    MetastoreServiceStreamSplitsExt, ParquetSplitRecord, PublishParquetSplitsRequestExt,
+    PublishSplitsRequestExt, StageParquetSplitsRequestExt, StageSplitsRequestExt,
     UpdateIndexRequestExt, UpdateSourceRequestExt, file_backed,
 };
 pub use metastore_factory::{MetastoreFactory, UnsupportedMetastore};
diff --git a/quickwit/quickwit-metastore/src/metastore/file_backed/file_backed_index/mod.rs b/quickwit/quickwit-metastore/src/metastore/file_backed/file_backed_index/mod.rs
@@ -29,8 +29,6 @@ use quickwit_config::{
     DocMapping, IndexingSettings, IngestSettings, RetentionPolicy, SearchSettings, SourceConfig,
 };
 use quickwit_parquet_engine::split::ParquetSplitMetadata;
-
-use crate::SplitState;
 use quickwit_proto::metastore::{
     AcquireShardsRequest, AcquireShardsResponse, DeleteQuery, DeleteShardsRequest,
     DeleteShardsResponse, DeleteTask, EntityKind, IndexStats, ListShardsSubrequest,
diff --git a/quickwit/quickwit-metastore/src/metastore/mod.rs b/quickwit/quickwit-metastore/src/metastore/mod.rs
@@ -752,8 +752,6 @@ impl ListSplitsResponseExt for ListSplitsResponse {
 
 use quickwit_parquet_engine::split::ParquetSplitMetadata;
 
-use crate::SplitState;
-
 /// A complete parquet split record from the database.
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct ParquetSplitRecord {
diff --git a/quickwit/quickwit-metastore/src/metastore/postgres/metastore.rs b/quickwit/quickwit-metastore/src/metastore/postgres/metastore.rs
@@ -27,6 +27,7 @@ use quickwit_common::{ServiceStream, get_bool_from_env, rate_limited_error};
 use quickwit_config::{
     IndexTemplate, IndexTemplateId, PostgresMetastoreConfig, validate_index_id_pattern,
 };
+use quickwit_parquet_engine::split::{ParquetSplitKind, ParquetSplitMetadata};
 use quickwit_proto::ingest::{Shard, ShardState};
 use quickwit_proto::metastore::{
     AcquireShardsRequest, AcquireShardsResponse, AddSourceRequest, CreateIndexRequest,
@@ -35,10 +36,9 @@ use quickwit_proto::metastore::{
     DeleteShardsResponse, DeleteSketchSplitsRequest, DeleteSourceRequest, DeleteSplitsRequest,
     DeleteTask, EmptyResponse, EntityKind, FindIndexTemplateMatchesRequest,
     FindIndexTemplateMatchesResponse, GetClusterIdentityRequest, GetClusterIdentityResponse,
-    GetIndexRoutingTableRequest, GetIndexRoutingTableResponse, GetIndexTemplateRequest,
-    GetIndexTemplateResponse, IndexMetadataFailure, IndexMetadataFailureReason,
-    IndexMetadataRequest, IndexMetadataResponse, IndexStats, IndexTemplateMatch,
-    IndexesMetadataRequest, IndexesMetadataResponse, LastDeleteOpstampRequest,
+    GetIndexTemplateRequest, GetIndexTemplateResponse, IndexMetadataFailure,
+    IndexMetadataFailureReason, IndexMetadataRequest, IndexMetadataResponse, IndexStats,
+    IndexTemplateMatch, IndexesMetadataRequest, IndexesMetadataResponse, LastDeleteOpstampRequest,
     LastDeleteOpstampResponse, ListDeleteTasksRequest, ListDeleteTasksResponse,
     ListIndexStatsRequest, ListIndexStatsResponse, ListIndexTemplatesRequest,
     ListIndexTemplatesResponse, ListIndexesMetadataRequest, ListIndexesMetadataResponse,
@@ -49,10 +49,9 @@ use quickwit_proto::metastore::{
     MetastoreResult, MetastoreService, MetastoreServiceStream, OpenShardSubrequest,
     OpenShardSubresponse, OpenShardsRequest, OpenShardsResponse, PruneShardsRequest,
     PublishMetricsSplitsRequest, PublishSketchSplitsRequest, PublishSplitsRequest,
-    ResetSourceCheckpointRequest, SetIndexRoutingTableRequest, SplitStats,
-    StageMetricsSplitsRequest, StageSketchSplitsRequest, StageSplitsRequest, ToggleSourceRequest,
-    UpdateIndexRequest, UpdateSourceRequest, UpdateSplitsDeleteOpstampRequest,
-    UpdateSplitsDeleteOpstampResponse, serde_utils,
+    ResetSourceCheckpointRequest, SplitStats, StageMetricsSplitsRequest, StageSketchSplitsRequest,
+    StageSplitsRequest, ToggleSourceRequest, UpdateIndexRequest, UpdateSourceRequest,
+    UpdateSplitsDeleteOpstampRequest, UpdateSplitsDeleteOpstampResponse, serde_utils,
 };
 use quickwit_proto::types::{IndexId, IndexUid, Position, PublishToken, ShardId, SourceId};
 use sea_query::{Alias, Asterisk, Expr, Func, PostgresQueryBuilder, Query, UnionType};
@@ -62,8 +61,6 @@ use time::OffsetDateTime;
 use tracing::{debug, info, instrument, warn};
 use uuid::Uuid;
 
-use quickwit_parquet_engine::split::{ParquetSplitKind, ParquetSplitMetadata};
-
 use super::error::convert_sqlx_err;
 use super::migrator::run_migrations;
 use super::model::{PgDeleteTask, PgIndex, PgIndexTemplate, PgShard, PgSplit, Splits};
@@ -375,7 +372,6 @@ where
     Ok(index_metadata)
 }
 
-
 #[async_trait]
 impl MetastoreService for PostgresqlMetastore {
     async fn check_connectivity(&self) -> anyhow::Result<()> {
@@ -1949,12 +1945,11 @@ impl PostgresqlMetastore {
         let mut zonemap_regexes_json_list: Vec<String> = Vec::with_capacity(splits_metadata.len());
 
         for metadata in &splits_metadata {
-            let insertable =
-                InsertableParquetSplit::from_metadata(metadata, SplitState::Staged)
-                    .map_err(|err| MetastoreError::JsonSerializeError {
-                        struct_name: "ParquetSplitMetadata".to_string(),
-                        message: err.to_string(),
-                    })?;
+            let insertable = InsertableParquetSplit::from_metadata(metadata, SplitState::Staged)
+                .map_err(|err| MetastoreError::JsonSerializeError {
+                    struct_name: "ParquetSplitMetadata".to_string(),
+                    message: err.to_string(),
+                })?;
 
             split_ids.push(insertable.split_id);
             split_states.push(insertable.split_state);
diff --git a/quickwit/quickwit-metastore/src/metastore/postgres/parquet_model.rs b/quickwit/quickwit-metastore/src/metastore/postgres/parquet_model.rs
@@ -19,8 +19,6 @@
 
 use std::collections::{HashMap, HashSet};
 
-use sea_query::Iden;
-
 use quickwit_parquet_engine::split::{
     ParquetSplitMetadata, TAG_DATACENTER, TAG_ENV, TAG_HOST, TAG_REGION, TAG_SERVICE,
 };
@@ -134,42 +132,31 @@ impl PgParquetSplit {
         debug_assert_eq!(metadata.time_range.start_secs, self.time_range_start as u64);
         debug_assert_eq!(metadata.time_range.end_secs, self.time_range_end as u64);
 
-        // SS-5 (SortSchema.tla): sort_fields must be identical in JSON metadata
-        // and the dedicated SQL column. Inconsistency would cause the compaction
-        // planner to select wrong splits or miss eligible ones.
-        quickwit_dst::check_invariant!(
-            quickwit_dst::invariants::InvariantId::SS5,
-            metadata.sort_fields == self.sort_fields,
-            ": sort_fields mismatch between JSON ('{}') and SQL column ('{}')",
-            metadata.sort_fields,
-            self.sort_fields
+        // SS-5: sort_fields must be identical in JSON metadata and the dedicated
+        // SQL column. Inconsistency would cause the compaction planner to select
+        // wrong splits or miss eligible ones.
+        debug_assert_eq!(
+            metadata.sort_fields, self.sort_fields,
+            "SS-5: sort_fields mismatch between JSON ('{}') and SQL column ('{}')",
+            metadata.sort_fields, self.sort_fields
         );
-
-        // SS-5 continued: window_start must match between JSON and SQL column.
-        quickwit_dst::check_invariant!(
-            quickwit_dst::invariants::InvariantId::SS5,
-            metadata.window_start() == self.window_start,
-            ": window_start mismatch between JSON ({:?}) and SQL column ({:?})",
+        debug_assert_eq!(
+            metadata.window_start(),
+            self.window_start,
+            "SS-5: window_start mismatch between JSON ({:?}) and SQL column ({:?})",
             metadata.window_start(),
             self.window_start
         );
-
-        // SS-5 continued: window_duration_secs must match.
-        quickwit_dst::check_invariant!(
-            quickwit_dst::invariants::InvariantId::SS5,
-            metadata.window_duration_secs() == self.window_duration_secs.unwrap_or(0) as u32,
-            ": window_duration_secs mismatch between JSON ({}) and SQL column ({:?})",
+        debug_assert_eq!(
+            metadata.window_duration_secs(),
+            self.window_duration_secs.unwrap_or(0) as u32,
+            "SS-5: window_duration_secs mismatch between JSON ({}) and SQL column ({:?})",
             metadata.window_duration_secs(),
             self.window_duration_secs
         );
-
-        // SS-4 (SortSchema.tla): sort_fields is immutable after write.
-        // We can't verify immutability at read time (no history available), but
-        // we verify the row_keys_proto round-trips consistently.
-        quickwit_dst::check_invariant!(
-            quickwit_dst::invariants::InvariantId::SS5,
-            metadata.row_keys_proto == self.row_keys,
-            ": row_keys_proto mismatch between JSON and SQL column"
+        debug_assert_eq!(
+            metadata.row_keys_proto, self.row_keys,
+            "SS-5: row_keys_proto mismatch between JSON and SQL column"
         );
 
         Ok(metadata)
@@ -209,8 +196,9 @@ impl TryFrom<PgParquetSplit> for ParquetSplitRecord {
 
 #[cfg(test)]
 mod tests {
+    use quickwit_parquet_engine::split::{ParquetSplitId, TimeRange};
+
     use super::*;
-    use quickwit_parquet_engine::split::{ParquetSplitId, ParquetSplitKind, TimeRange};
 
     #[test]
     fn test_insertable_from_metadata() {
@@ -265,9 +253,8 @@ mod tests {
             .add_zonemap_regex("metric_name", "cpu\\..*")
             .build();
 
-        let insertable =
-            InsertableParquetSplit::from_metadata(&metadata, SplitState::Published)
-                .expect("conversion should succeed");
+        let insertable = InsertableParquetSplit::from_metadata(&metadata, SplitState::Published)
+            .expect("conversion should succeed");
 
         assert_eq!(insertable.window_start, Some(1700000000));
         assert_eq!(insertable.window_duration_secs, 3600);
@@ -289,9 +276,8 @@ mod tests {
             .time_range(TimeRange::new(1000, 2000))
             .build();
 
-        let insertable =
-            InsertableParquetSplit::from_metadata(&metadata, SplitState::Staged)
-                .expect("conversion should succeed");
+        let insertable = InsertableParquetSplit::from_metadata(&metadata, SplitState::Staged)
+            .expect("conversion should succeed");
 
         assert!(insertable.window_start.is_none());
         assert_eq!(
diff --git a/quickwit/quickwit-metastore/src/tests/metrics.rs b/quickwit/quickwit-metastore/src/tests/metrics.rs
@@ -14,14 +14,13 @@
 
 use quickwit_common::rand::append_random_suffix;
 use quickwit_config::IndexConfig;
-use quickwit_parquet_engine::split::{ParquetSplitMetadata, ParquetSplitId, TimeRange};
+use quickwit_parquet_engine::split::{ParquetSplitId, ParquetSplitMetadata, TimeRange};
 use quickwit_proto::metastore::{
     CreateIndexRequest, DeleteMetricsSplitsRequest, EntityKind, ListMetricsSplitsRequest,
     MarkMetricsSplitsForDeletionRequest, MetastoreError, PublishMetricsSplitsRequest,
     StageMetricsSplitsRequest,
 };
 use quickwit_proto::types::IndexUid;
-use regex::Split;
 
 use super::DefaultForTest;
 use crate::tests::cleanup_index;
@@ -118,7 +117,7 @@ pub async fn test_metastore_stage_metrics_splits_upsert<
 
     // Stage a split with 100 rows.
     let split_v1 = ParquetSplitMetadata::metrics_builder()
-        .split_id(SplitId::new(&split_id))
+        .split_id(ParquetSplitId::new(&split_id))
         .index_uid(index_uid.to_string())
         .time_range(TimeRange::new(1000, 2000))
         .num_rows(100)
@@ -443,7 +442,7 @@ pub async fn test_metastore_publish_metrics_splits<
     assert_eq!(splits.len(), 2);
 
     for split in &splits {
-        assert_eq!(split.state, Split::Published);
+        assert_eq!(split.state, SplitState::Published);
     }
 
     // Verify no Staged splits remain.
@@ -586,7 +585,7 @@ pub async fn test_metastore_delete_metrics_splits<
         .unwrap();
 
     // Verify it is gone (list all states).
-    let query = ListMetricsSplitsQuery::for_index(index_uid.clone()).with_split_states([
+    let query = ListParquetSplitsQuery::for_index(index_uid.clone()).with_split_states([
         SplitState::Staged,
         SplitState::Published,
         SplitState::MarkedForDeletion,
diff --git a/quickwit/quickwit-metastore/src/tests/mod.rs b/quickwit/quickwit-metastore/src/tests/mod.rs
@@ -25,7 +25,7 @@ use quickwit_proto::metastore::{
 use quickwit_proto::tonic::transport::Channel;
 use quickwit_proto::types::IndexUid;
 
-use crate::metastore::{ListMetricsSplitsRequestExt, ListMetricsSplitsResponseExt};
+use crate::metastore::{ListParquetSplitsRequestExt, ListParquetSplitsResponseExt};
 
 pub(crate) mod delete_task;
 pub(crate) mod get_identity;
@@ -157,7 +157,7 @@ async fn cleanup_index(metastore: &mut dyn MetastoreServiceExt, index_uid: Index
             .unwrap();
     }
     // Also clean up any metrics splits (they have a separate FK constraint).
-    let metrics_query = crate::metastore::ListMetricsSplitsQuery::for_index(index_uid.clone())
+    let metrics_query = crate::metastore::ListParquetSplitsQuery::for_index(index_uid.clone())
         .with_split_states([
             SplitState::Staged,
             SplitState::Published,
diff --git a/quickwit/quickwit-parquet-engine/Cargo.toml b/quickwit/quickwit-parquet-engine/Cargo.toml
diff --git a/quickwit/quickwit-parquet-engine/src/ingest/sketch_processor.rs b/quickwit/quickwit-parquet-engine/src/ingest/sketch_processor.rs
diff --git a/quickwit/quickwit-parquet-engine/src/schema/mod.rs b/quickwit/quickwit-parquet-engine/src/schema/mod.rs
diff --git a/quickwit/quickwit-parquet-engine/src/split/metadata.rs b/quickwit/quickwit-parquet-engine/src/split/metadata.rs
diff --git a/quickwit/quickwit-parquet-engine/src/storage/split_writer.rs b/quickwit/quickwit-parquet-engine/src/storage/split_writer.rs
diff --git a/quickwit/quickwit-parquet-engine/src/storage/writer.rs b/quickwit/quickwit-parquet-engine/src/storage/writer.rs
