From 4f6f5ccdd717f7751565a7cc694381450e5a4572 Mon Sep 17 00:00:00 2001 From: Remi Dettai Date: Mon, 12 Oct 2026 10:39:46 +0200 Subject: [PATCH] Fix retention to use secondary --- .../src/actors/retention_policy_executor.rs | 148 ++++++- .../src/retention_policy_execution.rs | 413 +++++++++++++++++- .../src/tests/list_splits.rs | 71 +++ quickwit/quickwit-metastore/src/tests/mod.rs | 7 + 4 files changed, 620 insertions(+), 19 deletions(-) diff --git a/quickwit/quickwit-janitor/src/actors/retention_policy_executor.rs b/quickwit/quickwit-janitor/src/actors/retention_policy_executor.rs index 6e43ad5def3..609f93b00f4 100644 --- a/quickwit/quickwit-janitor/src/actors/retention_policy_executor.rs +++ b/quickwit/quickwit-janitor/src/actors/retention_policy_executor.rs @@ -27,7 +27,7 @@ use quickwit_proto::types::IndexUid; use serde::Serialize; use tracing::{debug, error, info}; -use crate::retention_policy_execution::run_execute_retention_policy; +use crate::retention_policy_execution::{RetentionTimestampMode, run_execute_retention_policy}; const RUN_INTERVAL: Duration = Duration::from_secs(60 * 60); // 1 hours @@ -60,6 +60,7 @@ pub struct RetentionPolicyExecutor { /// This act as local cache that is periodically updated while taking into /// account deleted indexes, updated or removed retention policy on indexes. index_configs: HashMap, + timestamp_mode: RetentionTimestampMode, counters: RetentionPolicyExecutorCounters, } @@ -68,6 +69,7 @@ impl RetentionPolicyExecutor { Self { metastore, index_configs: HashMap::new(), + timestamp_mode: RetentionTimestampMode::from_env(), counters: RetentionPolicyExecutorCounters::default(), } } @@ -211,6 +213,11 @@ impl Handler for RetentionPolicyExecutor { message.index_uid.clone(), self.metastore.clone(), retention_policy, + self.timestamp_mode, + index_config + .doc_mapping + .secondary_timestamp_field + .as_deref(), ctx, ) .await; @@ -501,4 +508,143 @@ mod tests { Ok(()) } + + fn make_split_with_secondary( + split_id: &str, + time_range: RangeInclusive, + secondary_time_range: RangeInclusive, + ) -> Split { + let mut split = make_split(split_id, Some(time_range)); + split.split_metadata.secondary_time_range = Some(secondary_time_range); + split + } + + /// Returns `primary-only` for primary queries and `secondary-only` for secondary + /// queries, plus `both` for either. Secondary queries are rejected when the index has no + /// secondary timestamp field. + fn mock_metastore_for_timestamp_mode( + secondary_timestamp_field_opt: Option<&'static str>, + expected_marked_split_ids: &'static [&'static str], + ) -> MockMetastoreService { + let mut mock_metastore = MockMetastoreService::new(); + mock_metastore + .expect_list_indexes_metadata() + .times(..) + .returning(move |_list_indexes_request| { + let mut index_config = make_index("index-1", Some("1 hour")); + index_config.doc_mapping.secondary_timestamp_field = + secondary_timestamp_field_opt.map(str::to_string); + let indexes_metadata = vec![IndexMetadata::new(index_config)]; + Ok(ListIndexesMetadataResponse::for_test(indexes_metadata)) + }); + mock_metastore + .expect_list_splits() + .times(..) + .returning(move |list_splits_request| { + let query = list_splits_request.deserialize_list_splits_query().unwrap(); + assert_eq!(query.split_states, &[SplitState::Published]); + let splits = match (query.max_time_range_end, query.max_secondary_time_range_end) { + (Some(_), None) => vec![ + make_split_with_secondary("both", 1000..=2000, 1000..=2000), + make_split_with_secondary("primary-only", 1000..=2000, i64::MAX..=i64::MAX), + ], + (None, Some(_)) => { + assert!( + secondary_timestamp_field_opt.is_some(), + "secondary filter used without a secondary timestamp field" + ); + vec![ + make_split_with_secondary("both", 1000..=2000, 1000..=2000), + make_split_with_secondary( + "secondary-only", + i64::MAX..=i64::MAX, + 1000..=2000, + ), + ] + } + filters => panic!("unexpected retention filters: {filters:?}"), + }; + let splits_response = ListSplitsResponse::try_from_splits(splits).unwrap(); + Ok(ServiceStream::from(vec![Ok(splits_response)])) + }); + mock_metastore + .expect_mark_splits_for_deletion() + .times(1..) + .returning(move |mark_splits_for_deletion_request| { + assert_eq!( + mark_splits_for_deletion_request.split_ids, + expected_marked_split_ids + ); + Ok(EmptyResponse {}) + }); + mock_metastore + } + + async fn run_once_with_timestamp_mode( + timestamp_mode: RetentionTimestampMode, + secondary_timestamp_field_opt: Option<&'static str>, + expected_marked_split_ids: &'static [&'static str], + ) { + let mock_metastore = mock_metastore_for_timestamp_mode( + secondary_timestamp_field_opt, + expected_marked_split_ids, + ); + let mut retention_policy_executor = + RetentionPolicyExecutor::new(MetastoreServiceClient::from_mock(mock_metastore)); + retention_policy_executor.timestamp_mode = timestamp_mode; + let universe = Universe::with_accelerated_time(); + let (_mailbox, handle) = universe.spawn_builder().spawn(retention_policy_executor); + handle.process_pending_and_observe().await; + + universe.sleep(shift_time_by()).await; + let counters = handle.process_pending_and_observe().await.state; + // Depending on the wall clock, the hourly schedule can fire more than once. + assert!(counters.num_execution_passes >= 1); + assert_eq!( + counters.num_expired_splits, + 2 * counters.num_execution_passes + ); + universe.assert_quit().await; + } + + #[tokio::test] + async fn test_retention_policy_execution_primary_mode() { + run_once_with_timestamp_mode( + RetentionTimestampMode::Primary, + Some("ingestion_time"), + &["both", "primary-only"], + ) + .await; + } + + #[tokio::test] + async fn test_retention_policy_execution_secondary_dry_run_mode() { + // The dry run lists both sets but only marks the primary one. + run_once_with_timestamp_mode( + RetentionTimestampMode::SecondaryDryRun, + Some("ingestion_time"), + &["both", "primary-only"], + ) + .await; + } + + #[tokio::test] + async fn test_retention_policy_execution_secondary_mode() { + run_once_with_timestamp_mode( + RetentionTimestampMode::Secondary, + Some("ingestion_time"), + &["both", "secondary-only"], + ) + .await; + } + + #[tokio::test] + async fn test_retention_policy_execution_secondary_modes_without_secondary_field() { + for timestamp_mode in [ + RetentionTimestampMode::SecondaryDryRun, + RetentionTimestampMode::Secondary, + ] { + run_once_with_timestamp_mode(timestamp_mode, None, &["both", "primary-only"]).await; + } + } } diff --git a/quickwit/quickwit-janitor/src/retention_policy_execution.rs b/quickwit/quickwit-janitor/src/retention_policy_execution.rs index e0ed448ff86..1e01c809127 100644 --- a/quickwit/quickwit-janitor/src/retention_policy_execution.rs +++ b/quickwit/quickwit-janitor/src/retention_policy_execution.rs @@ -12,9 +12,14 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::collections::HashSet; +use std::ops::RangeInclusive; +use std::str::FromStr; + +use itertools::Itertools; use quickwit_actors::ActorContext; use quickwit_common::pretty::PrettySample; -use quickwit_config::RetentionPolicy; +use quickwit_config::{RetentionPolicy, RetentionTimestampType}; use quickwit_metastore::{ ListSplitsQuery, ListSplitsRequestExt, MetastoreServiceStreamSplitsExt, SplitMetadata, SplitState, @@ -24,10 +29,53 @@ use quickwit_proto::metastore::{ }; use quickwit_proto::types::{IndexUid, SplitId}; use time::OffsetDateTime; +use time::format_description::well_known::Rfc3339; use tracing::{info, warn}; use crate::actors::RetentionPolicyExecutor; +const RETENTION_TIMESTAMP_MODE_ENV_KEY: &str = "QW_RETENTION_TIMESTAMP_MODE"; + +/// Maximum number of splits logged for each difference reported by the dry run. +const DRY_RUN_MAX_SAMPLE_SPLITS: usize = 10; + +/// Selects the split time range against which retention policies are evaluated. +/// +/// Configured for the whole janitor with the `QW_RETENTION_TIMESTAMP_MODE` environment +/// variable (`primary`, `secondary-dry-run` or `secondary`). +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub enum RetentionTimestampMode { + /// Splits expire based on their primary time range. + #[default] + Primary, + /// Splits expire based on their primary time range, but the splits that the `Secondary` + /// mode would treat differently are logged. + SecondaryDryRun, + /// Splits expire based on their secondary time range, or on their primary time range + /// when they don't have a secondary one. + Secondary, +} + +impl FromStr for RetentionTimestampMode { + type Err = String; + + fn from_str(mode_str: &str) -> Result { + match mode_str { + "primary" => Ok(Self::Primary), + "secondary-dry-run" => Ok(Self::SecondaryDryRun), + "secondary" => Ok(Self::Secondary), + _ => Err(format!("unknown retention timestamp mode `{mode_str}`")), + } + } +} + +impl RetentionTimestampMode { + /// Reads the mode from the environment. Unset or invalid values fall back to `Primary`. + pub fn from_env() -> Self { + quickwit_common::get_from_env(RETENTION_TIMESTAMP_MODE_ENV_KEY, Self::Primary, false) + } +} + /// Detect all expired splits based a retention policy and /// only mark them as `MarkedForDeletion`. Actual split deletion /// is taken care of by the garbage collector. @@ -35,21 +83,122 @@ use crate::actors::RetentionPolicyExecutor; /// * `index_id` - The target index id. /// * `metastore` - The metastore managing the target index. /// * `retention_policy` - The retention policy to used to evaluate the splits. +/// * `timestamp_mode` - The split time range the retention policy is evaluated against. +/// * `secondary_timestamp_field_opt` - The secondary timestamp field of the index, if any. /// * `ctx_opt` - A context for reporting progress (only useful within quickwit actor). pub async fn run_execute_retention_policy( index_uid: IndexUid, metastore: MetastoreServiceClient, retention_policy: &RetentionPolicy, + timestamp_mode: RetentionTimestampMode, + secondary_timestamp_field_opt: Option<&str>, ctx: &ActorContext, ) -> anyhow::Result> { // Select splits that are published and older than the retention period. let retention_period = retention_policy.retention_period()?; let current_timestamp = OffsetDateTime::now_utc().unix_timestamp(); 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_max_time_range_end(max_retention_timestamp); + // Without a secondary timestamp field, every split falls back to its primary time range, so + // the secondary filter is skipped altogether. + let expired_splits = match (timestamp_mode, secondary_timestamp_field_opt) { + (RetentionTimestampMode::Primary, _) | (_, None) => { + list_expired_splits( + &index_uid, + &metastore, + max_retention_timestamp, + RetentionTimestampType::Primary, + ctx, + ) + .await? + } + (RetentionTimestampMode::Secondary, Some(secondary_timestamp_field)) => { + let secondary_expired_splits = list_expired_splits( + &index_uid, + &metastore, + max_retention_timestamp, + RetentionTimestampType::Secondary, + ctx, + ) + .await?; + warn_on_primary_fallback( + &index_uid, + secondary_timestamp_field, + &secondary_expired_splits, + ); + secondary_expired_splits + } + (RetentionTimestampMode::SecondaryDryRun, Some(secondary_timestamp_field)) => { + let primary_expired_splits = list_expired_splits( + &index_uid, + &metastore, + max_retention_timestamp, + RetentionTimestampType::Primary, + ctx, + ) + .await?; + let secondary_expired_splits = list_expired_splits( + &index_uid, + &metastore, + max_retention_timestamp, + RetentionTimestampType::Secondary, + ctx, + ) + .await?; + warn_on_primary_fallback( + &index_uid, + secondary_timestamp_field, + &secondary_expired_splits, + ); + log_secondary_dry_run( + &index_uid, + max_retention_timestamp, + &primary_expired_splits, + &secondary_expired_splits, + ); + primary_expired_splits + } + }; + if expired_splits.is_empty() { + return Ok(expired_splits); + } + // Mark the expired splits for deletion. + let expired_split_ids: Vec = expired_splits + .iter() + .map(|split_metadata| split_metadata.split_id.to_string()) + .collect(); + info!( + index_id=%index_uid.index_id, + split_ids=?PrettySample::new(&expired_split_ids, 5), + ?timestamp_mode, + "Marking {} splits for deletion based on retention policy.", + expired_split_ids.len() + ); + let mark_splits_for_deletion_request = + MarkSplitsForDeletionRequest::new(index_uid, expired_split_ids); + ctx.protect_future(metastore.mark_splits_for_deletion(mark_splits_for_deletion_request)) + .await?; + Ok(expired_splits) +} + +/// Lists the published splits whose time range of type `timestamp_type` ends before +/// `max_retention_timestamp`. For `Secondary`, the metastore falls back to the primary time +/// range for splits without a secondary one. +async fn list_expired_splits( + index_uid: &IndexUid, + metastore: &MetastoreServiceClient, + max_retention_timestamp: i64, + timestamp_type: RetentionTimestampType, + ctx: &ActorContext, +) -> anyhow::Result> { + let query = + ListSplitsQuery::for_index(index_uid.clone()).with_split_state(SplitState::Published); + let query = match timestamp_type { + RetentionTimestampType::Primary => query.with_max_time_range_end(max_retention_timestamp), + RetentionTimestampType::Secondary => { + query.with_max_secondary_time_range_end(max_retention_timestamp) + } + }; let list_splits_request = ListSplitsRequest::try_from_list_splits_query(&query)?; let (expired_splits, ignored_splits): (Vec, Vec) = ctx .protect_future(metastore.list_splits(list_splits_request)) @@ -57,7 +206,12 @@ pub async fn run_execute_retention_policy( .collect_splits_metadata() .await? .into_iter() - .partition(|split_metadata| split_metadata.time_range.is_some()); + .partition(|split_metadata| match timestamp_type { + RetentionTimestampType::Primary => split_metadata.time_range.is_some(), + RetentionTimestampType::Secondary => { + split_metadata.time_range.is_some() || split_metadata.secondary_time_range.is_some() + } + }); if !ignored_splits.is_empty() { let ignored_split_ids: Vec = ignored_splits @@ -71,23 +225,246 @@ pub async fn run_execute_retention_policy( ignored_split_ids.len() ); } - if expired_splits.is_empty() { - return Ok(expired_splits); + Ok(expired_splits) +} + +/// Warns about expired splits that lack a secondary time range although the index has a +/// secondary timestamp field, because the secondary mode silently evaluates them against +/// their primary time range. +fn warn_on_primary_fallback( + index_uid: &IndexUid, + secondary_timestamp_field: &str, + secondary_expired_splits: &[SplitMetadata], +) { + let fallback_split_ids: Vec<&str> = secondary_expired_splits + .iter() + .filter(|split| split.secondary_time_range.is_none()) + .map(|split| split.split_id()) + .collect(); + if fallback_split_ids.is_empty() { + return; } - // Mark the expired splits for deletion. - let expired_split_ids: Vec = expired_splits + warn!( + index_id=%index_uid.index_id, + secondary_timestamp_field, + split_ids=?PrettySample::new(&fallback_split_ids, 5), + "Retention policy fell back to the primary timestamp for {} expired splits because they \ + lack a secondary time range.", + fallback_split_ids.len() + ); +} + +/// Splits on which the primary and secondary retention timestamp modes disagree. +#[derive(Debug, Default)] +struct SecondaryDryRunDiff<'a> { + /// Splits expired in secondary mode but not in primary mode: switching to the + /// secondary mode deletes them earlier than today. + expired_earlier: Vec<&'a SplitMetadata>, + /// Splits expired in primary mode but not in secondary mode: switching to the + /// secondary mode retains them longer than today. + retained_longer: Vec<&'a SplitMetadata>, +} + +fn diff_secondary_dry_run<'a>( + primary_expired_splits: &'a [SplitMetadata], + secondary_expired_splits: &'a [SplitMetadata], +) -> SecondaryDryRunDiff<'a> { + let primary_split_ids: HashSet<&str> = primary_expired_splits .iter() - .map(|split_metadata| split_metadata.split_id.to_string()) + .map(|split| split.split_id()) .collect(); + let secondary_split_ids: HashSet<&str> = secondary_expired_splits + .iter() + .map(|split| split.split_id()) + .collect(); + let mut expired_earlier: Vec<&SplitMetadata> = secondary_expired_splits + .iter() + .filter(|split| !primary_split_ids.contains(split.split_id())) + .collect(); + // The splits with the most recent primary timestamps are the most surprising ones. + expired_earlier.sort_by_key(|split| { + std::cmp::Reverse(split.time_range.as_ref().map(|range| *range.end())) + }); + let retained_longer: Vec<&SplitMetadata> = primary_expired_splits + .iter() + .filter(|split| !secondary_split_ids.contains(split.split_id())) + .collect(); + SecondaryDryRunDiff { + expired_earlier, + retained_longer, + } +} + +fn log_secondary_dry_run( + index_uid: &IndexUid, + max_retention_timestamp: i64, + primary_expired_splits: &[SplitMetadata], + secondary_expired_splits: &[SplitMetadata], +) { + let diff = diff_secondary_dry_run(primary_expired_splits, secondary_expired_splits); info!( index_id=%index_uid.index_id, - split_ids=?PrettySample::new(&expired_split_ids, 5), - "Marking {} splits for deletion based on retention policy.", - expired_split_ids.len() + cutoff=%format_timestamp(max_retention_timestamp), + num_expired_primary=primary_expired_splits.len(), + num_expired_secondary=secondary_expired_splits.len(), + num_expired_earlier=diff.expired_earlier.len(), + num_retained_longer=diff.retained_longer.len(), + "retention secondary timestamp dry run" ); - let mark_splits_for_deletion_request = - MarkSplitsForDeletionRequest::new(index_uid, expired_split_ids); - ctx.protect_future(metastore.mark_splits_for_deletion(mark_splits_for_deletion_request)) - .await?; - Ok(expired_splits) + if !diff.expired_earlier.is_empty() { + warn!( + index_id=%index_uid.index_id, + cutoff=%format_timestamp(max_retention_timestamp), + num_splits=diff.expired_earlier.len(), + num_docs=diff.expired_earlier.iter().map(|split| split.num_docs).sum::(), + size_bytes=diff.expired_earlier.iter().map(|split| split.footer_offsets.end).sum::(), + splits=%format_split_samples(&diff.expired_earlier), + "retention secondary timestamp dry run: splits that would be deleted in secondary mode but not in primary mode" + ); + } + if !diff.retained_longer.is_empty() { + info!( + index_id=%index_uid.index_id, + cutoff=%format_timestamp(max_retention_timestamp), + num_splits=diff.retained_longer.len(), + num_docs=diff.retained_longer.iter().map(|split| split.num_docs).sum::(), + size_bytes=diff.retained_longer.iter().map(|split| split.footer_offsets.end).sum::(), + splits=%format_split_samples(&diff.retained_longer), + "retention secondary timestamp dry run: splits that would be retained in secondary mode but are deleted in primary mode" + ); + } +} + +fn format_split_samples(splits: &[&SplitMetadata]) -> String { + let mut samples = splits + .iter() + .take(DRY_RUN_MAX_SAMPLE_SPLITS) + .map(|split| { + format!( + "{} time_range={} secondary_time_range={} create_timestamp={} num_docs={} \ + size_bytes={}", + split.split_id, + format_time_range(split.time_range.as_ref()), + format_time_range(split.secondary_time_range.as_ref()), + format_timestamp(split.create_timestamp), + split.num_docs, + split.footer_offsets.end, + ) + }) + .join("; "); + if splits.len() > DRY_RUN_MAX_SAMPLE_SPLITS { + samples.push_str(&format!( + "; and {} more", + splits.len() - DRY_RUN_MAX_SAMPLE_SPLITS + )); + } + samples +} + +fn format_time_range(time_range_opt: Option<&RangeInclusive>) -> String { + match time_range_opt { + Some(time_range) => format!( + "[{}, {}]", + format_timestamp(*time_range.start()), + format_timestamp(*time_range.end()) + ), + None => "none".to_string(), + } +} + +fn format_timestamp(timestamp: i64) -> String { + OffsetDateTime::from_unix_timestamp(timestamp) + .ok() + .and_then(|datetime| datetime.format(&Rfc3339).ok()) + .unwrap_or_else(|| timestamp.to_string()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn make_split_metadata( + split_id: &str, + time_range: RangeInclusive, + secondary_time_range: Option>, + ) -> SplitMetadata { + SplitMetadata { + split_id: split_id.to_string(), + time_range: Some(time_range), + secondary_time_range, + ..Default::default() + } + } + + #[test] + fn test_retention_timestamp_mode_from_str() { + assert_eq!( + "primary".parse::().unwrap(), + RetentionTimestampMode::Primary + ); + assert_eq!( + "secondary-dry-run" + .parse::() + .unwrap(), + RetentionTimestampMode::SecondaryDryRun + ); + assert_eq!( + "secondary".parse::().unwrap(), + RetentionTimestampMode::Secondary + ); + assert!("Secondary".parse::().is_err()); + assert!( + "secondary_dry_run" + .parse::() + .is_err() + ); + assert_eq!( + RetentionTimestampMode::default(), + RetentionTimestampMode::Primary + ); + } + + #[test] + fn test_diff_secondary_dry_run() { + let primary_expired_splits = vec![ + make_split_metadata("both", 0..=100, Some(10..=110)), + make_split_metadata("backfilled", 0..=100, Some(5000..=5010)), + ]; + let secondary_expired_splits = vec![ + make_split_metadata("both", 0..=100, Some(10..=110)), + make_split_metadata("future-1", 0..=8000, Some(10..=110)), + make_split_metadata("future-2", 0..=9000, Some(10..=110)), + ]; + let diff = diff_secondary_dry_run(&primary_expired_splits, &secondary_expired_splits); + let expired_earlier_ids: Vec<&str> = diff + .expired_earlier + .iter() + .map(|split| split.split_id()) + .collect(); + assert_eq!(expired_earlier_ids, ["future-2", "future-1"]); + let retained_longer_ids: Vec<&str> = diff + .retained_longer + .iter() + .map(|split| split.split_id()) + .collect(); + assert_eq!(retained_longer_ids, ["backfilled"]); + } + + #[test] + fn test_format_split_samples() { + let splits: Vec = (0..12) + .map(|i| make_split_metadata(&format!("split-{i}"), 0..=60, None)) + .collect(); + let split_refs: Vec<&SplitMetadata> = splits.iter().collect(); + let samples = format_split_samples(&split_refs[..1]); + assert_eq!( + samples, + "split-0 time_range=[1970-01-01T00:00:00Z, 1970-01-01T00:01:00Z] \ + secondary_time_range=none create_timestamp=1970-01-01T00:00:00Z num_docs=0 \ + size_bytes=0" + ); + let samples = format_split_samples(&split_refs); + assert!(samples.ends_with("; and 2 more")); + assert_eq!(samples.matches("time_range=[").count(), 10); + } } diff --git a/quickwit/quickwit-metastore/src/tests/list_splits.rs b/quickwit/quickwit-metastore/src/tests/list_splits.rs index dbc2d6be918..ccbd8297496 100644 --- a/quickwit/quickwit-metastore/src/tests/list_splits.rs +++ b/quickwit/quickwit-metastore/src/tests/list_splits.rs @@ -1774,3 +1774,74 @@ pub async fn test_metastore_list_splits_from_all_indexes< cleanup_index(&mut metastore, index_uid_1.clone()).await; cleanup_index(&mut metastore, index_uid_2.clone()).await; } + +pub async fn test_metastore_list_splits_by_max_secondary_time_range_end< + MetastoreToTest: MetastoreServiceExt + DefaultForTest, +>() { + let mut metastore = MetastoreToTest::default_for_test().await; + + let index_id = append_random_suffix("test-max-secondary-end"); + let index_uri = format!("ram:///indexes/{index_id}"); + let index_config = IndexConfig::for_test(&index_id, &index_uri); + + let create_index_request = CreateIndexRequest::try_from_index_config(&index_config).unwrap(); + let index_uid: IndexUid = metastore + .create_index(create_index_request) + .await + .unwrap() + .index_uid + .unwrap(); + + // (split name, time range, secondary time range), evaluated against a max end of 1000. + let split_specs = [ + ("no-secondary-expired", Some(0..=500), None), + ("no-secondary-not-expired", Some(0..=1500), None), + ("both-expired", Some(0..=500), Some(0..=900)), + ("only-primary-expired", Some(0..=500), Some(0..=1500)), + ("only-secondary-expired", Some(0..=1500), Some(0..=900)), + ("both-at-max-end", Some(0..=1000), Some(0..=1000)), + ("no-time-range", None, None), + ("only-secondary-time-range", None, Some(0..=900)), + ]; + let splits_metadata: Vec = split_specs + .iter() + .map(|(name, time_range, secondary_time_range)| SplitMetadata { + split_id: name.to_string(), + index_uid: index_uid.clone(), + time_range: time_range.clone(), + secondary_time_range: secondary_time_range.clone(), + ..Default::default() + }) + .collect(); + let stage_splits_request = + StageSplitsRequest::try_from_splits_metadata(index_uid.clone(), splits_metadata).unwrap(); + metastore.stage_splits(stage_splits_request).await.unwrap(); + + let query = + ListSplitsQuery::for_index(index_uid.clone()).with_max_secondary_time_range_end(1000); + let splits = metastore + .list_splits(ListSplitsRequest::try_from_list_splits_query(&query).unwrap()) + .await + .unwrap() + .collect_splits() + .await + .unwrap(); + let split_ids: Vec<&str> = splits + .iter() + .map(|split| split.split_metadata.split_id.as_str()) + .sorted() + .collect(); + let expected_split_ids: Vec = [ + "both-at-max-end", + "both-expired", + "no-secondary-expired", + "only-secondary-expired", + "only-secondary-time-range", + ] + .iter() + .map(|name| name.to_string()) + .collect(); + assert_eq!(split_ids, expected_split_ids); + + cleanup_index(&mut metastore, index_uid).await; +} diff --git a/quickwit/quickwit-metastore/src/tests/mod.rs b/quickwit/quickwit-metastore/src/tests/mod.rs index c6177da1b43..20c45815fd7 100644 --- a/quickwit/quickwit-metastore/src/tests/mod.rs +++ b/quickwit/quickwit-metastore/src/tests/mod.rs @@ -380,6 +380,13 @@ macro_rules! metastore_test_suite { $crate::tests::list_splits::test_metastore_list_splits_by_node_id::<$metastore_type>().await; } + #[tokio::test] + #[serial_test::file_serial] + async fn test_metastore_list_splits_by_max_secondary_time_range_end() { + let _ = tracing_subscriber::fmt::try_init(); + $crate::tests::list_splits::test_metastore_list_splits_by_max_secondary_time_range_end::<$metastore_type>().await; + } + #[tokio::test] #[serial_test::file_serial] async fn test_metastore_split_update_timestamp() {