fix error handling, linter

Author: mattmkim

quickwit/quickwit-index-management/src/parquet_garbage_collection.rs

@@ -22,7 +22,8 @@ use quickwit_common::Progress;
 use quickwit_metastore::{
     ListMetricsSplitsQuery, ListMetricsSplitsRequestExt, ListMetricsSplitsResponseExt,
 };
-use quickwit_parquet_engine::split::{MetricsSplitRecord, MetricsSplitState};
+use quickwit_metastore::SplitState;
+use quickwit_parquet_engine::split::MetricsSplitRecord;
 use quickwit_proto::metastore::{
     DeleteMetricsSplitsRequest, ListMetricsSplitsRequest, MarkMetricsSplitsForDeletionRequest,
     MetastoreService, MetastoreServiceClient,
@@ -90,12 +91,9 @@ pub async fn run_parquet_garbage_collect(
 
     let mut deletable_staged_splits = Vec::new();
     for index_uid in indexes.keys() {
-        match list_stale_staged_splits(&metastore, index_uid, staged_cutoff, progress_opt).await {
-            Ok(splits) => deletable_staged_splits.extend(splits),
-            Err(err) => {
-                error!(index_uid=%index_uid, error=?err, "failed to list stale staged parquet splits");
-            }
-        }
+        let splits =
+            list_stale_staged_splits(&metastore, index_uid, staged_cutoff, progress_opt).await?;
+        deletable_staged_splits.extend(splits);
     }
 
     if dry_run {
@@ -104,12 +102,9 @@ pub async fn run_parquet_garbage_collect(
 
         let mut splits_marked_for_deletion = Vec::new();
         for index_uid in indexes.keys() {
-            match list_marked_splits(&metastore, index_uid, deletion_cutoff, progress_opt).await {
-                Ok(splits) => splits_marked_for_deletion.extend(splits),
-                Err(err) => {
-                    error!(index_uid=%index_uid, error=?err, "failed to list marked parquet splits");
-                }
-            }
+            let splits =
+                list_marked_splits(&metastore, index_uid, deletion_cutoff, progress_opt).await?;
+            splits_marked_for_deletion.extend(splits);
         }
         splits_marked_for_deletion.extend(deletable_staged_splits);
 
@@ -127,38 +122,27 @@ pub async fn run_parquet_garbage_collect(
     }
 
     // Schedule all eligible staged splits for delete
-    if let Err(err) =
-        mark_splits_for_deletion(&metastore, &deletable_staged_splits, progress_opt).await
-    {
-        error!(error=?err, "failed to mark stale staged parquet splits");
-    }
+    mark_splits_for_deletion(&metastore, &deletable_staged_splits, progress_opt).await?;
 
     // Phase 2: Delete splits marked for deletion past the grace period
     let deletion_cutoff =
         OffsetDateTime::now_utc().unix_timestamp() - deletion_grace_period.as_secs() as i64;
 
     for (index_uid, storage) in &indexes {
-        match delete_marked_parquet_splits(
+        let batch_info = delete_marked_parquet_splits(
             &metastore,
             index_uid,
             storage.clone(),
             deletion_cutoff,
             progress_opt,
         )
-        .await
-        {
-            Ok(batch_info) => {
-                removal_info
-                    .removed_parquet_splits_entries
-                    .extend(batch_info.removed_parquet_splits_entries);
-                removal_info
-                    .failed_parquet_splits
-                    .extend(batch_info.failed_parquet_splits);
-            }
-            Err(err) => {
-                error!(index_uid=%index_uid, error=?err, "failed to delete marked parquet splits");
-            }
-        }
+        .await?;
+        removal_info
+            .removed_parquet_splits_entries
+            .extend(batch_info.removed_parquet_splits_entries);
+        removal_info
+            .failed_parquet_splits
+            .extend(batch_info.failed_parquet_splits);
     }
 
     metrics.record(
@@ -178,7 +162,7 @@ async fn list_stale_staged_splits(
     progress_opt: Option<&Progress>,
 ) -> anyhow::Result<Vec<MetricsSplitRecord>> {
     let query = ListMetricsSplitsQuery::for_index(index_uid.clone())
-        .with_split_states(vec![MetricsSplitState::Staged.to_string()])
+        .with_split_states(vec![SplitState::Staged])
         .with_update_timestamp_lte(staged_cutoff);
 
     let request = ListMetricsSplitsRequest::try_from_query(index_uid.clone(), &query)
@@ -198,7 +182,7 @@ async fn list_marked_splits(
     progress_opt: Option<&Progress>,
 ) -> anyhow::Result<Vec<MetricsSplitRecord>> {
     let query = ListMetricsSplitsQuery::for_index(index_uid.clone())
-        .with_split_states(vec![MetricsSplitState::MarkedForDeletion.to_string()])
+        .with_split_states(vec![SplitState::MarkedForDeletion])
         .with_update_timestamp_lte(deletion_cutoff);
 
     let request = ListMetricsSplitsRequest::try_from_query(index_uid.clone(), &query)
@@ -258,7 +242,7 @@ async fn delete_marked_parquet_splits(
     let mut removal_info = ParquetSplitRemovalInfo::default();
 
     let mut query = ListMetricsSplitsQuery::for_index(index_uid.clone())
-        .with_split_states(vec![MetricsSplitState::MarkedForDeletion.to_string()])
+        .with_split_states(vec![SplitState::MarkedForDeletion])
         .with_update_timestamp_lte(deletion_cutoff)
         .with_limit(DELETE_PARQUET_SPLITS_BATCH_SIZE);
 

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

@@ -113,6 +113,8 @@ impl Default for ListMetricsSplitsQuery {
             create_timestamp: Default::default(),
             mature: Bound::Unbounded,
             limit: None,
+            after_split_id: None,
+            max_time_range_end: None,
         }
     }
 }