rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-15 16:32:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: keep after-merge mode in LastRow selector

Browse Source 
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-09-09 17:25:39 +08:00
parent 12b2cc5ad3
commit 1572325f55
15 changed files with 80 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cmd/src/datanode/scanbench.rs
+1 -1
View File
@@ -389,7 +389,7 @@ fn resolve_series_row_selector(
scan_config: &ScanConfig,
) -> error::Result<Option<TimeSeriesRowSelector>> {
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}'"),
}
src/metric-engine/src/engine/read.rs
+2 -4
View File
@@ -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)
src/mito2/src/cache.rs
+1 -2
View File
@@ -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,
src/mito2/src/engine/row_selector_test.rs
+4 -5
View File
@@ -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()
},
)
src/mito2/src/engine/scan_test.rs
+5 -3
View File
@@ -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()
},
)
src/mito2/src/read/last_row.rs
+1 -1
View File
@@ -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) {
src/mito2/src/read/range_cache.rs
+17 -22
View File
@@ -60,7 +60,6 @@ pub(crate) struct ScanRequestFingerprint {
/// Filters with the time index column.
time_filters: Option<Arc<Vec<String>>>,
series_row_selector: Option<TimeSeriesRowSelector>,
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<String>,
pub(crate) time_filters: Vec<String>,
pub(crate) series_row_selector: Option<TimeSeriesRowSelector>,
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<String>,
time_filters: Vec<String>,
series_row_selector: Option<TimeSeriesRowSelector>,
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);
src/mito2/src/read/scan_region.rs
+6 -18
View File
@@ -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<TimeSeriesRowSelector>,
/// 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<TimeSeriesDistribution>,
/// 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,
src/mito2/src/read/scan_util.rs
+8 -7
View File
@@ -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(),
)
src/mito2/src/read/seq_scan.rs
+1 -1
View File
@@ -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,
src/mito2/src/sst/parquet/file_range.rs
+4 -4
View File
@@ -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
src/query/src/dummy_catalog.rs
-7
View File
@@ -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
src/query/src/optimizer/scan_hint.rs
+17 -26
View File
@@ -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::<HashMap<_, _>>();
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::<HashMap<_, _>>();
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]
src/query/src/optimizer/scan_hint/vector_search.rs
+2 -2
View File
@@ -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);
src/store-api/src/storage/requests.rs
+11 -9
View File
@@ -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<usize>,
/// Optional hint to select rows from time-series.
pub series_row_selector: Option<TimeSeriesRowSelector>,
/// 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 {