fix: Janitor list splits query Closes quickwit-oss#5848 (quickwit-oss#5850)

Author: Darkheir

quickwit/quickwit-janitor/src/retention_policy_execution.rs

@@ -48,7 +48,7 @@ pub async fn run_execute_retention_policy(
     let max_retention_timestamp = current_timestamp - retention_period.as_secs() as i64;
     let query = ListSplitsQuery::for_index(index_uid.clone())
         .with_split_state(SplitState::Published)
-        .with_time_range_end_lte(max_retention_timestamp);
+        .with_max_time_range_end(max_retention_timestamp);
 
     let list_splits_request = ListSplitsRequest::try_from_list_splits_query(&query)?;
     let (expired_splits, ignored_splits): (Vec<SplitMetadata>, Vec<SplitMetadata>) = ctx

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

@@ -735,6 +735,11 @@ fn split_query_predicate(split: &&Split, query: &ListSplitsQuery) -> bool {
         if !query.time_range.overlaps_with(range.clone()) {
             return false;
         }
+        if let Some(v) = query.max_time_range_end {
+            if range.end() > &v {
+                return false;
+            }
+        }
     }
 
     if let Some(node_id) = &query.node_id {
@@ -889,6 +894,12 @@ mod tests {
         assert!(split_query_predicate(&&split_1, &query));
         assert!(!split_query_predicate(&&split_2, &query));
         assert!(!split_query_predicate(&&split_3, &query));
+
+        let query = ListSplitsQuery::for_index(IndexUid::new_with_random_ulid("test-index"))
+            .with_max_time_range_end(50);
+        assert!(split_query_predicate(&&split_1, &query));
+        assert!(split_query_predicate(&&split_2, &query));
+        assert!(!split_query_predicate(&&split_3, &query));
     }
 
     #[test]

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

@@ -655,6 +655,9 @@ pub struct ListSplitsQuery {
     /// The time range to filter by.
     pub time_range: FilterRange<i64>,
 
+    /// The maximum time range end to filter by.
+    pub max_time_range_end: Option<i64>,
+
     /// The delete opstamp range to filter by.
     pub delete_opstamp: FilterRange<u64>,
 
@@ -721,6 +724,7 @@ impl ListSplitsQuery {
             split_states: Vec::new(),
             tags: None,
             time_range: Default::default(),
+            max_time_range_end: None,
             delete_opstamp: Default::default(),
             update_timestamp: Default::default(),
             create_timestamp: Default::default(),
@@ -744,6 +748,7 @@ impl ListSplitsQuery {
             split_states: Vec::new(),
             tags: None,
             time_range: Default::default(),
+            max_time_range_end: None,
             delete_opstamp: Default::default(),
             update_timestamp: Default::default(),
             create_timestamp: Default::default(),
@@ -763,6 +768,7 @@ impl ListSplitsQuery {
             split_states: Vec::new(),
             tags: None,
             time_range: Default::default(),
+            max_time_range_end: None,
             delete_opstamp: Default::default(),
             update_timestamp: Default::default(),
             create_timestamp: Default::default(),
@@ -836,6 +842,13 @@ impl ListSplitsQuery {
         self
     }
 
+    /// Retains only splits with a time range end that is
+    /// *less than or equal to* the provided value.
+    pub fn with_max_time_range_end(mut self, v: i64) -> Self {
+        self.max_time_range_end = Some(v);
+        self
+    }
+
     /// Sets the field's lower bound to match values that are
     /// *less than or equal to* the provided value.
     pub fn with_delete_opstamp_lte(mut self, v: u64) -> Self {

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

@@ -2110,6 +2110,17 @@ mod tests {
             sql.to_string(PostgresQueryBuilder),
             r#"SELECT * FROM "splits" WHERE "split_state" IN ('Staged')"#
         );
+
+        let mut select_statement = Query::select();
+        let sql = select_statement.column(Asterisk).from(Splits::Table);
+
+        let query = ListSplitsQuery::for_all_indexes().with_max_time_range_end(42);
+        append_query_filters_and_order_by(sql, &query);
+
+        assert_eq!(
+            sql.to_string(PostgresQueryBuilder),
+            r#"SELECT * FROM "splits" WHERE "time_range_end" <= 42"#
+        );
     }
 
     #[test]

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

@@ -122,6 +122,10 @@ pub(super) fn append_query_filters_and_order_by(
         sql.cond_where(generate_sql_condition(tags));
     };
 
+    if let Some(v) = query.max_time_range_end {
+        sql.cond_where(Expr::col(Splits::TimeRangeEnd).lte(v));
+    }
+
     match query.time_range.start {
         Bound::Included(v) => {
             sql.cond_where(any![