From 1572325f556d5f0dd8274dca12bf405b7737bf1b Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Wed, 9 Sep 2026 12:33:35 +0800 Subject: [PATCH] refactor: keep after-merge mode in LastRow selector Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/cmd/src/datanode/scanbench.rs | 2 +- src/metric-engine/src/engine/read.rs | 6 +-- src/mito2/src/cache.rs | 3 +- src/mito2/src/engine/row_selector_test.rs | 9 ++-- src/mito2/src/engine/scan_test.rs | 8 ++-- src/mito2/src/read/last_row.rs | 2 +- src/mito2/src/read/range_cache.rs | 39 ++++++++--------- src/mito2/src/read/scan_region.rs | 24 +++-------- src/mito2/src/read/scan_util.rs | 15 ++++--- src/mito2/src/read/seq_scan.rs | 2 +- src/mito2/src/sst/parquet/file_range.rs | 8 ++-- src/query/src/dummy_catalog.rs | 7 --- src/query/src/optimizer/scan_hint.rs | 43 ++++++++----------- .../src/optimizer/scan_hint/vector_search.rs | 4 +- src/store-api/src/storage/requests.rs | 20 +++++---- 15 files changed, 80 insertions(+), 112 deletions(-) diff --git a/src/cmd/src/datanode/scanbench.rs b/src/cmd/src/datanode/scanbench.rs index 61d34e9e133..82a3460e4b4 100644 --- a/src/cmd/src/datanode/scanbench.rs +++ b/src/cmd/src/datanode/scanbench.rs @@ -389,7 +389,7 @@ fn resolve_series_row_selector( scan_config: &ScanConfig, ) -> error::Result> { match scan_config.series_row_selector.as_deref() { - Some("last_row") => Ok(Some(TimeSeriesRowSelector::LastRow)), + Some("last_row") => Ok(Some(TimeSeriesRowSelector::LastRow { after_merge: false })), Some(other) => Err(error::IllegalConfigSnafu { msg: format!("Unknown series_row_selector '{other}'"), } diff --git a/src/metric-engine/src/engine/read.rs b/src/metric-engine/src/engine/read.rs index 785cd318fe6..d1afce5e22d 100644 --- a/src/metric-engine/src/engine/read.rs +++ b/src/metric-engine/src/engine/read.rs @@ -429,8 +429,7 @@ mod test { let scan_req = ScanRequest { projection, filters: vec![], - series_row_selector: Some(TimeSeriesRowSelector::LastRow), - series_row_selector_after_merge: true, + series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: true }), ..Default::default() }; @@ -448,9 +447,8 @@ mod test { assert_eq!(scan_req.filters.len(), 1); assert_eq!( scan_req.series_row_selector, - Some(TimeSeriesRowSelector::LastRow) + Some(TimeSeriesRowSelector::LastRow { after_merge: true }) ); - assert!(scan_req.series_row_selector_after_merge); assert_eq!( scan_req.filters[0], logical_expr::col(DATA_SCHEMA_TABLE_ID_COLUMN_NAME) diff --git a/src/mito2/src/cache.rs b/src/mito2/src/cache.rs index da155d5d4f3..2e98ef33454 100644 --- a/src/mito2/src/cache.rs +++ b/src/mito2/src/cache.rs @@ -2621,7 +2621,7 @@ mod tests { let key = SelectorResultKey { file_id, row_group_idx: 0, - selector: TimeSeriesRowSelector::LastRow, + selector: TimeSeriesRowSelector::LastRow { after_merge: false }, }; assert!(cache.get_selector_result(&key).is_none()); let result = Arc::new(SelectorResultValue::new( @@ -2770,7 +2770,6 @@ mod tests { filters: vec!["tag_0 = 1".to_string()], time_filters: vec![], series_row_selector: None, - series_row_selector_after_merge: false, append_mode: false, filter_deleted: true, merge_mode: crate::region::options::MergeMode::LastRow, diff --git a/src/mito2/src/engine/row_selector_test.rs b/src/mito2/src/engine/row_selector_test.rs index 8c4c05e696f..7fd937e85d5 100644 --- a/src/mito2/src/engine/row_selector_test.rs +++ b/src/mito2/src/engine/row_selector_test.rs @@ -103,7 +103,7 @@ async fn test_last_row(append_mode: bool, flat_format: bool) { .scanner( region_id, ScanRequest { - series_row_selector: Some(TimeSeriesRowSelector::LastRow), + series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: false }), ..Default::default() }, ) @@ -126,7 +126,7 @@ async fn scan_last_row( region_id, ScanRequest { filters, - series_row_selector: Some(TimeSeriesRowSelector::LastRow), + series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: false }), ..Default::default() }, ) @@ -254,7 +254,7 @@ async fn last_row_scanner(engine: &MitoEngine, region_id: RegionId) -> Scanner { .scanner( region_id, ScanRequest { - series_row_selector: Some(TimeSeriesRowSelector::LastRow), + series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: false }), ..Default::default() }, ) @@ -446,8 +446,7 @@ async fn test_last_row_returns_older_put_after_newest_delete_across_ssts() { .scanner( region_id, ScanRequest { - series_row_selector: Some(TimeSeriesRowSelector::LastRow), - series_row_selector_after_merge: true, + series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: true }), ..Default::default() }, ) diff --git a/src/mito2/src/engine/scan_test.rs b/src/mito2/src/engine/scan_test.rs index df5035cdc51..a9d138367ab 100644 --- a/src/mito2/src/engine/scan_test.rs +++ b/src/mito2/src/engine/scan_test.rs @@ -2197,7 +2197,7 @@ async fn test_exact_sequence_read_with_last_row_selector_keeps_in_range_rows() { memtable_min_sequence: Some(0), memtable_max_sequence: Some(1), exact_sequence_range: true, - series_row_selector: Some(TimeSeriesRowSelector::LastRow), + series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: false }), ..Default::default() }) .await @@ -2213,7 +2213,7 @@ async fn test_exact_sequence_read_with_last_row_selector_keeps_in_range_rows() { | series | 2.0 | 1970-01-01T00:00:02 | +--------+---------+---------------------+", scan(ScanRequest { - series_row_selector: Some(TimeSeriesRowSelector::LastRow), + series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: false }), ..Default::default() }) .await @@ -2583,7 +2583,9 @@ async fn test_non_preserve_compaction_sequence_collision_with_format(flat_format .scan_to_stream( region_id, ScanRequest { - series_row_selector: Some(TimeSeriesRowSelector::LastRow), + series_row_selector: Some(TimeSeriesRowSelector::LastRow { + after_merge: false, + }), ..Default::default() }, ) diff --git a/src/mito2/src/read/last_row.rs b/src/mito2/src/read/last_row.rs index c2316525f70..c526bfcaa86 100644 --- a/src/mito2/src/read/last_row.rs +++ b/src/mito2/src/read/last_row.rs @@ -144,7 +144,7 @@ impl FlatRowGroupLastRowCachedReader { let key = SelectorResultKey { file_id, row_group_idx, - selector: TimeSeriesRowSelector::LastRow, + selector: TimeSeriesRowSelector::LastRow { after_merge: false }, }; if let Some(value) = cache_strategy.get_selector_result(&key) { diff --git a/src/mito2/src/read/range_cache.rs b/src/mito2/src/read/range_cache.rs index 17e7ce57c8e..ddf8c3cc356 100644 --- a/src/mito2/src/read/range_cache.rs +++ b/src/mito2/src/read/range_cache.rs @@ -60,7 +60,6 @@ pub(crate) struct ScanRequestFingerprint { /// Filters with the time index column. time_filters: Option>>, series_row_selector: Option, - series_row_selector_after_merge: bool, append_mode: bool, filter_deleted: bool, merge_mode: MergeMode, @@ -88,7 +87,6 @@ pub(crate) struct ScanRequestFingerprintBuilder { pub(crate) filters: Vec, pub(crate) time_filters: Vec, pub(crate) series_row_selector: Option, - pub(crate) series_row_selector_after_merge: bool, pub(crate) append_mode: bool, pub(crate) filter_deleted: bool, pub(crate) merge_mode: MergeMode, @@ -104,7 +102,6 @@ impl ScanRequestFingerprintBuilder { filters, time_filters, series_row_selector, - series_row_selector_after_merge, append_mode, filter_deleted, merge_mode, @@ -120,7 +117,6 @@ impl ScanRequestFingerprintBuilder { }), time_filters: (!time_filters.is_empty()).then(|| Arc::new(time_filters)), series_row_selector, - series_row_selector_after_merge, append_mode, filter_deleted, merge_mode, @@ -172,7 +168,6 @@ impl ScanRequestFingerprint { inner: Arc::clone(&self.inner), time_filters: None, series_row_selector: self.series_row_selector, - series_row_selector_after_merge: self.series_row_selector_after_merge, append_mode: self.append_mode, filter_deleted: self.filter_deleted, merge_mode: self.merge_mode, @@ -187,7 +182,6 @@ impl ScanRequestFingerprint { inner: Arc::clone(&self.inner), time_filters: self.time_filters.clone(), series_row_selector: self.series_row_selector, - series_row_selector_after_merge: self.series_row_selector_after_merge, append_mode: self.append_mode, filter_deleted: self.filter_deleted, merge_mode: self.merge_mode, @@ -202,7 +196,6 @@ impl ScanRequestFingerprint { inner: Arc::clone(&self.inner), time_filters: self.time_filters.clone(), series_row_selector: self.series_row_selector, - series_row_selector_after_merge: self.series_row_selector_after_merge, append_mode: self.append_mode, filter_deleted: self.filter_deleted, merge_mode: self.merge_mode, @@ -875,7 +868,6 @@ pub fn bench_cache_flat_range_stream( filters: vec![], time_filters: vec![], series_row_selector: None, - series_row_selector_after_merge: false, append_mode: false, filter_deleted: false, merge_mode: MergeMode::LastRow, @@ -933,7 +925,6 @@ mod tests { filters: Vec, time_filters: Vec, series_row_selector: Option, - series_row_selector_after_merge: bool, filter_deleted: bool, partition_expr_version: u64, ) -> ScanRequestFingerprint { @@ -944,7 +935,6 @@ mod tests { filters, time_filters, series_row_selector, - series_row_selector_after_merge, append_mode: false, filter_deleted, merge_mode: MergeMode::LastRow, @@ -959,7 +949,7 @@ mod tests { let key = RangeScanCacheKey { region_id, row_groups: vec![], - scan: test_scan_fingerprint(vec![], vec![], None, false, false, 0), + scan: test_scan_fingerprint(vec![], vec![], None, false, 0), }; let metrics_set = ExecutionPlanMetricsSet::new(); @@ -1455,16 +1445,14 @@ mod tests { let ordinary = test_scan_fingerprint( vec!["k0 = 'foo'".to_string()], vec![], - Some(TimeSeriesRowSelector::LastRow), - false, + Some(TimeSeriesRowSelector::LastRow { after_merge: false }), true, 0, ); let after_merge = test_scan_fingerprint( vec!["k0 = 'foo'".to_string()], vec![], - Some(TimeSeriesRowSelector::LastRow), - true, + Some(TimeSeriesRowSelector::LastRow { after_merge: true }), true, 0, ); @@ -1475,15 +1463,14 @@ mod tests { #[test] fn true_selector_after_merge_is_preserved_by_fingerprint_transforms() { let normalized = - test_scan_fingerprint(vec!["k0 = 'foo'".to_string()], vec![], None, false, true, 0); + test_scan_fingerprint(vec!["k0 = 'foo'".to_string()], vec![], None, true, 0); assert!(normalized.time_filters().is_empty()); let fingerprint = test_scan_fingerprint( vec!["k0 = 'foo'".to_string()], vec!["ts >= 1000".to_string()], - Some(TimeSeriesRowSelector::LastRow), - true, + Some(TimeSeriesRowSelector::LastRow { after_merge: true }), true, 7, ); @@ -1497,10 +1484,18 @@ mod tests { assert_eq!(reset.filters(), fingerprint.filters()); assert!(reset.time_filters().is_empty()); assert_eq!(reset.series_row_selector, fingerprint.series_row_selector); - assert!(fingerprint.series_row_selector_after_merge); - assert!(reset.series_row_selector_after_merge); - assert!(candidate.series_row_selector_after_merge); - assert!(series_data.series_row_selector_after_merge); + assert_eq!( + fingerprint.series_row_selector, + Some(TimeSeriesRowSelector::LastRow { after_merge: true }) + ); + assert_eq!( + candidate.series_row_selector, + fingerprint.series_row_selector + ); + assert_eq!( + series_data.series_row_selector, + fingerprint.series_row_selector + ); assert_eq!(reset.append_mode, fingerprint.append_mode); assert_eq!(reset.filter_deleted, fingerprint.filter_deleted); assert_eq!(reset.merge_mode, fingerprint.merge_mode); diff --git a/src/mito2/src/read/scan_region.rs b/src/mito2/src/read/scan_region.rs index e6d4d9d3d47..56c53698e78 100644 --- a/src/mito2/src/read/scan_region.rs +++ b/src/mito2/src/read/scan_region.rs @@ -574,7 +574,6 @@ impl ScanRegion { .with_filter_deleted(self.filter_deleted) .with_merge_mode(self.version.options.merge_mode()) .with_series_row_selector(self.request.series_row_selector) - .with_series_row_selector_after_merge(self.request.series_row_selector_after_merge) .with_distribution(self.request.distribution) .with_explain_flat_format( self.version.options.sst_format == Some(crate::sst::FormatType::Flat), @@ -978,8 +977,6 @@ pub struct ScanInput { pub(crate) merge_mode: MergeMode, /// Hint to select rows from time series. pub(crate) series_row_selector: Option, - /// Whether the selector must run after cross-source merge and deduplication. - pub(crate) series_row_selector_after_merge: bool, /// Hint for the required distribution of the scanner. pub(crate) distribution: Option, /// Whether the region's configured SST format is flat. @@ -1054,7 +1051,6 @@ impl ScanInput { filter_deleted: true, merge_mode: MergeMode::default(), series_row_selector: None, - series_row_selector_after_merge: false, distribution: None, explain_flat_format: false, snapshot_sequence: None, @@ -1203,7 +1199,6 @@ impl ScanInputBuilder { filters, time_filters, series_row_selector: input.series_row_selector, - series_row_selector_after_merge: input.series_row_selector_after_merge, append_mode: input.append_mode, filter_deleted: input.filter_deleted, merge_mode: input.merge_mode, @@ -1386,13 +1381,6 @@ impl ScanInputBuilder { self } - /// Sets whether the row selector must run after merge and deduplication. - #[must_use] - pub(crate) fn with_series_row_selector_after_merge(mut self, after_merge: bool) -> Self { - self.input.series_row_selector_after_merge = after_merge; - self - } - #[must_use] pub(crate) fn with_snapshot_sequence( mut self, @@ -2515,8 +2503,7 @@ mod tests { ) .await .with_distribution(Some(TimeSeriesDistribution::PerSeries)) - .with_series_row_selector(Some(TimeSeriesRowSelector::LastRow)) - .with_series_row_selector_after_merge(true) + .with_series_row_selector(Some(TimeSeriesRowSelector::LastRow { after_merge: true })) .with_merge_mode(MergeMode::LastNonNull) .with_filter_deleted(false) .build(); @@ -2541,8 +2528,7 @@ mod tests { col("v0").gt(lit(1)).to_string(), ], time_filters: vec![col("ts").gt_eq(ts_lit(1000)).to_string()], - series_row_selector: Some(TimeSeriesRowSelector::LastRow), - series_row_selector_after_merge: true, + series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: true }), append_mode: false, filter_deleted: false, merge_mode: MergeMode::LastNonNull, @@ -2551,7 +2537,10 @@ mod tests { } .build(); assert_eq!(&expected, fingerprint); - assert!(input.series_row_selector_after_merge); + assert_eq!( + input.series_row_selector, + Some(TimeSeriesRowSelector::LastRow { after_merge: true }) + ); } #[tokio::test] @@ -2626,7 +2615,6 @@ mod tests { filters: vec![col("k0").eq(lit("foo")).to_string()], time_filters: vec![], series_row_selector: None, - series_row_selector_after_merge: false, append_mode: false, filter_deleted: true, merge_mode: MergeMode::LastRow, diff --git a/src/mito2/src/read/scan_util.rs b/src/mito2/src/read/scan_util.rs index 2fc9ed69af9..a67dfae158e 100644 --- a/src/mito2/src/read/scan_util.rs +++ b/src/mito2/src/read/scan_util.rs @@ -30,7 +30,7 @@ use futures::Stream; use prometheus::IntGauge; use smallvec::SmallVec; use snafu::ResultExt; -use store_api::storage::{RegionId, SequenceRange}; +use store_api::storage::{RegionId, SequenceRange, TimeSeriesRowSelector}; use crate::error::{ComputeArrowSnafu, Result}; use crate::memtable::MemScanMetrics; @@ -1571,12 +1571,13 @@ pub fn build_flat_file_range_scan_stream( let build_reader_start = Instant::now(); let Some(mut reader) = range .flat_reader( - if stream_ctx.input.sequence_range.is_none() - && !stream_ctx.input.series_row_selector_after_merge - { - stream_ctx.input.series_row_selector - } else { - None + match stream_ctx.input.series_row_selector { + Some(TimeSeriesRowSelector::LastRow { after_merge: false }) + if stream_ctx.input.sequence_range.is_none() => + { + stream_ctx.input.series_row_selector + } + _ => None, }, fetch_metrics.as_deref(), ) diff --git a/src/mito2/src/read/seq_scan.rs b/src/mito2/src/read/seq_scan.rs index 7253ba532f6..5d59ddfce11 100644 --- a/src/mito2/src/read/seq_scan.rs +++ b/src/mito2/src/read/seq_scan.rs @@ -263,7 +263,7 @@ impl SeqScan { }; let reader = match &stream_ctx.input.series_row_selector { - Some(TimeSeriesRowSelector::LastRow) => { + Some(TimeSeriesRowSelector::LastRow { .. }) => { Box::pin(FlatLastRowReader::new(reader).into_stream()) as _ } None => reader, diff --git a/src/mito2/src/sst/parquet/file_range.rs b/src/mito2/src/sst/parquet/file_range.rs index 20eaa81180d..f55aa524165 100644 --- a/src/mito2/src/sst/parquet/file_range.rs +++ b/src/mito2/src/sst/parquet/file_range.rs @@ -202,10 +202,10 @@ impl FileRange { )) .await?; - let use_last_row_reader = if selector - .map(|s| s == TimeSeriesRowSelector::LastRow) - .unwrap_or(false) - { + let use_last_row_reader = if matches!( + selector, + Some(TimeSeriesRowSelector::LastRow { after_merge: false }) + ) { // Only use LastRowReader if row group does not contain DELETE, all // rows are selected, and filters that still run after this reader // cannot change which row is last. Tag filters are safe because a diff --git a/src/query/src/dummy_catalog.rs b/src/query/src/dummy_catalog.rs index aea885d3808..3d1dfe5f375 100644 --- a/src/query/src/dummy_catalog.rs +++ b/src/query/src/dummy_catalog.rs @@ -286,13 +286,6 @@ impl DummyTableProvider { self.scan_request.lock().unwrap().series_row_selector = Some(selector); } - /// Sets the instant-derived LastRow hint, which must run after merge and deduplication. - pub fn with_last_row_selector_after_merge_hint(&self) { - let mut request = self.scan_request.lock().unwrap(); - request.series_row_selector = Some(TimeSeriesRowSelector::LastRow); - request.series_row_selector_after_merge = true; - } - /// Clones this provider for one logical table-scan use-site. /// /// The optimizer may attach different hints to scans that share the same diff --git a/src/query/src/optimizer/scan_hint.rs b/src/query/src/optimizer/scan_hint.rs index 6e78bac998a..89d02b79b71 100644 --- a/src/query/src/optimizer/scan_hint.rs +++ b/src/query/src/optimizer/scan_hint.rs @@ -125,7 +125,9 @@ impl ScanHintRule { // Apply the instant-derived hint after the aggregate hint. Both // select LastRow today, and this ordering preserves the existing // aggregate selector when the instant guard rejects a scan. - adapter.with_last_row_selector_after_merge_hint(); + adapter.with_time_series_selector_hint(TimeSeriesRowSelector::LastRow { + after_merge: true, + }); } table_scan.source = std::sync::Arc::new(DefaultTableSource::new(std::sync::Arc::new(adapter))); @@ -293,7 +295,9 @@ impl ScanHintRule { } if should_set_selector_hint { - adapter.with_time_series_selector_hint(TimeSeriesRowSelector::LastRow); + adapter.with_time_series_selector_hint(TimeSeriesRowSelector::LastRow { + after_merge: false, + }); } } } @@ -643,10 +647,9 @@ mod test { assert_eq!( scan_requests(&rewritten)[0].series_row_selector, - Some(TimeSeriesRowSelector::LastRow) + Some(TimeSeriesRowSelector::LastRow { after_merge: true }) ); - assert!(scan_requests(&rewritten)[0].series_row_selector_after_merge); - assert!(!provider.scan_request().series_row_selector_after_merge); + assert_eq!(provider.scan_request().series_row_selector, None); } @@ -660,9 +663,8 @@ mod test { .data; assert_eq!(scan_requests(&rewritten)[0].series_row_selector, None); - assert!(!scan_requests(&rewritten)[0].series_row_selector_after_merge); + assert_eq!(provider.scan_request().series_row_selector, None); - assert!(!provider.scan_request().series_row_selector_after_merge); } #[test] @@ -676,9 +678,8 @@ mod test { assert_eq!( scan_requests(&rewritten)[0].series_row_selector, - Some(TimeSeriesRowSelector::LastRow) + Some(TimeSeriesRowSelector::LastRow { after_merge: true }) ); - assert!(scan_requests(&rewritten)[0].series_row_selector_after_merge); } #[test] @@ -692,9 +693,8 @@ mod test { assert_eq!( scan_requests(&rewritten)[0].series_row_selector, - Some(TimeSeriesRowSelector::LastRow) + Some(TimeSeriesRowSelector::LastRow { after_merge: true }) ); - assert!(scan_requests(&rewritten)[0].series_row_selector_after_merge); } #[test] @@ -708,7 +708,6 @@ mod test { .data; assert_eq!(scan_requests(&rewritten)[0].series_row_selector, None); - assert!(!scan_requests(&rewritten)[0].series_row_selector_after_merge); } #[test] @@ -721,7 +720,6 @@ mod test { .data; assert_eq!(scan_requests(&rewritten)[0].series_row_selector, None); - assert!(!scan_requests(&rewritten)[0].series_row_selector_after_merge); } #[test] @@ -734,7 +732,6 @@ mod test { .data; assert_eq!(scan_requests(&rewritten)[0].series_row_selector, None); - assert!(!scan_requests(&rewritten)[0].series_row_selector_after_merge); } #[test] @@ -747,7 +744,6 @@ mod test { .data; assert_eq!(scan_requests(&rewritten)[0].series_row_selector, None); - assert!(!scan_requests(&rewritten)[0].series_row_selector_after_merge); } #[test] @@ -769,7 +765,7 @@ mod test { .collect::>(); assert_eq!( requests["outer"].series_row_selector, - Some(TimeSeriesRowSelector::LastRow) + Some(TimeSeriesRowSelector::LastRow { after_merge: true }) ); assert_eq!(requests["inner"].series_row_selector, None); } @@ -793,11 +789,11 @@ mod test { .collect::>(); assert_eq!( requests["outer"].series_row_selector, - Some(TimeSeriesRowSelector::LastRow) + Some(TimeSeriesRowSelector::LastRow { after_merge: true }) ); assert_eq!( requests["inner"].series_row_selector, - Some(TimeSeriesRowSelector::LastRow) + Some(TimeSeriesRowSelector::LastRow { after_merge: true }) ); } @@ -847,9 +843,8 @@ mod test { .data; assert_eq!( scan_requests(&rewritten)[0].series_row_selector, - Some(TimeSeriesRowSelector::LastRow) + Some(TimeSeriesRowSelector::LastRow { after_merge: true }) ); - assert!(scan_requests(&rewritten)[0].series_row_selector_after_merge); } #[test] @@ -869,13 +864,10 @@ mod test { assert_eq!(requests.len(), 2); assert_eq!( requests[0].series_row_selector, - Some(TimeSeriesRowSelector::LastRow) + Some(TimeSeriesRowSelector::LastRow { after_merge: true }) ); assert_eq!(requests[1].series_row_selector, None); - assert!(requests[0].series_row_selector_after_merge); - assert!(!requests[1].series_row_selector_after_merge); assert_eq!(provider.scan_request().series_row_selector, None); - assert!(!provider.scan_request().series_row_selector_after_merge); } #[test] @@ -941,9 +933,8 @@ mod test { let scan_req = scan_requests(&rewritten)[0].clone(); assert_eq!( scan_req.series_row_selector, - Some(TimeSeriesRowSelector::LastRow) + Some(TimeSeriesRowSelector::LastRow { after_merge: false }) ); - assert!(!scan_req.series_row_selector_after_merge); } #[test] diff --git a/src/query/src/optimizer/scan_hint/vector_search.rs b/src/query/src/optimizer/scan_hint/vector_search.rs index 2ffffa437ca..e60ac9c1223 100644 --- a/src/query/src/optimizer/scan_hint/vector_search.rs +++ b/src/query/src/optimizer/scan_hint/vector_search.rs @@ -634,7 +634,7 @@ mod tests { let request = scan_request_from_plan(&rewritten); assert_eq!( request.series_row_selector, - Some(store_api::storage::TimeSeriesRowSelector::LastRow) + Some(store_api::storage::TimeSeriesRowSelector::LastRow { after_merge: false }) ); assert!(request.vector_search.is_none()); } @@ -704,7 +704,7 @@ mod tests { assert_eq!( request_a.series_row_selector, - Some(store_api::storage::TimeSeriesRowSelector::LastRow) + Some(store_api::storage::TimeSeriesRowSelector::LastRow { after_merge: false }) ); assert!(request_a.vector_search.is_none()); assert_eq!(request_b.series_row_selector, None); diff --git a/src/store-api/src/storage/requests.rs b/src/store-api/src/storage/requests.rs index 7d040f85abf..0ef76e8c6ad 100644 --- a/src/store-api/src/storage/requests.rs +++ b/src/store-api/src/storage/requests.rs @@ -82,7 +82,11 @@ pub trait VectorIndexEngine: Send + Sync { #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Display)] pub enum TimeSeriesRowSelector { /// Only keep the last row of each time-series. - LastRow, + #[strum(to_string = "LastRow")] + LastRow { + /// Whether selection runs after cross-source merge and deduplication. + after_merge: bool, + }, } /// A hint on how to distribute time-series data on the scan output. @@ -112,11 +116,6 @@ pub struct ScanRequest { pub limit: Option, /// Optional hint to select rows from time-series. pub series_row_selector: Option, - /// Whether the row selector must run only after cross-source merge and deduplication. - /// - /// This is required by instant-derived LastRow scans. Ordinary selector users - /// keep the default `false` and may use source-local selector optimization. - pub series_row_selector_after_merge: bool, /// Optional constraint on the sequence number of the rows to read. /// If set, only rows with a sequence number **lesser or equal** to this value /// will be returned. @@ -210,7 +209,10 @@ impl Display for ScanRequest { series_row_selector )?; } - if self.series_row_selector_after_merge { + if matches!( + self.series_row_selector, + Some(TimeSeriesRowSelector::LastRow { after_merge: true }) + ) { write!( f, "{}series_row_selector_after_merge: true", @@ -335,14 +337,14 @@ mod tests { ); let request = ScanRequest { + series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: true }), snapshot_on_scan: true, - series_row_selector_after_merge: true, exact_sequence_range: true, ..Default::default() }; assert_eq!( request.to_string(), - "ScanRequest { series_row_selector_after_merge: true, snapshot_on_scan: true, exact_sequence_range: true }" + "ScanRequest { series_row_selector: LastRow, series_row_selector_after_merge: true, snapshot_on_scan: true, exact_sequence_range: true }" ); let request = ScanRequest {