From a29a2a5ede274e1957f449028b61246b2025f2ab Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Thu, 13 Aug 2026 21:01:29 +0800 Subject: [PATCH] feat(mito2): support exact sequence range reads Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- .../src/bin/query_perf_fixture/direct_sst.rs | 1 + src/cmd/src/datanode/objbench.rs | 1 + src/cmd/src/datanode/parquetbench.rs | 1 + src/mito2/src/access_layer.rs | 11 +- src/mito2/src/cache.rs | 1 + src/mito2/src/cache/write_cache.rs | 12 +- src/mito2/src/compaction/compactor.rs | 66 +- src/mito2/src/compaction/window.rs | 1 + src/mito2/src/engine.rs | 123 +- src/mito2/src/engine/alter_test.rs | 283 ++++ src/mito2/src/engine/edit_region_test.rs | 36 + src/mito2/src/engine/scan_test.rs | 1304 ++++++++++++++++- src/mito2/src/error.rs | 18 + src/mito2/src/flush.rs | 4 + src/mito2/src/read/range_cache.rs | 14 +- src/mito2/src/read/scan_region.rs | 146 +- src/mito2/src/read/scan_util.rs | 177 ++- src/mito2/src/region/options.rs | 69 + src/mito2/src/sst/file.rs | 36 + src/mito2/src/sst/index.rs | 1 + src/mito2/src/worker/handle_alter.rs | 52 +- src/mito2/src/worker/handle_manifest.rs | 8 +- src/store-api/src/mito_engine_options.rs | 4 + src/store-api/src/region_request.rs | 46 +- src/store-api/src/storage/requests.rs | 15 +- src/table/src/metadata.rs | 71 +- .../cases/downgrade_compatibility/case.toml | 6 +- .../cases/downgrade_compatibility/setup.sql | 23 + .../downgrade_compatibility/verify.result | 11 + .../cases/downgrade_compatibility/verify.sql | 2 + 30 files changed, 2474 insertions(+), 69 deletions(-) diff --git a/src/cmd/src/bin/query_perf_fixture/direct_sst.rs b/src/cmd/src/bin/query_perf_fixture/direct_sst.rs index 1bf436e6a8e..7297a54b2cb 100644 --- a/src/cmd/src/bin/query_perf_fixture/direct_sst.rs +++ b/src/cmd/src/bin/query_perf_fixture/direct_sst.rs @@ -310,6 +310,7 @@ fn file_meta_from_sst_info( num_series: info.num_series, primary_key_min: None, primary_key_max: None, + preserve_row_sequence: false, } } diff --git a/src/cmd/src/datanode/objbench.rs b/src/cmd/src/datanode/objbench.rs index f33734687dc..f3ffd8becf9 100644 --- a/src/cmd/src/datanode/objbench.rs +++ b/src/cmd/src/datanode/objbench.rs @@ -259,6 +259,7 @@ impl ObjbenchCommand { inverted_index_config: MitoConfig::default().inverted_index, fulltext_index_config, bloom_filter_index_config: MitoConfig::default().bloom_filter_index, + preserve_row_sequence: false, #[cfg(feature = "vector_index")] vector_index_config: Default::default(), }; diff --git a/src/cmd/src/datanode/parquetbench.rs b/src/cmd/src/datanode/parquetbench.rs index 301c6440ec1..d138a1561ce 100644 --- a/src/cmd/src/datanode/parquetbench.rs +++ b/src/cmd/src/datanode/parquetbench.rs @@ -324,6 +324,7 @@ impl ParquetbenchCommand { num_series: 0, primary_key_min: None, primary_key_max: None, + preserve_row_sequence: false, }, Arc::new(NoopFilePurger), ); diff --git a/src/mito2/src/access_layer.rs b/src/mito2/src/access_layer.rs index 05595bbae3b..79535604d80 100644 --- a/src/mito2/src/access_layer.rs +++ b/src/mito2/src/access_layer.rs @@ -342,6 +342,11 @@ impl AccessLayer { let region_id = request.metadata.region_id; let region_metadata = request.metadata.clone(); let cache_manager = request.cache_manager.clone(); + let override_sequence = if request.preserve_row_sequence { + None + } else { + request.max_sequence + }; let sst_info = if let Some(write_cache) = cache_manager.write_cache() { // Write to the write cache. @@ -397,14 +402,14 @@ impl AccessLayer { writer .write_all_flat_as_primary_key( request.source, - request.max_sequence, + override_sequence, write_opts, ) .await? } FormatType::Flat => { writer - .write_all_flat(request.source, request.max_sequence, write_opts) + .write_all_flat(request.source, override_sequence, write_opts) .await? } } @@ -565,6 +570,8 @@ pub struct SstWriteRequest { pub max_sequence: Option, pub sst_write_format: FormatType, + pub preserve_row_sequence: bool, + /// Configs for index pub index_options: IndexOptions, pub index_config: IndexConfig, diff --git a/src/mito2/src/cache.rs b/src/mito2/src/cache.rs index 27db7daa5ab..29fc9466bfc 100644 --- a/src/mito2/src/cache.rs +++ b/src/mito2/src/cache.rs @@ -2760,6 +2760,7 @@ mod tests { append_mode: false, filter_deleted: true, merge_mode: crate::region::options::MergeMode::LastRow, + sequence_range: None, partition_expr_version: 0, } .build(), diff --git a/src/mito2/src/cache/write_cache.rs b/src/mito2/src/cache/write_cache.rs index a85e3eb4021..3ff51b0d772 100644 --- a/src/mito2/src/cache/write_cache.rs +++ b/src/mito2/src/cache/write_cache.rs @@ -211,6 +211,11 @@ impl WriteCache { metrics: &mut Metrics, ) -> Result { let region_id = write_request.metadata.region_id; + let override_sequence = if write_request.preserve_row_sequence { + None + } else { + write_request.max_sequence + }; let store = self.file_cache.local_store(); let path_provider = WriteCachePathProvider::new(self.file_cache.clone()); @@ -248,14 +253,14 @@ impl WriteCache { writer .write_all_flat_as_primary_key( write_request.source, - write_request.max_sequence, + override_sequence, write_opts, ) .await? } crate::sst::FormatType::Flat => { writer - .write_all_flat(write_request.source, write_request.max_sequence, write_opts) + .write_all_flat(write_request.source, override_sequence, write_opts) .await? } }; @@ -553,6 +558,7 @@ mod tests { max_sequence: None, sst_write_format: Default::default(), cache_manager: Default::default(), + preserve_row_sequence: false, index_options: IndexOptions::default(), index_config: Default::default(), inverted_index_config: Default::default(), @@ -657,6 +663,7 @@ mod tests { max_sequence: None, sst_write_format: Default::default(), cache_manager: cache_manager.clone(), + preserve_row_sequence: false, index_options: IndexOptions::default(), index_config: Default::default(), inverted_index_config: Default::default(), @@ -751,6 +758,7 @@ mod tests { max_sequence: None, sst_write_format: Default::default(), cache_manager: cache_manager.clone(), + preserve_row_sequence: false, index_options: IndexOptions::default(), index_config: Default::default(), inverted_index_config: Default::default(), diff --git a/src/mito2/src/compaction/compactor.rs b/src/mito2/src/compaction/compactor.rs index 65c57ba9b79..ebe3d54b3f2 100644 --- a/src/mito2/src/compaction/compactor.rs +++ b/src/mito2/src/compaction/compactor.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::num::NonZero; +use std::num::{NonZero, NonZeroU64}; use std::sync::Arc; use std::time::Duration; @@ -52,7 +52,7 @@ use crate::region::version::VersionRef; use crate::region::{ManifestContext, RegionLeaderState, RegionRoleState}; use crate::schedule::scheduler::LocalScheduler; use crate::sst::FormatType; -use crate::sst::file::{FileMeta, UncommittedSsts}; +use crate::sst::file::{FileHandle, FileMeta, UncommittedSsts}; use crate::sst::file_purger::LocalFilePurger; use crate::sst::index::intermediate::IntermediateManager; use crate::sst::index::puffin_manager::PuffinManagerFactory; @@ -404,6 +404,22 @@ pub trait SstMerger: Send + Sync + 'static { #[derive(Clone)] pub struct DefaultSstMerger; +/// Computes the maximum sequence bound of the output of merging `inputs`. +/// +/// Returns `None` if any input's sequence bound is unknown (`None`): a partial +/// max over the known inputs would launder the unknown file's rows into a +/// trusted sequence bound, letting exact sequence-range scans silently skip or +/// wrongly filter rows from that file. Only when every input bound is known is +/// the `max` computed. +fn max_input_sequence(inputs: &[FileHandle]) -> Option { + let mut max: Option = None; + for input in inputs { + let sequence = input.meta_ref().sequence?; + max = Some(max.map_or(sequence, |current| current.max(sequence))); + } + max +} + #[async_trait::async_trait] impl SstMerger for DefaultSstMerger { async fn merge_single_output( @@ -439,12 +455,17 @@ impl SstMerger for DefaultSstMerger { .iter() .map(|f| f.file_id().to_string()) .join(","); - let max_sequence = output - .inputs - .iter() - .map(|f| f.meta_ref().sequence) - .max() - .flatten(); + let max_sequence = max_input_sequence(&output.inputs); + // The compaction output can only preserve per-row sequences when the + // region option is enabled AND every input file carries the + // preserved-sequence marker. Otherwise a legacy (unmarked) input would + // be laundered into a trusted output and exact sequence-range scans + // would silently skip or wrongly filter its normalized sequences. + let output_preserves_sequence = compaction_region.region_options.preserve_row_sequence + && output + .inputs + .iter() + .all(|f| f.meta_ref().preserve_row_sequence); let builder = CompactionSstReaderBuilder { metadata: compaction_region.region_metadata.clone(), sst_layer: compaction_region.access_layer.clone(), @@ -474,6 +495,7 @@ impl SstMerger for DefaultSstMerger { } else { FormatType::PrimaryKey }, + preserve_row_sequence: output_preserves_sequence, index_options, index_config, inverted_index_config, @@ -527,6 +549,7 @@ impl SstMerger for DefaultSstMerger { num_series: sst_info.num_series, primary_key_min, primary_key_max, + preserve_row_sequence: output_preserves_sequence, } }) .collect::>(); @@ -791,6 +814,33 @@ mod tests { FileHandle::new(meta, Arc::new(NoopFilePurger)) } + #[test] + fn test_max_input_sequence() { + let meta_with = |sequence| FileMeta { + sequence, + ..dummy_file_meta() + }; + + // All inputs with known bounds yield their max. + let inputs = vec![ + new_file_handle(meta_with(NonZeroU64::new(3))), + new_file_handle(meta_with(NonZeroU64::new(9))), + new_file_handle(meta_with(NonZeroU64::new(5))), + ]; + assert_eq!(NonZeroU64::new(9), max_input_sequence(&inputs)); + + // A single unknown bound poisons the whole output: the known inputs' + // max must not be laundered onto the merged output. + let inputs = vec![ + new_file_handle(meta_with(NonZeroU64::new(9))), + new_file_handle(meta_with(None)), + ]; + assert_eq!(None, max_input_sequence(&inputs)); + + // No inputs at all yield `None`. + assert_eq!(None, max_input_sequence(&[])); + } + /// Build a minimal [`CompactionRegion`] suitable for tests where the /// [`SstMerger`] is mocked and never touches the access layer. async fn new_test_compaction_region() -> CompactionRegion { diff --git a/src/mito2/src/compaction/window.rs b/src/mito2/src/compaction/window.rs index 422d600a577..a9f3f021b84 100644 --- a/src/mito2/src/compaction/window.rs +++ b/src/mito2/src/compaction/window.rs @@ -336,6 +336,7 @@ mod tests { max_row_group_row_count: None, primary_key_encoding: None, write_buffer_size: None, + preserve_row_sequence: false, }, compaction_time_window: None, } diff --git a/src/mito2/src/engine.rs b/src/mito2/src/engine.rs index a128531b535..265a19de2e1 100644 --- a/src/mito2/src/engine.rs +++ b/src/mito2/src/engine.rs @@ -135,8 +135,8 @@ use crate::config::MitoConfig; use crate::engine::puffin_index::{IndexEntryContext, collect_index_entries_from_puffin}; use crate::error::{ IncrementalQueryStaleSnafu, InvalidRequestSnafu, JoinSnafu, MitoManifestInfoSnafu, RecvSnafu, - RegionNotFoundSnafu, Result, SerdeJsonSnafu, SerializeColumnMetadataSnafu, - SnapshotFenceStaleSnafu, + RegionNotFoundSnafu, Result, SequenceRangeUnsupportedSnafu, SerdeJsonSnafu, + SerializeColumnMetadataSnafu, SnapshotFenceStaleSnafu, }; #[cfg(feature = "enterprise")] use crate::extension::BoxedExtensionRangeProviderFactory; @@ -147,7 +147,7 @@ use crate::metrics::{ HANDLE_REQUEST_ELAPSED, SCAN_MEMORY_EXHAUSTED_TOTAL, SCAN_MEMORY_USAGE_BYTES, SCAN_REQUESTS_REJECTED_TOTAL, }; -use crate::read::scan_region::{ScanRegion, Scanner}; +use crate::read::scan_region::{ScanRegion, Scanner, exact_sequence_range}; use crate::read::stream::ScanBatchStream; use crate::region::MitoRegionRef; use crate::region::opener::PartitionExprFetcherRef; @@ -1054,36 +1054,97 @@ impl EngineInner { request.memtable_max_sequence = Some(version_data.committed_sequence); } - if let Some(given_seq) = request.memtable_min_sequence { - let min_readable_seq = version.flushed_sequence; - ensure!( - given_seq >= min_readable_seq, - IncrementalQueryStaleSnafu { - region_id, - given_seq, - min_readable_seq, - } - ); - } + // Shared capability check used identically by the engine (fence + // relaxation) and the reader (row-level filtering) — see + // `read::scan_region::exact_sequence_range`. Only Flow's explicit + // `sequence_range` mode requests exactness; historical `memtable_only` + // scans never set `exact_sequence_range` and therefore keep every fence + // below regardless of region options. + let exact_sequence_range = exact_sequence_range(&request, &version); - if let Some(given_seq) = request.memtable_max_sequence - && !request.skip_sst_files - { - // Explicit snapshot fences that include SST reads are enforceable - // only while the requested upper bound is not older than the - // region's flushed frontier. If H has already been flushed into SST, - // mito cannot apply a memtable-only sequence upper bound to that - // SST scan, so fail and let Flow rebind the fenced repair instead - // of reading rows beyond H. - let min_enforceable_seq = version.flushed_sequence; - ensure!( - given_seq >= min_enforceable_seq, - SnapshotFenceStaleSnafu { - region_id, - given_seq, - min_enforceable_seq, + if request.exact_sequence_range { + // Exact `sequence_range` mode: when the capability is available the + // requested (C, H] delta is enforced row-level on memtables and every + // SST, so the lower/upper fences can be relaxed. When it is not, we + // must fail with a structured stale/unsupported error so Flow falls + // back instead of silently reading an approximate superset. + if exact_sequence_range.is_none() { + if let Some(given_seq) = request.memtable_min_sequence { + let min_readable_seq = version.flushed_sequence; + ensure!( + given_seq >= min_readable_seq, + IncrementalQueryStaleSnafu { + region_id, + given_seq, + min_readable_seq, + } + ); } - ); + + if let Some(given_seq) = request.memtable_max_sequence { + let min_enforceable_seq = version.flushed_sequence; + ensure!( + given_seq >= min_enforceable_seq, + SnapshotFenceStaleSnafu { + region_id, + given_seq, + min_enforceable_seq, + } + ); + } + + // Both bounds are enforceable against the flushed frontier, yet the + // region cannot serve an exact row-level delta (preserve option off, + // or a file without the preserved-sequence marker). Main semantics + // would silently return rows outside (C, H], so fail instead. + return SequenceRangeUnsupportedSnafu { + region_id, + min_seq: request.memtable_min_sequence.unwrap_or_default(), + max_seq: request.memtable_max_sequence.unwrap_or_default(), + reason: if version.options.preserve_row_sequence { + "region has files without preserved per-row sequences".to_string() + } else { + "region does not preserve per-row sequences (preserve_row_sequence is off)" + .to_string() + }, + } + .fail(); + } + } else { + // Non-exact mode (including historical `memtable_only` scans): keep + // main's fences exactly as-is. The preserve option never relaxes them + // because `exact_sequence_range` is the only explicit exact intent. + if let Some(given_seq) = request.memtable_min_sequence { + let min_readable_seq = version.flushed_sequence; + ensure!( + given_seq >= min_readable_seq, + IncrementalQueryStaleSnafu { + region_id, + given_seq, + min_readable_seq, + } + ); + } + + if let Some(given_seq) = request.memtable_max_sequence + && !request.skip_sst_files + { + // Explicit snapshot fences that include SST reads are enforceable + // only while the requested upper bound is not older than the + // region's flushed frontier. If H has already been flushed into SST, + // mito cannot apply a memtable-only sequence upper bound to that + // SST scan, so fail and let Flow rebind the fenced repair instead + // of reading rows beyond H. + let min_enforceable_seq = version.flushed_sequence; + ensure!( + given_seq >= min_enforceable_seq, + SnapshotFenceStaleSnafu { + region_id, + given_seq, + min_enforceable_seq, + } + ); + } } // Get cache. diff --git a/src/mito2/src/engine/alter_test.rs b/src/mito2/src/engine/alter_test.rs index 2461f2ca0cd..3dad61e768f 100644 --- a/src/mito2/src/engine/alter_test.rs +++ b/src/mito2/src/engine/alter_test.rs @@ -22,6 +22,7 @@ use api::v1::helper::{row, tag_column_schema}; use api::v1::value::ValueData; use api::v1::{ArrowIpc, BulkWalEntry, ColumnDataType, Row, Rows, SemanticType, Value, WalEntry}; use common_error::ext::ErrorExt; +use common_error::status_code::StatusCode; use common_meta::ddl::utils::{parse_column_metadatas, parse_manifest_infos_from_extensions}; use common_recordbatch::{DfRecordBatch, RecordBatches}; use common_test_util::flight::encode_to_flight_data; @@ -37,6 +38,7 @@ use store_api::region_engine::{RegionEngine, RegionManifestInfo, RegionRole}; use store_api::region_request::{ AddColumn, AddColumnLocation, AlterKind, PathType, RegionAlterRequest, RegionBulkInsertsRequest, RegionOpenRequest, RegionRequest, SetIndexOption, SetRegionOption, + UnsetRegionOption, }; use store_api::storage::{ColumnId, RegionId, ScanRequest}; @@ -1985,6 +1987,287 @@ async fn test_alter_region_append_mode_invalid() { check_append_mode(&engine, true); } +#[tokio::test] +async fn test_alter_region_preserve_row_sequence_lifecycle() { + common_telemetry::init_default_ut_logging(); + + let mut env = TestEnv::with_prefix("test_alter_region_preserve_row_sequence_lifecycle").await; + let engine = env.create_engine(MitoConfig::default()).await; + + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("append_mode", "true") + .build(); + let column_schemas = rows_schema(&request); + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + let alter = |kind: AlterKind| { + engine.handle_request(region_id, RegionRequest::Alter(RegionAlterRequest { kind })) + }; + + // Fill the memtable while preserve_row_sequence is disabled. + put_rows( + &engine, + region_id, + Rows { + schema: column_schemas.clone(), + rows: build_rows(0, 3), + }, + ) + .await; + let version = engine.get_region(region_id).unwrap().version(); + assert!(version.memtables.num_rows() > 0); + let flushed_sequence = version.flushed_sequence; + + // Enable alone on the non-empty memtable: applies in memory through the + // fast path, neither flushing nor losing rows. + alter(AlterKind::SetRegionOptions { + options: vec![SetRegionOption::PreserveRowSequence(true)], + }) + .await + .unwrap(); + let version = engine.get_region(region_id).unwrap().version(); + assert!(version.options.preserve_row_sequence); + assert_eq!(0, version.ssts.levels()[0].files.len()); + assert!(version.memtables.num_rows() > 0); + assert_eq!(flushed_sequence, version.flushed_sequence); + + // Rows written after the ALTER share the pre-ALTER memtable; a later flush + // writes both into one SST marked with preserve_row_sequence. + put_rows( + &engine, + region_id, + Rows { + schema: column_schemas.clone(), + rows: build_rows(3, 5), + }, + ) + .await; + flush_region(&engine, region_id, None).await; + let version = engine.get_region(region_id).unwrap().version(); + assert!(version.options.preserve_row_sequence); + let level0 = &version.ssts.levels()[0].files; + assert_eq!(1, level0.len()); + assert!( + level0 + .values() + .next() + .unwrap() + .meta_ref() + .preserve_row_sequence + ); + let flushed_sequence = version.flushed_sequence; + + // Exact scan (2, 5] returns pre-ALTER row "2" (seq 3) and post-ALTER rows + // "3" and "4" (seqs 4 and 5): distinct row sequences survive the flush. + let scan_exact = async |min: Option, max: Option| -> String { + let stream = engine + .scan_to_stream( + region_id, + ScanRequest { + memtable_min_sequence: min, + memtable_max_sequence: max, + exact_sequence_range: true, + ..Default::default() + }, + ) + .await + .unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + sort_batches_and_print(&batches, &["tag_0", "ts"]) + }; + let expected = "\ ++-------+---------+---------------------+ +| tag_0 | field_0 | ts | ++-------+---------+---------------------+ +| 2 | 2.0 | 1970-01-01T00:00:02 | +| 3 | 3.0 | 1970-01-01T00:00:03 | +| 4 | 4.0 | 1970-01-01T00:00:04 | ++-------+---------+---------------------+"; + assert_eq!(expected, scan_exact(Some(2), Some(5)).await); + + // Fill the memtable again and unset on the non-empty memtable: applies in + // memory without creating another SST. + put_rows( + &engine, + region_id, + Rows { + schema: column_schemas, + rows: build_rows(5, 7), + }, + ) + .await; + alter(AlterKind::UnsetRegionOptions { + keys: vec![UnsetRegionOption::PreserveRowSequence], + }) + .await + .unwrap(); + let version = engine.get_region(region_id).unwrap().version(); + assert!(!version.options.preserve_row_sequence); + assert_eq!(1, version.ssts.levels()[0].files.len()); + assert!(version.memtables.num_rows() > 0); + + // The exact capability is unsupported immediately: both bounds are still + // enforceable against the flushed frontier, yet row-level filtering is off. + let err = engine + .scanner( + region_id, + ScanRequest { + memtable_min_sequence: Some(flushed_sequence), + memtable_max_sequence: Some(flushed_sequence), + exact_sequence_range: true, + ..Default::default() + }, + ) + .await + .err() + .expect("expected sequence-range unsupported error"); + assert!(matches!(err, error::Error::SequenceRangeUnsupported { .. })); + + // A subsequent flush under the disabled state writes an unmarked SST while + // the earlier marked SST stays untouched. + flush_region(&engine, region_id, None).await; + let version = engine.get_region(region_id).unwrap().version(); + let level0 = &version.ssts.levels()[0].files; + assert_eq!(2, level0.len()); + let newest = level0 + .values() + .max_by_key(|f| f.meta_ref().sequence) + .unwrap(); + assert!(!newest.meta_ref().preserve_row_sequence); + assert_eq!( + 1, + level0 + .values() + .filter(|f| f.meta_ref().preserve_row_sequence) + .count() + ); +} + +#[tokio::test] +async fn test_alter_region_append_mode_preserve_combined_flushes_in_both_orders() { + common_telemetry::init_default_ut_logging(); + + let mut env = TestEnv::with_prefix( + "test_alter_region_append_mode_preserve_combined_flushes_in_both_orders", + ) + .await; + let engine = env.create_engine(MitoConfig::default()).await; + + // Both orders of the combined ALTER must behave identically: append_mode's + // requirement for an empty memtable is not bypassed by preserve_row_sequence. + for (region_id, options) in [ + ( + RegionId::new(1, 1), + vec![ + SetRegionOption::AppendMode(true), + SetRegionOption::PreserveRowSequence(true), + ], + ), + ( + RegionId::new(1, 2), + vec![ + SetRegionOption::PreserveRowSequence(true), + SetRegionOption::AppendMode(true), + ], + ), + ] { + let request = CreateRequestBuilder::new().build(); + let column_schemas = rows_schema(&request); + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + // Non-empty memtable: the combined ALTER must flush before applying. + put_rows( + &engine, + region_id, + Rows { + schema: column_schemas, + rows: build_rows(0, 3), + }, + ) + .await; + + engine + .handle_request( + region_id, + RegionRequest::Alter(RegionAlterRequest { + kind: AlterKind::SetRegionOptions { options }, + }), + ) + .await + .unwrap(); + + let version = engine.get_region(region_id).unwrap().version(); + assert!(version.options.append_mode && version.options.preserve_row_sequence); + // The append_mode change forced a flush: the memtable is now empty and + // the pre-ALTER rows live in the flushed SST. + assert_eq!(0, version.memtables.num_rows()); + + let request = ScanRequest::default(); + let stream = engine.scan_to_stream(region_id, request).await.unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + assert_eq!(3, batches.iter().map(|b| b.num_rows()).sum::()); + } +} + +#[tokio::test] +async fn test_alter_region_preserve_row_sequence_requires_append_mode() { + common_telemetry::init_default_ut_logging(); + + let mut env = + TestEnv::with_prefix("test_alter_region_preserve_row_sequence_requires_append_mode").await; + let engine = env.create_engine(MitoConfig::default()).await; + + let region_id = RegionId::new(1, 1); + engine + .handle_request( + region_id, + RegionRequest::Create(CreateRequestBuilder::new().build()), + ) + .await + .unwrap(); + + let alter = |options: Vec| { + engine.handle_request( + region_id, + RegionRequest::Alter(RegionAlterRequest { + kind: AlterKind::SetRegionOptions { options }, + }), + ) + }; + + let err = alter(vec![SetRegionOption::PreserveRowSequence(true)]) + .await + .unwrap_err(); + let err = err.as_any().downcast_ref::().unwrap(); + assert_matches!( + err, + error::Error::InvalidMetadata { source, .. } + if source.to_string().contains("preserve_row_sequence is only supported for append-only tables") + ); + assert_eq!(err.status_code(), StatusCode::InvalidArguments); + + // Disabling on a non-append region is a no-op success and leaves the + // option off. + alter(vec![SetRegionOption::PreserveRowSequence(false)]) + .await + .unwrap(); + assert!( + !engine + .get_region(region_id) + .unwrap() + .version() + .options + .preserve_row_sequence + ); +} + /// Builds a batch against the schema before [add_nullable_field1] /// (schema version 0): `(tag_0, field_0, ts)`. fn build_schema_v0_batch() -> DfRecordBatch { diff --git a/src/mito2/src/engine/edit_region_test.rs b/src/mito2/src/engine/edit_region_test.rs index 5aa514de436..992060d75d7 100644 --- a/src/mito2/src/engine/edit_region_test.rs +++ b/src/mito2/src/engine/edit_region_test.rs @@ -563,6 +563,42 @@ async fn test_region_edit_with_file_sequence_is_not_merged() { assert_eq!(Some(3), region_file_sequence(®ion, third_file_id)); } +#[tokio::test] +async fn test_region_edit_clears_preserve_row_sequence() { + let mut env = TestEnv::new().await; + let (engine, _) = create_engine_with_request_listener(&mut env).await; + + let region_id = RegionId::new(1, 1); + engine + .handle_request( + region_id, + RegionRequest::Create(CreateRequestBuilder::new().build()), + ) + .await + .unwrap(); + let region = engine.get_region(region_id).unwrap(); + + let file_id = FileId::random(); + let mut edit = test_region_edit(region.region_id, file_id); + // The caller claims the file preserves per-row sequences... + edit.files_to_add[0].preserve_row_sequence = true; + + engine.edit_region(region.region_id, edit).await.unwrap(); + + // ...but a generic region edit assigns a new destination sequence domain + // without proving or rewriting the physical per-row sequence column, so the + // marker must be cleared while the assigned sequence stays committed + 1. + let version = region.version(); + let file = version.ssts.levels()[0] + .files + .iter() + .find(|(id, _)| **id == file_id) + .unwrap() + .1; + assert!(!file.meta_ref().preserve_row_sequence); + assert_eq!(Some(1), file.meta_ref().sequence.map(|s| s.get())); +} + async fn wait_until_region_is_in_editing(region: &MitoRegionRef) { tokio::time::timeout(Duration::from_secs(3), async { while region.state() != RegionRoleState::Leader(RegionLeaderState::Editing) { diff --git a/src/mito2/src/engine/scan_test.rs b/src/mito2/src/engine/scan_test.rs index b9eed37cb30..a087b776305 100644 --- a/src/mito2/src/engine/scan_test.rs +++ b/src/mito2/src/engine/scan_test.rs @@ -13,33 +13,47 @@ // limitations under the License. use std::collections::{BTreeMap, HashMap}; +use std::sync::Arc; use api::helper::encode_json_value; -use api::v1::Rows; use api::v1::helper::row; use api::v1::value::ValueData; +use api::v1::{ArrowIpc, Rows}; use common_base::readable_size::ReadableSize; use common_error::ext::{ErrorExt, WhateverResult}; use common_error::status_code::StatusCode; -use common_recordbatch::RecordBatches; +use common_recordbatch::{DfRecordBatch, RecordBatches}; +use common_test_util::flight::encode_to_flight_data; +use common_time::Timestamp; use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet; use datafusion_common::ScalarValue; use datafusion_expr::{col, lit}; -use datatypes::arrow::array::AsArray; -use datatypes::arrow::datatypes::{Float64Type, TimestampMillisecondType}; +use datatypes::arrow::array::{ + ArrayRef, AsArray, Float64Array, StringArray, TimestampMillisecondArray, +}; +use datatypes::arrow::datatypes::{ + DataType, Field, Float64Type, Schema, TimeUnit, TimestampMillisecondType, +}; use datatypes::json::value::JsonValue; use datatypes::prelude::ConcreteDataType; use datatypes::types::json_type::{JsonNativeType, JsonObjectType}; use futures::TryStreamExt; use serde_json::json; use store_api::region_engine::{PrepareRequest, RegionEngine, RegionScanner}; -use store_api::region_request::RegionRequest; -use store_api::storage::{RegionId, ScanRequest, TimeSeriesDistribution}; +use store_api::region_request::{ + AlterKind, RegionAlterRequest, RegionBulkInsertsRequest, RegionCompactRequest, RegionRequest, + SetRegionOption, +}; +use store_api::storage::{ + FileId, RegionId, ScanRequest, TimeSeriesDistribution, TimeSeriesRowSelector, +}; use crate::config::MitoConfig; use crate::error::Error; +use crate::manifest::action::RegionEdit; use crate::read::read_columns::{ReadColumn, ReadColumns}; use crate::read::scan_region::Scanner; +use crate::sst::file::FileMeta; use crate::test_util; use crate::test_util::{CreateRequestBuilder, TestEnv}; @@ -1069,3 +1083,1281 @@ async fn test_range_cache_separates_or_equality_time_filters() { batches.pretty_print().unwrap() ); } + +#[tokio::test] +async fn test_exact_sequence_read_compacted_sst_with_preserve_row_sequence() { + let mut env = TestEnv::new().await; + let engine = env.create_engine(MitoConfig::default()).await; + let region_id = RegionId::new(1, 1); + + env.get_schema_metadata_manager() + .register_region_table_info( + region_id.table_id(), + "test_table", + "test_catalog", + "test_schema", + None, + env.get_kv_backend(), + ) + .await; + + let request = CreateRequestBuilder::new() + .insert_option("compaction.type", "twcs") + .insert_option("compaction.twcs.trigger_file_num", "2") + .insert_option("append_mode", "true") + .insert_option("preserve_row_sequence", "true") + .build(); + let column_schemas = test_util::rows_schema(&request); + + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + // Flush 3 SSTs so the twcs compaction (trigger file num 2) can merge them. + for (start, end) in [(0, 3), (3, 6), (6, 9)] { + let rows = Rows { + schema: column_schemas.clone(), + rows: test_util::build_rows(start, end), + }; + test_util::put_rows(&engine, region_id, rows).await; + test_util::flush_region(&engine, region_id, None).await; + } + + let output = engine + .handle_request( + region_id, + RegionRequest::Compact(RegionCompactRequest::default()), + ) + .await + .unwrap(); + assert_eq!(output.affected_rows, 0); + + // The compacted SST still preserves per-row sequences: the exact (2, 7] range + // returns rows with sequence 3..=7 even though C and H are older than the + // flushed frontier (which would normally fail the stale fences). + let stream = engine + .scan_to_stream( + region_id, + ScanRequest { + memtable_min_sequence: Some(2), + memtable_max_sequence: Some(7), + exact_sequence_range: true, + ..Default::default() + }, + ) + .await + .unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + assert_eq!( + batches.pretty_print().unwrap(), + "\ ++-------+---------+---------------------+ +| tag_0 | field_0 | ts | ++-------+---------+---------------------+ +| 2 | 2.0 | 1970-01-01T00:00:02 | +| 3 | 3.0 | 1970-01-01T00:00:03 | +| 4 | 4.0 | 1970-01-01T00:00:04 | +| 5 | 5.0 | 1970-01-01T00:00:05 | +| 6 | 6.0 | 1970-01-01T00:00:06 | ++-------+---------+---------------------+" + ); +} + +// P0 regression: historical MemtableOnly reads must keep every fence even when +// the region preserves per-row sequences. The exact capability is only granted +// to the explicit `sequence_range` mode; a memtable-only scan whose checkpoint +// is behind the flushed frontier must stay stale instead of silently returning +// an incomplete delta. +#[tokio::test] +async fn test_exact_sequence_read_memtable_only_keeps_fences_with_preserve_option() { + let mut env = TestEnv::with_prefix("test_exact_sequence_read_memtable_only_keeps_fences").await; + let engine = env.create_engine(MitoConfig::default()).await; + + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("append_mode", "true") + .insert_option("preserve_row_sequence", "true") + .build(); + let column_schemas = test_util::rows_schema(&request); + + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + let rows = Rows { + schema: column_schemas, + rows: test_util::build_rows(0, 3), + }; + test_util::put_rows(&engine, region_id, rows).await; + test_util::flush_region(&engine, region_id, None).await; + + let err = engine + .scanner( + region_id, + ScanRequest { + memtable_min_sequence: Some(0), + memtable_max_sequence: Some(3), + skip_sst_files: true, + ..Default::default() + }, + ) + .await + .err() + .expect("expected stale cursor error for memtable-only after flush"); + assert!( + matches!(err, Error::IncrementalQueryStale { .. }), + "unexpected err: {err}" + ); +} + +// SequenceRange mode on a region without preserve_row_sequence: +// the exact capability is unavailable, both bounds are enforceable, so the +// engine must return a structured unsupported error instead of approximating. +#[tokio::test] +async fn test_exact_sequence_range_unsupported_when_option_off() { + let mut env = TestEnv::with_prefix("test_exact_sequence_range_unsupported_option_off").await; + let engine = env.create_engine(MitoConfig::default()).await; + + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("append_mode", "true") + .build(); + let column_schemas = test_util::rows_schema(&request); + + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + let rows = Rows { + schema: column_schemas, + rows: test_util::build_rows(0, 3), + }; + test_util::put_rows(&engine, region_id, rows).await; + test_util::flush_region(&engine, region_id, None).await; + + // Bounds (3, 4] are both >= flushed frontier (3), so neither fence fires; + // the region still cannot serve exact row-level filtering -> unsupported. + let err = engine + .scanner( + region_id, + ScanRequest { + memtable_min_sequence: Some(3), + memtable_max_sequence: Some(4), + exact_sequence_range: true, + ..Default::default() + }, + ) + .await + .err() + .expect("expected sequence-range unsupported error"); + assert!( + matches!(err, Error::SequenceRangeUnsupported { .. }), + "unexpected err: {err}" + ); + assert_eq!(err.status_code(), StatusCode::Unsupported); + + // A stale checkpoint still yields the structured stale error (rebind path). + let err = engine + .scanner( + region_id, + ScanRequest { + memtable_min_sequence: Some(0), + memtable_max_sequence: Some(4), + exact_sequence_range: true, + ..Default::default() + }, + ) + .await + .err() + .expect("expected stale cursor error"); + assert!( + matches!(err, Error::IncrementalQueryStale { .. }), + "unexpected err: {err}" + ); +} + +// SequenceRange mode when a legacy file without the preserved-sequence marker +// is present: exact capability unavailable -> structured unsupported error. +#[tokio::test] +async fn test_exact_sequence_range_unsupported_when_legacy_file() { + let mut env = TestEnv::with_prefix("test_exact_sequence_range_unsupported_legacy_file").await; + let engine = env.create_engine(MitoConfig::default()).await; + + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("append_mode", "true") + .insert_option("preserve_row_sequence", "true") + .build(); + let column_schemas = test_util::rows_schema(&request); + + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + let rows = Rows { + schema: column_schemas, + rows: test_util::build_rows(0, 3), + }; + test_util::put_rows(&engine, region_id, rows).await; + test_util::flush_region(&engine, region_id, None).await; + + // Inject a legacy file without the preserved-sequence marker. The region + // edit handler fills its max sequence with `committed + 1` (4), which is + // greater than C=3: the unmarked file may still contain in-range rows, so + // the exact capability is unavailable and the scan must fail with the + // structured unsupported error. (An unmarked file with an unknown max + // sequence fails closed in `files_allow_exact_sequence_range`, covered by + // the unit test.) + let edit = RegionEdit { + files_to_add: vec![FileMeta { + region_id, + file_id: FileId::random(), + time_range: ( + Timestamp::new_millisecond(0), + Timestamp::new_millisecond(1000), + ), + level: 0, + file_size: 0, + max_row_group_uncompressed_size: 0, + available_indexes: Default::default(), + indexes: vec![], + index_file_size: 0, + index_version: 0, + num_rows: 0, + num_row_groups: 0, + sequence: None, + partition_expr: None, + num_series: 0, + preserve_row_sequence: false, + ..Default::default() + }], + files_to_remove: vec![], + timestamp_ms: None, + compaction_time_window: None, + flushed_entry_id: None, + flushed_sequence: None, + committed_sequence: None, + }; + engine.edit_region(region_id, edit).await.unwrap(); + + let err = engine + .scanner( + region_id, + ScanRequest { + memtable_min_sequence: Some(3), + memtable_max_sequence: Some(4), + exact_sequence_range: true, + ..Default::default() + }, + ) + .await + .err() + .expect("expected sequence-range unsupported error"); + assert!( + matches!(err, Error::SequenceRangeUnsupported { .. }), + "unexpected err: {err}" + ); + assert_eq!(err.status_code(), StatusCode::Unsupported); +} + +// Regression for #8865: a legacy (unmarked) SST written before the preserve +// option was enabled must not permanently block exact scans. Once the option +// is enabled, an exact `(C, H]` scan whose lower bound C is at/after the +// unmarked file's max sequence is disjoint from it: the unmarked file is +// pruned at file selection (its rows must never pass through unfiltered) and +// only the new marked rows are returned, while normal scans still return +// everything. +#[tokio::test] +async fn test_exact_sequence_after_enable_with_old_unmarked_sst() { + let mut env = + TestEnv::with_prefix("test_exact_sequence_after_enable_with_old_unmarked_sst").await; + let engine = env.create_engine(MitoConfig::default()).await; + + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("append_mode", "true") + .build(); + let column_schemas = test_util::rows_schema(&request); + + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + // Old data flushed while the option was off: unmarked SST with max seq 3. + test_util::put_rows( + &engine, + region_id, + Rows { + schema: column_schemas.clone(), + rows: test_util::build_rows(0, 3), + }, + ) + .await; + test_util::flush_region(&engine, region_id, None).await; + + // Enable the preserve option; later flushes carry the marker. + engine + .handle_request( + region_id, + RegionRequest::Alter(RegionAlterRequest { + kind: AlterKind::SetRegionOptions { + options: vec![SetRegionOption::PreserveRowSequence(true)], + }, + }), + ) + .await + .unwrap(); + + // New marked data: seq 4..9 in two flushed SSTs. + for (start, end) in [(3, 6), (6, 9)] { + test_util::put_rows( + &engine, + region_id, + Rows { + schema: column_schemas.clone(), + rows: test_util::build_rows(start, end), + }, + ) + .await; + test_util::flush_region(&engine, region_id, None).await; + } + + let mut markers = engine + .get_region(region_id) + .unwrap() + .version() + .ssts + .levels() + .iter() + .flat_map(|level| level.files.values()) + .map(|file| file.meta_ref().preserve_row_sequence) + .collect::>(); + markers.sort_unstable(); + assert_eq!(vec![false, true, true], markers); + + // Exact (6, 9]: C=6 is at/after the unmarked file's max (3), so the old + // unmarked file is pruned and only the new marked rows are returned. + let stream = engine + .scan_to_stream( + region_id, + ScanRequest { + memtable_min_sequence: Some(6), + memtable_max_sequence: Some(9), + exact_sequence_range: true, + ..Default::default() + }, + ) + .await + .unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + let pretty = batches.pretty_print().unwrap(); + for tag in ["6", "7", "8"] { + assert!( + pretty.contains(&format!("| {tag} ")), + "new row {tag} missing from exact scan:\n{pretty}" + ); + } + for tag in ["0", "1", "2", "3", "4", "5"] { + assert!( + !pretty.contains(&format!("| {tag} ")), + "old row {tag} leaked into exact scan:\n{pretty}" + ); + } + assert_eq!( + 3, + batches.iter().map(|b| b.num_rows()).sum::(), + "exact (6, 9] rows:\n{pretty}" + ); + + // Normal scans still return everything. + let stream = engine + .scan_to_stream(region_id, ScanRequest::default()) + .await + .unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + assert_eq!(9, batches.iter().map(|b| b.num_rows()).sum::()); +} + +// Exact sequence-range mode must read every time-matching SST, so the legacy +// `sst_min_sequence` file-pruning hint is incompatible: it could silently skip +// a preserved file that still contains rows inside `(min, max]`. The request +// must be rejected with the structured unsupported error, not approximated. +#[tokio::test] +async fn test_exact_sequence_range_rejects_sst_min_sequence_hint() { + let mut env = + TestEnv::with_prefix("test_exact_sequence_range_rejects_sst_min_sequence_hint").await; + let engine = env.create_engine(MitoConfig::default()).await; + + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("append_mode", "true") + .insert_option("preserve_row_sequence", "true") + .build(); + let column_schemas = test_util::rows_schema(&request); + + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + let rows = Rows { + schema: column_schemas, + rows: test_util::build_rows(0, 3), + }; + test_util::put_rows(&engine, region_id, rows).await; + test_util::flush_region(&engine, region_id, None).await; + + // Non-trivial `sst_min_sequence` hint combined with an exact range: the + // pruning hint could drop the preserved file (whose max sequence is 3) + // for lower thresholds, losing in-range rows, so reject explicitly. + let err = engine + .scanner( + region_id, + ScanRequest { + memtable_min_sequence: Some(0), + memtable_max_sequence: Some(1), + sst_min_sequence: Some(2), + exact_sequence_range: true, + ..Default::default() + }, + ) + .await + .err() + .expect("expected sequence-range unsupported error"); + assert!( + matches!(err, Error::SequenceRangeUnsupported { .. }), + "unexpected err: {err}" + ); + assert_eq!(err.status_code(), StatusCode::Unsupported); + + // Without the hint the exact scan must still read the SST (no data loss). + let stream = engine + .scan_to_stream( + region_id, + ScanRequest { + memtable_min_sequence: Some(0), + memtable_max_sequence: Some(1), + exact_sequence_range: true, + ..Default::default() + }, + ) + .await + .unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + let row_count: usize = batches.iter().map(|b| b.num_rows()).sum(); + assert_eq!(1, row_count, "expected seq 1 only"); +} + +// Exact `(0, 1]` scans must not let the row-group-level `LastRow` shortcut drop +// in-range rows: a series with seq 1 at t1 and seq 2 at t2 must return seq 1 +// (t1) instead of being reduced to seq 2 (t2) and then filtered out. Non-exact +// `LastRow` scans keep their existing behavior (only the last row per series). +#[tokio::test] +async fn test_exact_sequence_read_with_last_row_selector_keeps_in_range_rows() { + let mut env = TestEnv::with_prefix("test_exact_sequence_read_with_last_row_selector").await; + let engine = env.create_engine(MitoConfig::default()).await; + + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("append_mode", "true") + .insert_option("preserve_row_sequence", "true") + .build(); + let column_schemas = test_util::rows_schema(&request); + + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + // Same series/tag: seq 1 at t1 (field 1.0), seq 2 at t2 (field 2.0). + let rows = Rows { + schema: column_schemas, + rows: test_util::build_rows_for_key("series", 1, 3, 1), + }; + test_util::put_rows(&engine, region_id, rows).await; + test_util::flush_region(&engine, region_id, None).await; + + let scan = async |request: ScanRequest| -> String { + let stream = engine.scan_to_stream(region_id, request).await.unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + batches.pretty_print().unwrap() + }; + + // Exact (0, 1] with the LastRow selector: seq 1 (t1) must survive the + // row-level sequence filter and be selected as the last row. + assert_eq!( + "\ ++--------+---------+---------------------+ +| tag_0 | field_0 | ts | ++--------+---------+---------------------+ +| series | 1.0 | 1970-01-01T00:00:01 | ++--------+---------+---------------------+", + scan(ScanRequest { + memtable_min_sequence: Some(0), + memtable_max_sequence: Some(1), + exact_sequence_range: true, + series_row_selector: Some(TimeSeriesRowSelector::LastRow), + ..Default::default() + }) + .await + ); + + // Non-exact LastRow scan (no regression): still returns only the last row + // of the series, seq 2 (t2). + assert_eq!( + "\ ++--------+---------+---------------------+ +| tag_0 | field_0 | ts | ++--------+---------+---------------------+ +| series | 2.0 | 1970-01-01T00:00:02 | ++--------+---------+---------------------+", + scan(ScanRequest { + series_row_selector: Some(TimeSeriesRowSelector::LastRow), + ..Default::default() + }) + .await + ); +} + +// Exact delta spanning both SSTs and the memtable: rows after C that are partly +// flushed and partly still in the memtable must all be returned, with the H +// watermark respected. +#[tokio::test] +async fn test_exact_sequence_read_partial_sst_partial_memtable() { + let mut env = + TestEnv::with_prefix("test_exact_sequence_read_partial_sst_partial_memtable").await; + let engine = env.create_engine(MitoConfig::default()).await; + + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("append_mode", "true") + .insert_option("preserve_row_sequence", "true") + .build(); + let column_schemas = test_util::rows_schema(&request); + + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + // seq 1..3 flushed into an SST. + let rows = Rows { + schema: column_schemas.clone(), + rows: test_util::build_rows(0, 3), + }; + test_util::put_rows(&engine, region_id, rows).await; + test_util::flush_region(&engine, region_id, None).await; + + // seq 4..6 stay in the memtable. + let rows = Rows { + schema: column_schemas, + rows: test_util::build_rows(3, 6), + }; + test_util::put_rows(&engine, region_id, rows).await; + + let scan_exact = async |min: Option, max: Option| -> String { + let stream = engine + .scan_to_stream( + region_id, + ScanRequest { + memtable_min_sequence: min, + memtable_max_sequence: max, + exact_sequence_range: true, + ..Default::default() + }, + ) + .await + .unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + batches.pretty_print().unwrap() + }; + + // The merged output order across SST + memtable sources is not guaranteed, + // and each source batch repeats the pretty-printed header row, so compare + // only the data rows (excluding header/separator lines). + let result = scan_exact(Some(2), Some(6)).await; + let mut rows = result + .lines() + .filter(|l| l.starts_with("| ") && !l.starts_with("| tag_0 ")) + .collect::>(); + rows.sort_unstable(); + assert_eq!( + vec![ + "| 2 | 2.0 | 1970-01-01T00:00:02 |", + "| 3 | 3.0 | 1970-01-01T00:00:03 |", + "| 4 | 4.0 | 1970-01-01T00:00:04 |", + "| 5 | 5.0 | 1970-01-01T00:00:05 |", + ], + rows, + "unexpected set for (2, 6]:\n{result}" + ); +} + +/// Builds a bulk insert request with rows `[start, end)` in the flat schema. +fn build_bulk_insert_request( + region_id: RegionId, + start: usize, + end: usize, +) -> RegionBulkInsertsRequest { + let schema = Arc::new(Schema::new(vec![ + Field::new("tag_0", DataType::Utf8, true), + Field::new("field_0", DataType::Float64, true), + Field::new( + "ts", + DataType::Timestamp(TimeUnit::Millisecond, None), + false, + ), + ])); + let tag = Arc::new(StringArray::from_iter_values( + (start..end).map(|value| value.to_string()), + )) as ArrayRef; + let field = Arc::new(Float64Array::from_iter_values( + (start..end).map(|value| value as f64), + )) as ArrayRef; + let ts = Arc::new(TimestampMillisecondArray::from_iter_values( + (start..end).map(|value| value as i64 * 1000), + )) as ArrayRef; + let payload = DfRecordBatch::try_new(schema, vec![tag, field, ts]).unwrap(); + let (schema, record_batch) = encode_to_flight_data(payload.clone()); + + RegionBulkInsertsRequest { + region_id, + payload, + raw_data: ArrowIpc { + schema: schema.data_header, + data_header: record_batch.data_header, + payload: record_batch.data_body, + }, + partition_expr_version: None, + aligned_schema_version: None, + } +} + +// Bulk parts fold per-part sequences (commit-unit granularity): an exact range +// keeps or drops each whole part. The same set must hold before flush, after +// flush, and after compaction. +#[tokio::test] +async fn test_exact_sequence_read_bulk_parts() { + let mut env = TestEnv::with_prefix("test_exact_sequence_read_bulk_parts").await; + let engine = env.create_engine(MitoConfig::default()).await; + + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("append_mode", "true") + .insert_option("preserve_row_sequence", "true") + .insert_option("memtable.type", "bulk") + .build(); + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + // Two bulk parts: [0,3) and [3,6). Each part carries one folded sequence. + engine + .handle_request( + region_id, + RegionRequest::BulkInserts(build_bulk_insert_request(region_id, 0, 3)), + ) + .await + .unwrap(); + engine + .handle_request( + region_id, + RegionRequest::BulkInserts(build_bulk_insert_request(region_id, 3, 6)), + ) + .await + .unwrap(); + + let scan_exact = async |min: Option, max: Option| -> String { + let stream = engine + .scan_to_stream( + region_id, + ScanRequest { + memtable_min_sequence: min, + memtable_max_sequence: max, + exact_sequence_range: true, + ..Default::default() + }, + ) + .await + .unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + batches.pretty_print().unwrap() + }; + + let expected_all = "\ ++-------+---------+---------------------+ +| tag_0 | field_0 | ts | ++-------+---------+---------------------+ +| 0 | 0.0 | 1970-01-01T00:00:00 | +| 1 | 1.0 | 1970-01-01T00:00:01 | +| 2 | 2.0 | 1970-01-01T00:00:02 | +| 3 | 3.0 | 1970-01-01T00:00:03 | +| 4 | 4.0 | 1970-01-01T00:00:04 | +| 5 | 5.0 | 1970-01-01T00:00:05 | ++-------+---------+---------------------+"; + + // A wide range covering every folded part sequence returns all rows. + assert_eq!(expected_all, scan_exact(Some(0), Some(100)).await); + + // Commit-unit granularity: a lower bound that excludes the first part's + // folded sequence drops the whole first part, keeps the second. + let scan = scan_exact(Some(2), Some(100)).await; + assert!( + !scan.contains("| 0 ") && !scan.contains("| 1 ") && !scan.contains("| 2 "), + "first bulk part should be fully excluded, got:\n{scan}" + ); + assert!(scan.contains("| 3 "), "second bulk part missing:\n{scan}"); + + test_util::flush_region(&engine, region_id, None).await; + assert_eq!(scan_exact(Some(2), Some(100)).await, scan); + + engine + .handle_request( + region_id, + RegionRequest::Compact(RegionCompactRequest::default()), + ) + .await + .unwrap(); + assert_eq!(scan_exact(Some(2), Some(100)).await, scan); +} + +// Bulk commit publication ordering: the committed sequence must never cover a +// bulk part before its rows are physically installed in the memtable. A scan +// opening in the window between ordinary-memtable handling and bulk +// installation binds H to the pre-bulk committed sequence, sees no bulk rows, +// and a follow-up exact scan over the fresh range returns the bulk rows once. +#[tokio::test] +async fn test_bulk_write_sequence_not_committed_before_install() { + let mut env = + TestEnv::with_prefix("test_bulk_write_sequence_not_committed_before_install").await; + let engine = env.create_engine(MitoConfig::default()).await; + + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("append_mode", "true") + .insert_option("preserve_row_sequence", "true") + .insert_option("memtable.type", "bulk") + .build(); + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + let committed_sequence = || { + engine + .find_region(region_id) + .unwrap() + .find_committed_sequence() + }; + assert_eq!(0, committed_sequence()); + + let mut barrier = crate::region_write_ctx::test_hooks::arm_bulk_install_barrier(region_id); + + // Start a bulk write in the background; it pauses before installation. + let engine_clone = engine.clone(); + let write_handle = tokio::spawn(async move { + engine_clone + .handle_request( + region_id, + RegionRequest::BulkInserts(build_bulk_insert_request(region_id, 0, 3)), + ) + .await + .unwrap(); + }); + + // Wait until the write paused between memtable handling and bulk install. + tokio::time::timeout( + std::time::Duration::from_secs(10), + barrier.wait_until_reached(), + ) + .await + .expect("bulk write never reached the install barrier"); + + // Open the snapshot-bound scanner while the write is still paused at the + // barrier: it must bind H to the pre-bulk committed sequence (0) because + // the bulk part is not installed yet. + let scanner = engine + .scanner( + region_id, + ScanRequest { + memtable_min_sequence: Some(0), + snapshot_on_scan: true, + exact_sequence_range: true, + ..Default::default() + }, + ) + .await + .unwrap(); + assert_eq!( + Some(0), + scanner.snapshot_sequence(), + "snapshot-bound scan must bind H to the pre-bulk committed sequence" + ); + + // The committed sequence must still be the pre-bulk value: publication + // happens strictly after the bulk part is in the memtable. + assert_eq!( + 0, + committed_sequence(), + "committed sequence leaked before the bulk part was installed" + ); + + // Release the barrier: the bulk part installs and the committed sequence + // advances to cover it. + barrier.release(); + write_handle.await.expect("bulk write should complete"); + assert_eq!(3, committed_sequence()); + + // The scanner opened while paused stays bound at the pre-bulk H and so + // sees no bulk rows. + let stream = scanner.scan().await.unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + assert_eq!( + 0, + batches.iter().map(|b| b.num_rows()).sum::(), + "bulk rows visible to a snapshot bound before installation:\n{}", + batches.pretty_print().unwrap() + ); + + // The follow-up exact scan over (pre_H, post_H] returns the bulk rows once. + let stream = engine + .scan_to_stream( + region_id, + ScanRequest { + memtable_min_sequence: Some(0), + memtable_max_sequence: Some(3), + exact_sequence_range: true, + ..Default::default() + }, + ) + .await + .unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + assert_eq!( + 3, + batches.iter().map(|b| b.num_rows()).sum::(), + "bulk rows must be returned exactly once:\n{}", + batches.pretty_print().unwrap() + ); + let pretty = batches.pretty_print().unwrap(); + for tag in ["0", "1", "2"] { + assert!( + pretty.contains(&format!("| {tag} ")), + "bulk row {tag} missing from (0, 3]:\n{pretty}" + ); + } +} + +// Compaction must not launder a legacy (unmarked) input SST into a trusted +// output: with the region option ON but an input lacking the preserved-sequence +// marker, the rewritten output must stay unmarked and exact scans must keep +// returning the structured unsupported error (never silent data). +#[tokio::test] +async fn test_compaction_output_not_laundered_from_legacy_input() { + let mut env = + TestEnv::with_prefix("test_compaction_output_not_laundered_from_legacy_input").await; + // Suppress automatic compactions (flush- and edit-triggered) so the + // explicit Compact below is the only compaction in flight (deterministic). + let engine = env + .create_engine(MitoConfig { + min_compaction_interval: std::time::Duration::from_secs(60 * 60), + schedule_compaction_after_edit: false, + ..Default::default() + }) + .await; + + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("append_mode", "true") + .insert_option("preserve_row_sequence", "true") + .insert_option("compaction.type", "twcs") + .insert_option("compaction.twcs.trigger_file_num", "2") + .build(); + let column_schemas = test_util::rows_schema(&request); + + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + // Two real marked SSTs on disk so the twcs compaction rewrites them. + for (start, end) in [(0, 3), (3, 6)] { + let rows = Rows { + schema: column_schemas.clone(), + rows: test_util::build_rows(start, end), + }; + test_util::put_rows(&engine, region_id, rows).await; + test_util::flush_region(&engine, region_id, None).await; + } + + let region = engine.get_region(region_id).unwrap(); + let version = region.version(); + let level0 = &version.ssts.levels()[0].files; + assert_eq!(2, level0.len()); + let first_file = level0 + .values() + .next() + .expect("flushed SST") + .meta_ref() + .clone(); + assert!(first_file.preserve_row_sequence); + + // Seed the first physical file as a legacy (unmarked) file: replace the + // version's FileMeta so the marker is false while the file stays on disk. + let mut unmarked = first_file.clone(); + unmarked.preserve_row_sequence = false; + engine + .edit_region( + region_id, + RegionEdit { + files_to_add: vec![unmarked], + files_to_remove: vec![], + timestamp_ms: None, + compaction_time_window: None, + flushed_entry_id: None, + flushed_sequence: None, + committed_sequence: None, + }, + ) + .await + .unwrap(); + + let region = engine.get_region(region_id).unwrap(); + let version = region.version(); + let level0 = &version.ssts.levels()[0].files; + assert_eq!(2, level0.len()); + assert!( + level0 + .values() + .any(|file| !file.meta_ref().preserve_row_sequence), + "seeded file should be unmarked" + ); + + // Compact: the rewritten output must NOT be laundered back to marked. + engine + .handle_request( + region_id, + RegionRequest::Compact(RegionCompactRequest::default()), + ) + .await + .unwrap(); + + let region = engine.get_region(region_id).unwrap(); + let version = region.version(); + let outputs = version + .ssts + .levels() + .iter() + .flat_map(|level| level.files.values()) + .collect::>(); + assert_eq!( + 1, + outputs.len(), + "two inputs should rewrite into one output file" + ); + assert!( + !outputs[0].meta_ref().preserve_row_sequence, + "legacy input must not be laundered into a marked output" + ); + + // The compacted output is unmarked. The region-edit handler filled its max + // sequence with `committed + 1` (7). Exact scans with a lower bound C + // at/after 7 are disjoint from it and served (the (7, 8] range is empty), + // while C below 7 is unsupported because the unmarked file may still + // contain in-range rows. + let stream = engine + .scan_to_stream( + region_id, + ScanRequest { + memtable_min_sequence: Some(7), + memtable_max_sequence: Some(8), + exact_sequence_range: true, + ..Default::default() + }, + ) + .await + .unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + assert_eq!( + 0, + batches.iter().map(|b| b.num_rows()).sum::(), + "exact (7, 8] must be empty:\n{}", + batches.pretty_print().unwrap() + ); + + let err = engine + .scanner( + region_id, + ScanRequest { + memtable_min_sequence: Some(6), + memtable_max_sequence: Some(7), + exact_sequence_range: true, + ..Default::default() + }, + ) + .await + .err() + .expect("expected sequence-range unsupported error for C below unmarked max"); + assert!( + matches!(err, Error::SequenceRangeUnsupported { .. }), + "unexpected err: {err}" + ); +} + +// Multi-input primary-key-format compaction: two preserved pk-format files +// rewrite into one output that still carries the marker, and an exact scan +// over the compacted output returns every row in range. +#[tokio::test] +async fn test_exact_sequence_read_pk_format_compaction_multiple_inputs() { + let mut env = + TestEnv::with_prefix("test_exact_sequence_read_pk_format_compaction_multiple_inputs").await; + let engine = env.create_engine(MitoConfig::default()).await; + + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("append_mode", "true") + .insert_option("preserve_row_sequence", "true") + .insert_option("sst_format", "primary_key") + .insert_option("compaction.type", "twcs") + .insert_option("compaction.twcs.trigger_file_num", "2") + .build(); + let column_schemas = test_util::rows_schema(&request); + + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + // Two preserved pk-format input files: seq 1..3 and seq 4..6. + for (start, end) in [(0, 3), (3, 6)] { + let rows = Rows { + schema: column_schemas.clone(), + rows: test_util::build_rows(start, end), + }; + test_util::put_rows(&engine, region_id, rows).await; + test_util::flush_region(&engine, region_id, None).await; + } + + engine + .handle_request( + region_id, + RegionRequest::Compact(RegionCompactRequest::default()), + ) + .await + .unwrap(); + + // The rewritten output still carries the preserved-sequence marker. + let region = engine.get_region(region_id).unwrap(); + let version = region.version(); + let outputs = version + .ssts + .levels() + .iter() + .flat_map(|level| level.files.values()) + .collect::>(); + assert_eq!( + 1, + outputs.len(), + "two pk inputs should rewrite into one output file" + ); + assert!(outputs[0].meta_ref().preserve_row_sequence); + + // Exact scan over the compacted pk-format output returns all rows in range. + let stream = engine + .scan_to_stream( + region_id, + ScanRequest { + memtable_min_sequence: Some(1), + memtable_max_sequence: Some(5), + exact_sequence_range: true, + ..Default::default() + }, + ) + .await + .unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + let pretty = batches.pretty_print().unwrap(); + for tag in ["1", "2", "3", "4"] { + assert!( + pretty.contains(&format!("| {tag} ")), + "row {tag} missing from compacted pk output:\n{pretty}" + ); + } + assert_eq!( + 4, + batches.iter().map(|b| b.num_rows()).sum::(), + "exact range rows:\n{pretty}" + ); + assert!( + !pretty.contains("| 0 ") && !pretty.contains("| 5 "), + "rows outside (1, 5] leaked:\n{pretty}" + ); +} + +// Exact sequence-range reads through the active production `PerSeries` series +// scan path: row-level filtering must hold across the flushed SST and the +// memtable regardless of the requested time-series distribution. +#[tokio::test] +async fn test_exact_sequence_read_series_scan_per_series() { + let mut env = TestEnv::with_prefix("test_exact_sequence_read_series_scan_per_series").await; + let engine = env + .create_engine(MitoConfig { + default_flat_format: true, + ..Default::default() + }) + .await; + + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("append_mode", "true") + .insert_option("preserve_row_sequence", "true") + .build(); + let column_schemas = test_util::rows_schema(&request); + + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + // seq 1..3 flushed into an SST, seq 4..6 stay in the memtable. + let rows = Rows { + schema: column_schemas.clone(), + rows: test_util::build_rows(0, 3), + }; + test_util::put_rows(&engine, region_id, rows).await; + test_util::flush_region(&engine, region_id, None).await; + let rows = Rows { + schema: column_schemas, + rows: test_util::build_rows(3, 6), + }; + test_util::put_rows(&engine, region_id, rows).await; + + let scan_exact_series = async |min: Option, max: Option| -> String { + let stream = engine + .scan_to_stream( + region_id, + ScanRequest { + memtable_min_sequence: min, + memtable_max_sequence: max, + exact_sequence_range: true, + distribution: Some(TimeSeriesDistribution::PerSeries), + ..Default::default() + }, + ) + .await + .unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + batches.pretty_print().unwrap() + }; + + let result = scan_exact_series(Some(2), Some(6)).await; + let mut rows = result + .lines() + .filter(|l| l.starts_with("| ") && !l.starts_with("| tag_0 ")) + .collect::>(); + rows.sort_unstable(); + assert_eq!( + vec![ + "| 2 | 2.0 | 1970-01-01T00:00:02 |", + "| 3 | 3.0 | 1970-01-01T00:00:03 |", + "| 4 | 4.0 | 1970-01-01T00:00:04 |", + "| 5 | 5.0 | 1970-01-01T00:00:05 |", + ], + rows, + "unexpected set for (2, 6] on PerSeries path:\n{result}" + ); +} + +// Range-cache fingerprint: identical files and filters with different (C, H] +// sequence ranges must never share a cache entry, otherwise the second scan +// would replay the first scan's filtered rows. +#[tokio::test] +async fn test_range_cache_key_separates_sequence_ranges() { + let mut env = TestEnv::new().await; + let engine = env + .create_engine(MitoConfig { + default_flat_format: true, + // Explicitly enable the range result cache: the sharing bug only + // reproduces when the second scan can replay the first scan's + // cached batches. + range_result_cache_size: ReadableSize::mb(64), + ..Default::default() + }) + .await; + + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("append_mode", "true") + .insert_option("preserve_row_sequence", "true") + .build(); + let column_schemas = test_util::rows_schema(&request); + + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + // Single flushed flat SST with rows seq 1..6. + test_util::put_rows( + &engine, + region_id, + Rows { + schema: column_schemas, + rows: test_util::build_rows(0, 6), + }, + ) + .await; + test_util::flush_region(&engine, region_id, None).await; + + let tag_filter = || col("tag_0").gt_eq(lit(ScalarValue::Utf8(Some("0".to_string())))); + let scan_exact = async |min: Option, max: Option| -> Vec { + let stream = engine + .scan_to_stream( + region_id, + ScanRequest { + filters: vec![tag_filter()], + memtable_min_sequence: min, + memtable_max_sequence: max, + exact_sequence_range: true, + ..Default::default() + }, + ) + .await + .unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + let mut rows = batches + .pretty_print() + .unwrap() + .lines() + .filter(|l| l.starts_with("| ") && !l.starts_with("| tag_0 ")) + .map(|l| l.to_string()) + .collect::>(); + rows.sort_unstable(); + rows + }; + + // (1, 3] -> rows with sequences 2..3: tags "1" and "2". + let first = scan_exact(Some(1), Some(3)).await; + assert_eq!( + vec![ + "| 1 | 1.0 | 1970-01-01T00:00:01 |", + "| 2 | 2.0 | 1970-01-01T00:00:02 |", + ], + first + ); + + // (3, 4] -> row with sequence 4: tag "3" only. If the cache key ignored + // the sequence range, this would replay the (1, 3] cached rows instead. + let second = scan_exact(Some(3), Some(4)).await; + assert_eq!( + vec!["| 3 | 3.0 | 1970-01-01T00:00:03 |"], + second, + "different (C, H] shared a range-cache entry" + ); +} diff --git a/src/mito2/src/error.rs b/src/mito2/src/error.rs index a484d43a318..c779ee42992 100644 --- a/src/mito2/src/error.rs +++ b/src/mito2/src/error.rs @@ -272,6 +272,22 @@ pub enum Error { location: Location, }, + #[snafu(display( + "SEQUENCE_RANGE_UNSUPPORTED: exact sequence-range read unsupported, region: {}, min_seq: {}, max_seq: {}, reason: {}, retry_hint: FALLBACK_MEMTABLE_ONLY_OR_FULL_RECOMPUTE", + region_id, + min_seq, + max_seq, + reason + ))] + SequenceRangeUnsupported { + region_id: RegionId, + min_seq: u64, + max_seq: u64, + reason: String, + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("Old manifest missing for region {}", region_id))] MissingOldManifest { region_id: RegionId, @@ -1449,6 +1465,8 @@ impl ErrorExt for Error { IncrementalQueryStale { .. } | SnapshotFenceStale { .. } => StatusCode::RequestOutdated, + SequenceRangeUnsupported { .. } => StatusCode::Unsupported, + RegionMetadataNotFound { .. } | Join { .. } | WorkerStopped { .. } diff --git a/src/mito2/src/flush.rs b/src/mito2/src/flush.rs index a4a42970e4c..dbb21f29d57 100644 --- a/src/mito2/src/flush.rs +++ b/src/mito2/src/flush.rs @@ -661,6 +661,7 @@ impl RegionFlushTask { sst_info, partition_expr.clone(), pk_range, + version.options.preserve_row_sequence, )); } if hook.is_some() { @@ -783,6 +784,7 @@ impl RegionFlushTask { sst_info: &SstInfo, partition_expr: Option, primary_key_range: Option<(Bytes, Bytes)>, + preserve_row_sequence: bool, ) -> FileMeta { let (primary_key_min, primary_key_max) = match primary_key_range { Some((min, max)) => (Some(min), Some(max)), @@ -806,6 +808,7 @@ impl RegionFlushTask { num_series: sst_info.num_series, primary_key_min, primary_key_max, + preserve_row_sequence, } } @@ -832,6 +835,7 @@ impl RegionFlushTask { } else { FormatType::PrimaryKey }, + preserve_row_sequence: version.options.preserve_row_sequence, index_options: self.index_options.clone(), index_config: self.engine_config.index.clone(), inverted_index_config: self.engine_config.inverted_index.clone(), diff --git a/src/mito2/src/read/range_cache.rs b/src/mito2/src/read/range_cache.rs index 036a574c1f5..7ad0d101d93 100644 --- a/src/mito2/src/read/range_cache.rs +++ b/src/mito2/src/read/range_cache.rs @@ -31,7 +31,7 @@ use datatypes::value::scalar_value_to_timestamp; use futures::TryStreamExt; use snafu::ResultExt; use store_api::region_engine::PartitionRange; -use store_api::storage::{FileId, RegionId, TimeSeriesRowSelector}; +use store_api::storage::{FileId, RegionId, SequenceRange, TimeSeriesRowSelector}; use table::predicate::is_string_timestamp_literal; use tokio::sync::{mpsc, oneshot}; @@ -63,6 +63,10 @@ pub(crate) struct ScanRequestFingerprint { append_mode: bool, filter_deleted: bool, merge_mode: MergeMode, + /// Exact sequence range applied row-level on SST reads, or `None` for scans + /// without row-level sequence filtering. Kept in the cache key so filtered + /// range results are never reused for scans with a different (or no) range. + sequence_range: Option, stage: RangeScanStage, /// We keep the partition expr version to ensure we won't reuse the fingerprint after we change the partition expr. /// We store the version instead of the whole partition expr or partition expr filters. @@ -86,6 +90,7 @@ pub(crate) struct ScanRequestFingerprintBuilder { pub(crate) append_mode: bool, pub(crate) filter_deleted: bool, pub(crate) merge_mode: MergeMode, + pub(crate) sequence_range: Option, pub(crate) partition_expr_version: u64, } @@ -100,6 +105,7 @@ impl ScanRequestFingerprintBuilder { append_mode, filter_deleted, merge_mode, + sequence_range, partition_expr_version, } = self; @@ -114,6 +120,7 @@ impl ScanRequestFingerprintBuilder { append_mode, filter_deleted, merge_mode, + sequence_range, stage: RangeScanStage::Data, partition_expr_version, } @@ -164,6 +171,7 @@ impl ScanRequestFingerprint { append_mode: self.append_mode, filter_deleted: self.filter_deleted, merge_mode: self.merge_mode, + sequence_range: self.sequence_range, stage: self.stage, partition_expr_version: self.partition_expr_version, } @@ -177,6 +185,7 @@ impl ScanRequestFingerprint { append_mode: self.append_mode, filter_deleted: self.filter_deleted, merge_mode: self.merge_mode, + sequence_range: self.sequence_range, stage: RangeScanStage::CandidateSeries, partition_expr_version: self.partition_expr_version, } @@ -190,6 +199,7 @@ impl ScanRequestFingerprint { append_mode: self.append_mode, filter_deleted: self.filter_deleted, merge_mode: self.merge_mode, + sequence_range: self.sequence_range, stage: RangeScanStage::SeriesData(range), partition_expr_version: self.partition_expr_version, } @@ -860,6 +870,7 @@ pub fn bench_cache_flat_range_stream( append_mode: false, filter_deleted: false, merge_mode: MergeMode::LastRow, + sequence_range: None, partition_expr_version: 0, } .build(); @@ -926,6 +937,7 @@ mod tests { append_mode: false, filter_deleted, merge_mode: MergeMode::LastRow, + sequence_range: None, partition_expr_version, } .build() diff --git a/src/mito2/src/read/scan_region.rs b/src/mito2/src/read/scan_region.rs index 250f0d7fc27..c239434ad84 100644 --- a/src/mito2/src/read/scan_region.rs +++ b/src/mito2/src/read/scan_region.rs @@ -55,7 +55,9 @@ use tokio_stream::wrappers::ReceiverStream; use crate::access_layer::AccessLayerRef; use crate::cache::CacheStrategy; use crate::config::DEFAULT_MAX_CONCURRENT_SCAN_FILES; -use crate::error::{InvalidPartitionExprSnafu, InvalidRequestSnafu, Result}; +use crate::error::{ + InvalidPartitionExprSnafu, InvalidRequestSnafu, Result, SequenceRangeUnsupportedSnafu, +}; #[cfg(feature = "enterprise")] use crate::extension::{BoxedExtensionRange, BoxedExtensionRangeProvider}; use crate::memtable::{MemtableRange, RangesOptions}; @@ -84,6 +86,7 @@ use crate::sst::index::inverted_index::applier::builder::InvertedIndexApplierBui use crate::sst::index::vector_index::applier::{VectorIndexApplier, VectorIndexApplierRef}; use crate::sst::parquet::file_range::PreFilterMode; use crate::sst::parquet::reader::ReaderMetrics; +use crate::sst::version::SstVersion; #[cfg(feature = "vector_index")] const VECTOR_INDEX_OVERFETCH_MULTIPLIER: usize = 2; @@ -415,6 +418,23 @@ impl ScanRegion { async fn scan_input(self) -> Result { let metadata = &self.version.metadata; let sst_min_sequence = self.request.sst_min_sequence.and_then(NonZeroU64::new); + // Exact `sequence_range` mode filters rows row-level on every + // time-matching SST. The legacy `sst_min_sequence` file-pruning hint + // would silently skip preserved files whose max sequence does not + // exceed it (e.g. `(5, 10]` with `sst_min_sequence = 8` drops a file + // with max = 8, losing sequences 6..8), so the conflicting combination + // is rejected explicitly instead of being silently approximated. + if sst_min_sequence.is_some() && self.exact_sequence_range().is_some() { + return SequenceRangeUnsupportedSnafu { + region_id: self.region_id(), + min_seq: self.request.memtable_min_sequence.unwrap_or_default(), + max_seq: self.request.memtable_max_sequence.unwrap_or_default(), + reason: + "sst_min_sequence pruning hint is incompatible with exact sequence-range reads" + .to_string(), + } + .fail(); + } let time_range = self.build_time_range_predicate(); let predicate = PredicateGroup::new(metadata, &self.request.filters)?; @@ -468,10 +488,27 @@ impl ScanRegion { }; let ssts = &self.version.ssts; + // Exact `(C, H]` scans skip files whose max sequence cannot reach the + // lower bound C: the whole file predates the range. This applies to + // marked files (no in-range rows) and to unmarked files already proven + // disjoint by `files_allow_exact_sequence_range` (their rows must not + // pass through unfiltered). + let sequence_range = self.exact_sequence_range(); + let exact_min_sequence = match &sequence_range { + Some(SequenceRange::GtLtEq { min, .. }) => Some(*min), + _ => None, + }; let mut files = Vec::new(); if !self.request.skip_sst_files { for level in ssts.levels() { for file in level.files.values() { + if let Some(min) = exact_min_sequence + && let Some(file_sequence) = file.meta_ref().sequence + && file_sequence.get() <= min + { + continue; + } + let exceed_min_sequence = match (sst_min_sequence, file.meta_ref().sequence) { (Some(min_sequence), Some(file_sequence)) => file_sequence > min_sequence, // If the file's sequence is None (or actually is zero), it could mean the file @@ -588,6 +625,7 @@ impl ScanRegion { .then_some(self.request.memtable_max_sequence) .flatten(), ) + .with_sequence_range(sequence_range) .with_query_stat_counters(self.query_stat_counters); #[cfg(feature = "vector_index")] let input = input @@ -715,6 +753,14 @@ impl ScanRegion { self.version.metadata.region_id } + /// Returns the exact sequence range for row-level sequence filtering, or + /// `None` when the scan is not an exact `sequence_range` request or the + /// region cannot guarantee exactness (no per-row sequences preserved, or a + /// file in the version lacks the preserved-sequence marker). + fn exact_sequence_range(&self) -> Option { + exact_sequence_range(&self.request, &self.version) + } + /// Build time range predicate from filters. fn build_time_range_predicate(&self) -> TimestampRange { let time_index = self.version.metadata.time_index_column(); @@ -933,6 +979,13 @@ pub struct ScanInput { explain_flat_format: bool, /// Snapshot upper bound bound at scan open and propagated back to the caller. pub(crate) snapshot_sequence: Option, + /// Exact sequence range to filter rows row-level on SST reads. + /// + /// Set only when the region preserves per-row sequences + /// (`preserve_row_sequence` on an append-only table) and the scan + /// carries an explicit exact range. When set, SST readers apply the same + /// `(checkpoint, upper_bound]` row-level filter as memtables. + pub(crate) sequence_range: Option, /// Whether this scan is for compaction. pub(crate) compaction: bool, /// Counters that should receive query-load metrics. @@ -973,6 +1026,7 @@ impl ScanInput { distribution: None, explain_flat_format: false, snapshot_sequence: None, + sequence_range: None, compaction: false, query_stat_counters: None, #[cfg(feature = "enterprise")] @@ -1165,6 +1219,12 @@ impl ScanInput { self } + #[must_use] + pub(crate) fn with_sequence_range(mut self, sequence_range: Option) -> Self { + self.sequence_range = sequence_range; + self + } + /// Sets whether this scan is for compaction. #[must_use] pub(crate) fn with_compaction(mut self, compaction: bool) -> Self { @@ -1583,6 +1643,52 @@ fn pre_filter_mode(append_mode: bool, merge_mode: MergeMode) -> PreFilterMode { } } +/// Returns true if every unmarked SST in `ssts` can be excluded from an exact +/// `(min_seq, max]` scan: it must carry a known max sequence not exceeding the +/// lower bound `min_seq`, so none of its rows can fall inside the range. +/// Marked files always qualify (the reader filters their rows row-level); +/// unmarked files with an unknown max sequence fail closed. +pub(crate) fn files_allow_exact_sequence_range(ssts: &SstVersion, min_seq: SequenceNumber) -> bool { + ssts.levels().iter().all(|level| { + level.files.values().all(|file| { + let meta = file.meta_ref(); + meta.preserve_row_sequence || meta.sequence.is_some_and(|seq| seq.get() <= min_seq) + }) + }) +} + +/// Single source of truth for the exact row-level sequence-range capability. +/// +/// Returns the exact `GtLtEq { min, max }` range when **all** of the following +/// hold — and only then — so the engine and the reader can never drift: +/// - the scan explicitly requests `exact_sequence_range` (Flow's +/// `sequence_range` mode; historical `memtable_only` scans never set it), and +/// - the scan does not skip SST files (exactness requires reading them), and +/// - the region preserves per-row sequences (`preserve_row_sequence`), +/// and +/// - every unmarked SST carries a known max sequence not exceeding the lower +/// bound `min`, so it cannot contribute rows to `(min, max]`. +/// +/// Returns `None` otherwise. Callers must keep main's fences/fallback semantics +/// and never silently approximate. +pub(crate) fn exact_sequence_range( + request: &ScanRequest, + version: &crate::region::version::Version, +) -> Option { + if !request.exact_sequence_range + || request.skip_sst_files + || !version.options.preserve_row_sequence + { + return None; + } + let min = request.memtable_min_sequence?; + let max = request.memtable_max_sequence?; + if !files_allow_exact_sequence_range(&version.ssts, min) { + return None; + } + Some(SequenceRange::GtLtEq { min, max }) +} + fn json_nested_paths(column_name: &str, json_type: &JsonNativeType) -> Vec { let mut paths = Vec::new(); let mut current = vec![column_name.to_string()]; @@ -1715,6 +1821,7 @@ pub(crate) fn build_scan_fingerprint(input: &ScanInput) -> Option, marked: bool| FileMeta { + sequence, + preserve_row_sequence: marked, + ..Default::default() + }; + + // Marked file (always OK) + unmarked files with known max sequences. + let mut ssts = SstVersion::new(); + ssts.add_files( + purger.clone(), + [ + file(NonZeroU64::new(5), true), + file(NonZeroU64::new(5), false), + file(NonZeroU64::new(9), false), + ] + .into_iter(), + ); + // C=5: the unmarked file with max 9 overlaps (5, H] -> not allowed. + assert!(!files_allow_exact_sequence_range(&ssts, 5)); + // C=9: every unmarked file's max <= C -> allowed. + assert!(files_allow_exact_sequence_range(&ssts, 9)); + + // Unknown max sequence fails closed even when C is large. + let mut ssts = SstVersion::new(); + ssts.add_files(purger, [file(None, false)].into_iter()); + assert!(!files_allow_exact_sequence_range(&ssts, 100)); + } } diff --git a/src/mito2/src/read/scan_util.rs b/src/mito2/src/read/scan_util.rs index 428cf22740d..2e6a1b1a3ef 100644 --- a/src/mito2/src/read/scan_util.rs +++ b/src/mito2/src/read/scan_util.rs @@ -29,9 +29,10 @@ use datatypes::timestamp::timestamp_array_to_primitive; use futures::Stream; use prometheus::IntGauge; use smallvec::SmallVec; -use store_api::storage::RegionId; +use snafu::ResultExt; +use store_api::storage::{RegionId, SequenceRange}; -use crate::error::Result; +use crate::error::{ComputeArrowSnafu, Result}; use crate::memtable::MemScanMetrics; use crate::metrics::{ IN_PROGRESS_SCAN, PRECISE_FILTER_ROWS_TOTAL, READ_BATCHES_RETURN, READ_ROW_GROUPS_TOTAL, @@ -48,7 +49,7 @@ use crate::sst::index::bloom_filter::applier::BloomFilterIndexApplyMetrics; use crate::sst::index::fulltext_index::applier::FulltextIndexApplyMetrics; use crate::sst::index::inverted_index::applier::InvertedIndexApplyMetrics; use crate::sst::parquet::file_range::{FileRange, PreFilterMode}; -use crate::sst::parquet::flat_format::time_index_column_index; +use crate::sst::parquet::flat_format::{sequence_column_index, time_index_column_index}; use crate::sst::parquet::reader::{MetadataCacheMetrics, ReaderFilterMetrics, ReaderMetrics}; use crate::sst::parquet::row_group::ParquetFetchMetrics; use crate::sst::parquet::{DEFAULT_READ_BATCH_SIZE, DEFAULT_ROW_GROUP_SIZE}; @@ -1501,6 +1502,49 @@ pub(crate) async fn scan_flat_file_ranges( )) } +/// Filters a flat-format record batch by the exact sequence range. +/// +/// Returns `None` when the entire batch is filtered out. The sequence column is +/// the second-to-last internal column of the flat format +/// (`__primary_key`, `__sequence`, `__op_type`), so it must be applied before any +/// projection/compat conversion that drops internal columns. +/// +/// `file_preserves_sequence` gates the filter per file: only files written with +/// preserved per-row sequences (the `preserve_row_sequence` marker) may be +/// row-level sequence filtered. Without the marker, rows are passed through +/// untouched to keep main's file-level semantics. +fn filter_flat_batch_by_sequence( + record_batch: RecordBatch, + sequence_range: Option, + file_preserves_sequence: bool, +) -> Result> { + let Some(sequence) = sequence_range else { + return Ok(Some(record_batch)); + }; + if !file_preserves_sequence { + return Ok(Some(record_batch)); + } + + let num_rows = record_batch.num_rows(); + if num_rows == 0 { + return Ok(Some(record_batch)); + } + let sequence_column = record_batch.column(sequence_column_index(record_batch.num_columns())); + let predicate = sequence + .filter(sequence_column) + .context(ComputeArrowSnafu)?; + let select_count = predicate.true_count(); + if select_count == 0 { + return Ok(None); + } + if select_count == num_rows { + return Ok(Some(record_batch)); + } + let filtered_batch = datatypes::arrow::compute::filter_record_batch(&record_batch, &predicate) + .context(ComputeArrowSnafu)?; + Ok(Some(filtered_batch)) +} + /// Build the stream of scanning the input [`FileRange`]s using flat reader that returns RecordBatch. #[tracing::instrument( skip_all, @@ -1527,7 +1571,19 @@ pub fn build_flat_file_range_scan_stream( let build_reader_start = Instant::now(); let Some(mut reader) = range .flat_reader( - stream_ctx.input.series_row_selector, + // In exact `sequence_range` mode the row-group-level LastRow + // shortcut would reduce each row group to its last-timestamp + // row *before* the row-level sequence filter runs, silently + // dropping in-range rows (a series with seq 1 at t1 and seq 2 + // at t2 under `(0, 1]` keeps only seq 2 and then filters it + // out). Bypass the shortcut so the final per-row selector + // (`FlatLastRowReader`, applied after source merging and the + // sequence filter) selects on the filtered rows instead. + if stream_ctx.input.sequence_range.is_some() { + None + } else { + stream_ctx.input.series_row_selector + }, fetch_metrics.as_deref(), ) .await? @@ -1548,6 +1604,14 @@ pub fn build_flat_file_range_scan_stream( record_batch }; + let Some(record_batch) = filter_flat_batch_by_sequence( + record_batch, + stream_ctx.input.sequence_range, + range.file_handle().meta_ref().preserve_row_sequence, + )? else { + continue; + }; + if let Some(flat_compat) = may_compat { let batch = flat_compat.compat(record_batch)?; yield batch; @@ -2033,3 +2097,108 @@ mod tests { assert_eq!(split_ts(&[42]), vec![vec![42]]); } } + +#[cfg(test)] +mod sequence_filter_tests { + use std::sync::Arc; + + use datatypes::arrow::array::{Int64Array, StringArray, UInt8Array, UInt64Array}; + use datatypes::arrow::datatypes::{DataType, Field, Schema}; + use datatypes::arrow::record_batch::RecordBatch; + use store_api::storage::SequenceRange; + + use super::filter_flat_batch_by_sequence; + + /// Builds a flat-format record batch: `(tag, field, ts, __primary_key, __sequence, __op_type)`. + fn batch(sequences: &[u64]) -> RecordBatch { + let schema = Arc::new(Schema::new(vec![ + Field::new("tag_0", DataType::Utf8, false), + Field::new("field_0", DataType::Int64, false), + Field::new( + "ts", + DataType::Timestamp(datatypes::arrow::datatypes::TimeUnit::Millisecond, None), + false, + ), + Field::new("__primary_key", DataType::UInt8, false), + Field::new("__sequence", DataType::UInt64, false), + Field::new("__op_type", DataType::UInt8, false), + ])); + let tags = StringArray::from_iter_values((0..sequences.len()).map(|i| i.to_string())); + let fields = Int64Array::from_iter_values(0..sequences.len() as i64); + let ts = datatypes::arrow::array::TimestampMillisecondArray::from_iter_values( + (0..sequences.len()).map(|i| i as i64 * 1000), + ); + let pk = UInt8Array::from(vec![0u8; sequences.len()]); + let seq = UInt64Array::from_iter_values(sequences.iter().copied()); + let op = UInt8Array::from(vec![0u8; sequences.len()]); + RecordBatch::try_new( + schema, + vec![ + Arc::new(tags), + Arc::new(fields), + Arc::new(ts), + Arc::new(pk), + Arc::new(seq), + Arc::new(op), + ], + ) + .unwrap() + } + + fn remaining_tags(batch: &RecordBatch) -> Vec { + batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .iter() + .map(|v| v.unwrap().to_string()) + .collect() + } + + #[test] + fn test_filter_flat_batch_by_sequence_no_range_or_legacy_file() { + let b = batch(&[1, 2, 3, 4]); + + let out = filter_flat_batch_by_sequence(b.clone(), None, true).unwrap(); + assert_eq!(remaining_tags(&out.unwrap()), vec!["0", "1", "2", "3"]); + + let out = filter_flat_batch_by_sequence( + b.clone(), + Some(SequenceRange::GtLtEq { min: 2, max: 3 }), + false, + ) + .unwrap(); + assert_eq!(remaining_tags(&out.unwrap()), vec!["0", "1", "2", "3"]); + } + + #[test] + fn test_filter_flat_batch_by_sequence_exact_range() { + let b = batch(&[1, 2, 3, 4]); + + let out = filter_flat_batch_by_sequence( + b.clone(), + Some(SequenceRange::GtLtEq { min: 2, max: 3 }), + true, + ) + .unwrap(); + assert_eq!(remaining_tags(&out.unwrap()), vec!["2"]); + + let out = filter_flat_batch_by_sequence( + b.clone(), + Some(SequenceRange::GtLtEq { min: 10, max: 20 }), + true, + ) + .unwrap(); + assert!(out.is_none()); + + let empty = batch(&[]); + let out = filter_flat_batch_by_sequence( + empty, + Some(SequenceRange::GtLtEq { min: 0, max: 10 }), + true, + ) + .unwrap(); + assert_eq!(out.unwrap().num_rows(), 0); + } +} diff --git a/src/mito2/src/region/options.rs b/src/mito2/src/region/options.rs index 9023a877113..b905568539d 100644 --- a/src/mito2/src/region/options.rs +++ b/src/mito2/src/region/options.rs @@ -118,6 +118,19 @@ pub struct RegionOptions { /// the configured size; zero disables both limits. #[serde(skip_serializing_if = "Option::is_none")] pub write_buffer_size: Option, + /// Whether to preserve per-row sequence numbers through flush and compaction. + /// + /// Only meaningful for append-only tables (`append_mode = true`): when enabled, + /// every row keeps its exact sequence number in memtables, flushed SSTs and + /// compacted SSTs, and scans with an exact `(checkpoint, upper_bound]` sequence + /// range can filter SST rows row-level instead of falling back to file-level + /// sequence metadata. + #[serde(default, skip_serializing_if = "is_false")] + pub preserve_row_sequence: bool, +} + +fn is_false(value: &bool) -> bool { + !*value } impl RegionOptions { @@ -150,6 +163,14 @@ impl RegionOptions { } ); } + if self.preserve_row_sequence { + ensure!( + self.append_mode, + InvalidRegionOptionsSnafu { + reason: "preserve_row_sequence is only supported for append-only tables (append_mode must be true)", + } + ); + } Ok(()) } @@ -286,6 +307,7 @@ impl RegionOptions { max_row_group_row_count: options.max_row_group_row_count, primary_key_encoding, write_buffer_size: options.write_buffer_size, + preserve_row_sequence: options.preserve_row_sequence, }; opts.validate()?; @@ -400,6 +422,8 @@ struct RegionOptionsWithoutEnum { sst_format: Option, #[serde_as(as = "NoneAsEmptyString")] max_row_group_row_count: Option, + #[serde_as(as = "DisplayFromStr")] + preserve_row_sequence: bool, } impl Default for RegionOptionsWithoutEnum { @@ -415,6 +439,7 @@ impl Default for RegionOptionsWithoutEnum { merge_mode: options.merge_mode, sst_format: options.sst_format, max_row_group_row_count: options.max_row_group_row_count, + preserve_row_sequence: options.preserve_row_sequence, } } } @@ -969,10 +994,34 @@ mod tests { max_row_group_row_count: None, primary_key_encoding: None, write_buffer_size: None, + preserve_row_sequence: false, }; assert_eq!(expect, options); } + #[test] + fn test_with_preserve_row_sequence() { + // Option requires append mode. + let map = make_map(&[("append_mode", "false"), ("preserve_row_sequence", "true")]); + let err = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap_err(); + assert_eq!(StatusCode::InvalidArguments, err.status_code()); + assert!(err.to_string().contains("preserve_row_sequence")); + + let map = make_map(&[("append_mode", "true"), ("preserve_row_sequence", "true")]); + let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap(); + assert!(options.append_mode); + assert!(options.preserve_row_sequence); + + // Default is false and disabled keeps main behavior. + let map = make_map(&[("append_mode", "true")]); + let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap(); + assert!(!options.preserve_row_sequence); + + let map = make_map(&[]); + let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap(); + assert_eq!(RegionOptions::default(), options); + } + #[test] fn test_region_options_serde() { let options = RegionOptions { @@ -1002,6 +1051,7 @@ mod tests { max_row_group_row_count: None, primary_key_encoding: None, write_buffer_size: Some(ReadableSize::mb(128)), + preserve_row_sequence: false, }; let region_options_json_str = serde_json::to_string(&options).unwrap(); let got: RegionOptions = serde_json::from_str(®ion_options_json_str).unwrap(); @@ -1069,10 +1119,29 @@ mod tests { max_row_group_row_count: None, primary_key_encoding: None, write_buffer_size: None, + preserve_row_sequence: false, }; assert_eq!(options, got); } + #[test] + fn test_region_options_serde_preserve_row_sequence_roundtrip() { + let options = RegionOptions { + append_mode: true, + preserve_row_sequence: true, + ..Default::default() + }; + let json = serde_json::to_string(&options).unwrap(); + assert!(json.contains("preserve_row_sequence")); + let got: RegionOptions = serde_json::from_str(&json).unwrap(); + assert_eq!(options, got); + + // Old manifests without the key default to false. + let old_json = r#"{"append_mode": true}"#; + let got: RegionOptions = serde_json::from_str(old_json).unwrap(); + assert!(!got.preserve_row_sequence); + } + #[test] fn test_max_row_group_row_count() { // Default falls back to DEFAULT_ROW_GROUP_SIZE. diff --git a/src/mito2/src/sst/file.rs b/src/mito2/src/sst/file.rs index 6fddb78a190..db1c7a89d3f 100644 --- a/src/mito2/src/sst/file.rs +++ b/src/mito2/src/sst/file.rs @@ -282,6 +282,18 @@ pub struct FileMeta { deserialize_with = "deserialize_bytes_option" )] pub primary_key_max: Option, + /// Whether the file preserves per-row sequence numbers usable for exact + /// row-level sequence filtering. + /// + /// Set when the file is written by a flush or compaction of a region with + /// `preserve_row_sequence` enabled (append-only tables). Absent + /// or `false` in legacy files, which must not be row-level sequence filtered. + #[serde(default, skip_serializing_if = "is_false")] + pub preserve_row_sequence: bool, +} + +fn is_false(value: &bool) -> bool { + !*value } impl Debug for FileMeta { @@ -1116,6 +1128,30 @@ mod tests { deserialized_file_meta.file_id, FileId::from_str("bc5896ec-e4d8-4017-a80d-f2de73188d55").unwrap() ); + assert!(!deserialized_file_meta.preserve_row_sequence); + } + + #[test] + fn test_file_meta_preserve_row_sequence_serde() { + let file_meta = FileMeta { + preserve_row_sequence: true, + ..Default::default() + }; + + let serialized = serde_json::to_string(&file_meta).unwrap(); + let value: serde_json::Value = serde_json::from_str(&serialized).unwrap(); + assert_eq!(value["preserve_row_sequence"], true); + + let deserialized: FileMeta = serde_json::from_str(&serialized).unwrap(); + assert_eq!(file_meta, deserialized); + + let file_meta_false = FileMeta { + preserve_row_sequence: false, + ..file_meta.clone() + }; + let serialized_false = serde_json::to_string(&file_meta_false).unwrap(); + let value_false: serde_json::Value = serde_json::from_str(&serialized_false).unwrap(); + assert!(value_false.get("preserve_row_sequence").is_none()); } #[test] fn test_is_index_consistent_with_region() { diff --git a/src/mito2/src/sst/index.rs b/src/mito2/src/sst/index.rs index 581d1edebdf..0b983a372d0 100644 --- a/src/mito2/src/sst/index.rs +++ b/src/mito2/src/sst/index.rs @@ -1687,6 +1687,7 @@ mod tests { max_sequence: None, sst_write_format: Default::default(), cache_manager: Default::default(), + preserve_row_sequence: false, index_options: IndexOptions::default(), index_config, inverted_index_config: Default::default(), diff --git a/src/mito2/src/worker/handle_alter.rs b/src/mito2/src/worker/handle_alter.rs index 69065436da2..40047cb67d0 100644 --- a/src/mito2/src/worker/handle_alter.rs +++ b/src/mito2/src/worker/handle_alter.rs @@ -196,7 +196,11 @@ impl RegionWorkerLoop { /// Handles requests that changes region options, like TTL. It only affects memory state /// since changes are persisted in the `DatanodeTableValue` in metasrv. /// - /// If the options require empty memtable, it only does validation. + /// Options that can be applied without an empty memtable (e.g. TTL, + /// `preserve_row_sequence`) are applied directly through the fast path. + /// If another option in the same ALTER requires an empty memtable (e.g. + /// `append_mode`), only the complete final options are staged and the + /// existing flush path is followed. /// /// Returns the staged options if they need further alteration. fn handle_alter_region_options_fast( @@ -300,6 +304,19 @@ impl RegionWorkerLoop { all_options_altered = false; } } + SetRegionOption::PreserveRowSequence(new_preserve) => { + // Toggling preserve_row_sequence only affects in-memory + // RegionOptions and how future flushes/compactions mark SST + // metadata; it does not require an empty memtable, so it + // never blocks the fast path. + if new_preserve != current_options.preserve_row_sequence { + info!( + "Update region preserve_row_sequence: {}, previous: {:?} new: {:?}", + region.region_id, current_options.preserve_row_sequence, new_preserve + ); + current_options.preserve_row_sequence = new_preserve; + } + } SetRegionOption::SkipWal => { if !current_options.skip_wal { info!("Stop writing WAL for region: {}", region.region_id); @@ -308,15 +325,33 @@ impl RegionWorkerLoop { } } } + let kind = AlterKind::SetRegionOptions { options }; + // Validate the complete final options. The loop above validates + // per-option, but a combined ALTER may flip append_mode and toggle + // preserve_row_sequence in the same request, which is only valid when + // viewed together (order-independent). Validating the staged outcome + // also guarantees that a preserve-only toggle applied via the fast path + // below cannot create an invalid state (preserve requires append_mode). + let candidate = new_region_options_on_empty_memtable(¤t_options, &kind) + .unwrap_or_else(|| current_options.clone()); + candidate.validate().map_err(|e| { + store_api::metadata::InvalidRegionRequestSnafu { + region_id: region.region_id, + err: e.to_string(), + } + .build() + })?; if all_options_altered { - region.version_control.alter_options(current_options); + // No option independently requires an empty memtable + // (preserve_row_sequence toggles apply in memory): apply the staged + // options directly without flushing. + region.version_control.alter_options(candidate); Ok(None) } else { - let kind = AlterKind::SetRegionOptions { options }; - Ok(new_region_options_on_empty_memtable( - ¤t_options, - &kind, - )) + // Some option independently requires an empty memtable (e.g. + // append_mode, sst_format or max_row_group_row_count): stage the + // complete final options and follow the existing flush path. + Ok(Some(candidate)) } } } @@ -368,6 +403,9 @@ fn new_region_options_on_empty_memtable( SetRegionOption::MaxRowGroupRowCount(new_row_count) => { current_options.max_row_group_row_count = *new_row_count; } + SetRegionOption::PreserveRowSequence(new_preserve) => { + current_options.preserve_row_sequence = *new_preserve; + } } } Some(current_options) diff --git a/src/mito2/src/worker/handle_manifest.rs b/src/mito2/src/worker/handle_manifest.rs index 700debadadd..cfa67042fbb 100644 --- a/src/mito2/src/worker/handle_manifest.rs +++ b/src/mito2/src/worker/handle_manifest.rs @@ -316,9 +316,15 @@ impl RegionWorkerLoop { let file_sequence = region.version_control.committed_sequence() + 1; edit.committed_sequence = Some(file_sequence); - // For every file added through region edit, we should fill the file sequence + // For every file added through region edit, we should fill the file sequence. + // Generic region edits (direct/import/staging) assign a new destination + // sequence domain but cannot prove or rewrite the physical per-row sequence + // column. The added files are therefore atomic/untrusted: clear the + // `preserve_row_sequence` marker so exact sequence-range scans fail closed + // until the scan reaches the assigned max sequence. for file in &mut edit.files_to_add { file.sequence = NonZeroU64::new(file_sequence); + file.preserve_row_sequence = false; } // Allow retrieving `is_staging` before spawn the edit region task. diff --git a/src/store-api/src/mito_engine_options.rs b/src/store-api/src/mito_engine_options.rs index 02ed1d39543..6890f06cc32 100644 --- a/src/store-api/src/mito_engine_options.rs +++ b/src/store-api/src/mito_engine_options.rs @@ -72,6 +72,8 @@ pub const SST_FORMAT_KEY: &str = "sst_format"; pub const MAX_ROW_GROUP_ROW_COUNT: &str = "max_row_group_row_count"; /// Upper bound for [`MAX_ROW_GROUP_ROW_COUNT`]. pub const MAX_ROW_GROUP_ROW_COUNT_LIMIT: usize = 10 * 1024 * 1024; +/// Option key for preserving per-row sequence numbers through flush and compaction. +pub const PRESERVE_ROW_SEQUENCE: &str = "preserve_row_sequence"; // Note: Adding new options here should also check if this option should be removed in [metric_engine::engine::create::region_options_for_metadata_region]. /// Returns true if the `key` is a valid option key for the mito engine. @@ -104,6 +106,7 @@ pub fn is_mito_engine_option_key(key: &str) -> bool { MERGE_MODE_KEY, SST_FORMAT_KEY, MAX_ROW_GROUP_ROW_COUNT, + PRESERVE_ROW_SEQUENCE, ] .contains(&key) } @@ -151,6 +154,7 @@ mod tests { )); assert!(is_mito_engine_option_key("append_mode")); assert!(is_mito_engine_option_key("max_row_group_row_count")); + assert!(is_mito_engine_option_key("preserve_row_sequence")); assert!(!is_mito_engine_option_key("foo")); } } diff --git a/src/store-api/src/region_request.rs b/src/store-api/src/region_request.rs index 3bb2bb916c1..e9d9f203733 100644 --- a/src/store-api/src/region_request.rs +++ b/src/store-api/src/region_request.rs @@ -56,7 +56,7 @@ use crate::metric_engine_consts::PHYSICAL_TABLE_METADATA_KEY; use crate::metrics; use crate::mito_engine_options::{ APPEND_MODE_KEY, AUTO_FLUSH_INTERVAL_KEY, MAX_ROW_GROUP_ROW_COUNT, - MAX_ROW_GROUP_ROW_COUNT_LIMIT, SKIP_WAL_KEY, SST_FORMAT_KEY, TTL_KEY, + MAX_ROW_GROUP_ROW_COUNT_LIMIT, PRESERVE_ROW_SEQUENCE, SKIP_WAL_KEY, SST_FORMAT_KEY, TTL_KEY, TWCS_MAX_OUTPUT_FILE_SIZE, TWCS_TIME_WINDOW, TWCS_TRIGGER_FILE_NUM, WRITE_BUFFER_SIZE_KEY, }; use crate::path_utils::table_dir; @@ -1458,6 +1458,8 @@ pub enum SetRegionOption { AutoFlushInterval(Option), // Modifying the max number of rows in a parquet row group. MaxRowGroupRowCount(Option), + /// Modifying the preserve per-row sequence numbers option. + PreserveRowSequence(bool), // Stops writing new WAL entries. This operation is irreversible. SkipWal, } @@ -1517,6 +1519,12 @@ impl TryFrom<&PbOption> for SetRegionOption { .ok_or_else(|| InvalidSetRegionOptionRequestSnafu { key, value }.build())?; Ok(Self::MaxRowGroupRowCount(Some(row_count))) } + PRESERVE_ROW_SEQUENCE => { + let preserve = value + .parse::() + .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?; + Ok(Self::PreserveRowSequence(preserve)) + } SKIP_WAL_KEY if value == "true" => Ok(Self::SkipWal), _ => InvalidSetRegionOptionRequestSnafu { key, value }.fail(), } @@ -1538,6 +1546,7 @@ impl From<&UnsetRegionOption> for SetRegionOption { UnsetRegionOption::Ttl => SetRegionOption::Ttl(Default::default()), UnsetRegionOption::MaxRowGroupRowCount => SetRegionOption::MaxRowGroupRowCount(None), UnsetRegionOption::WriteBufferSize => SetRegionOption::WriteBufferSize(None), + UnsetRegionOption::PreserveRowSequence => SetRegionOption::PreserveRowSequence(false), } } } @@ -1553,6 +1562,7 @@ impl TryFrom<&str> for UnsetRegionOption { TWCS_MAX_OUTPUT_FILE_SIZE => Ok(Self::TwcsMaxOutputFileSize), TWCS_TIME_WINDOW => Ok(Self::TwcsTimeWindow), MAX_ROW_GROUP_ROW_COUNT => Ok(Self::MaxRowGroupRowCount), + PRESERVE_ROW_SEQUENCE => Ok(Self::PreserveRowSequence), _ => InvalidUnsetRegionOptionRequestSnafu { key }.fail(), } } @@ -1566,6 +1576,8 @@ pub enum UnsetRegionOption { Ttl, MaxRowGroupRowCount, WriteBufferSize, + /// Unset the preserve per-row sequence numbers option. + PreserveRowSequence, } impl UnsetRegionOption { @@ -1577,6 +1589,7 @@ impl UnsetRegionOption { Self::TwcsMaxOutputFileSize => TWCS_MAX_OUTPUT_FILE_SIZE, Self::TwcsTimeWindow => TWCS_TIME_WINDOW, Self::MaxRowGroupRowCount => MAX_ROW_GROUP_ROW_COUNT, + Self::PreserveRowSequence => PRESERVE_ROW_SEQUENCE, } } } @@ -1985,6 +1998,37 @@ mod tests { ); } + #[test] + fn test_set_region_option_preserve_row_sequence_try_from() { + for (value, expected) in [("true", true), ("false", false)] { + let pb = PbOption { + key: PRESERVE_ROW_SEQUENCE.to_string(), + value: value.to_string(), + }; + assert_eq!( + SetRegionOption::PreserveRowSequence(expected), + SetRegionOption::try_from(&pb).unwrap() + ); + } + + for value in ["1", "invalid"] { + let pb = PbOption { + key: PRESERVE_ROW_SEQUENCE.to_string(), + value: value.to_string(), + }; + assert!(SetRegionOption::try_from(&pb).is_err()); + } + + assert_eq!( + UnsetRegionOption::PreserveRowSequence, + UnsetRegionOption::try_from(PRESERVE_ROW_SEQUENCE).unwrap() + ); + assert_eq!( + SetRegionOption::PreserveRowSequence(false), + (&UnsetRegionOption::PreserveRowSequence).into() + ); + } + #[test] fn test_from_proto_alter_request() { RegionAlterRequest::try_from(AlterRequest { diff --git a/src/store-api/src/storage/requests.rs b/src/store-api/src/storage/requests.rs index 7ae1bc59879..cb615ea1dfa 100644 --- a/src/store-api/src/storage/requests.rs +++ b/src/store-api/src/storage/requests.rs @@ -129,6 +129,15 @@ pub struct ScanRequest { pub skip_sst_files: bool, /// Whether to bind the effective snapshot upper bound when opening the scan. pub snapshot_on_scan: bool, + /// Explicit intent to read an exact row-level sequence delta `(min, max]` + /// across memtables and all SST files (Flow's `sequence_range` incremental + /// mode). The engine performs exact row-level filtering only when the region + /// preserves per-row sequences and every participating SST file is trusted; + /// otherwise it returns a structured stale/unsupported error so the caller + /// falls back instead of silently approximating. + /// + /// Historical `memtable_only` reads must never set this flag. + pub exact_sequence_range: bool, /// Optional hint for the distribution of time-series data. pub distribution: Option, /// Optional hint for KNN vector search. When set, the scan should use @@ -223,6 +232,9 @@ impl Display for ScanRequest { self.snapshot_on_scan )?; } + if self.exact_sequence_range { + write!(f, "{}exact_sequence_range: true", delimiter.as_str())?; + } if self.preserve_pk_dictionary_encoding { write!( f, @@ -312,11 +324,12 @@ mod tests { let request = ScanRequest { snapshot_on_scan: true, + exact_sequence_range: true, ..Default::default() }; assert_eq!( request.to_string(), - "ScanRequest { snapshot_on_scan: true }" + "ScanRequest { snapshot_on_scan: true, exact_sequence_range: true }" ); let request = ScanRequest { diff --git a/src/table/src/metadata.rs b/src/table/src/metadata.rs index 7b8958e22ce..6386fbc393e 100644 --- a/src/table/src/metadata.rs +++ b/src/table/src/metadata.rs @@ -30,7 +30,7 @@ use snafu::{OptionExt, ResultExt, ensure}; use store_api::metric_engine_consts::PHYSICAL_TABLE_METADATA_KEY; use store_api::mito_engine_options::{ APPEND_MODE_KEY, AUTO_FLUSH_INTERVAL_KEY, COMPACTION_TYPE, COMPACTION_TYPE_TWCS, - MAX_ROW_GROUP_ROW_COUNT, MERGE_MODE_KEY, SKIP_WAL_KEY, SST_FORMAT_KEY, + MAX_ROW_GROUP_ROW_COUNT, MERGE_MODE_KEY, PRESERVE_ROW_SEQUENCE, SKIP_WAL_KEY, SST_FORMAT_KEY, }; use store_api::region_request::{SetRegionOption, UnsetRegionOption}; use store_api::storage::{ColumnDescriptor, ColumnDescriptorBuilder, ColumnId}; @@ -401,6 +401,15 @@ impl TableMeta { new_options.extra_options.remove(MAX_ROW_GROUP_ROW_COUNT); } } + SetRegionOption::PreserveRowSequence(preserve) => { + if *preserve { + new_options + .extra_options + .insert(PRESERVE_ROW_SEQUENCE.to_string(), preserve.to_string()); + } else { + new_options.extra_options.remove(PRESERVE_ROW_SEQUENCE); + } + } SetRegionOption::SkipWal => { new_options.skip_wal = true; // Keep the explicit table option so it remains distinguishable @@ -1695,6 +1704,66 @@ mod tests { ); } + #[test] + fn test_set_preserve_row_sequence_option() { + let schema = Arc::new(new_test_schema()); + let meta = TableMetaBuilder::empty() + .schema(schema) + .primary_key_indices(vec![0]) + .engine("engine") + .next_column_id(3) + .build() + .unwrap(); + + let apply = |meta: &TableMeta, kind: &AlterKind| { + meta.builder_with_alter_kind("my_table", kind) + .unwrap() + .build() + .unwrap() + }; + + let with_true = apply( + &meta, + &AlterKind::SetTableOptions { + options: vec![SetRegionOption::PreserveRowSequence(true)], + }, + ); + assert_eq!( + Some("true"), + with_true + .options + .extra_options + .get(PRESERVE_ROW_SEQUENCE) + .map(String::as_str) + ); + + let set_false = apply( + &with_true, + &AlterKind::SetTableOptions { + options: vec![SetRegionOption::PreserveRowSequence(false)], + }, + ); + assert!( + !set_false + .options + .extra_options + .contains_key(PRESERVE_ROW_SEQUENCE) + ); + + let unset = apply( + &with_true, + &AlterKind::UnsetTableOptions { + keys: vec![UnsetRegionOption::PreserveRowSequence], + }, + ); + assert!( + !unset + .options + .extra_options + .contains_key(PRESERVE_ROW_SEQUENCE) + ); + } + #[test] fn test_set_skip_wal_updates_typed_and_extra_options() { let mut meta = TableMetaBuilder::empty() diff --git a/tests/compatibility/cases/downgrade_compatibility/case.toml b/tests/compatibility/cases/downgrade_compatibility/case.toml index 930a55cd042..ba5ee5065ed 100644 --- a/tests/compatibility/cases/downgrade_compatibility/case.toml +++ b/tests/compatibility/cases/downgrade_compatibility/case.toml @@ -1,8 +1,8 @@ name = "downgrade_compatibility" -reason = "Verify v1.1.4 can reopen a table whose region WAL options were written by the current binary." -introduced_by = "fix: preserve legacy region WAL options format" +reason = "Verify v1.1.4 can reopen a table whose region WAL options were written by the current binary, and can read all rows of an append-only table flushed and compacted with preserve_row_sequence enabled." +introduced_by = "fix: preserve legacy region WAL options format; preserve_row_sequence" topologies = ["distributed", "standalone"] from_range = [">=v1.2.0"] to_range = ["=v1.1.4"] -features = ["table", "wal", "downgrade"] +features = ["table", "wal", "downgrade", "append", "preserve_row_sequence"] owner = "metasrv" diff --git a/tests/compatibility/cases/downgrade_compatibility/setup.sql b/tests/compatibility/cases/downgrade_compatibility/setup.sql index 998c5025b26..a9415570d31 100644 --- a/tests/compatibility/cases/downgrade_compatibility/setup.sql +++ b/tests/compatibility/cases/downgrade_compatibility/setup.sql @@ -9,3 +9,26 @@ INSERT INTO t_downgrade_compatibility VALUES ('2024-02-09 00:01:00+0000', 'host_b', 2); ADMIN FLUSH_TABLE('t_downgrade_compatibility'); + +CREATE TABLE t_preserve_sequence_downgrade( + ts TIMESTAMP TIME INDEX, + host STRING, + val INT, + PRIMARY KEY(host) +) +ENGINE=mito +WITH(append_mode='true', preserve_row_sequence='true'); + +INSERT INTO t_preserve_sequence_downgrade VALUES +('2024-02-09 00:00:00+0000', 'host_a', 1), +('2024-02-09 00:01:00+0000', 'host_b', 2); + +ADMIN FLUSH_TABLE('t_preserve_sequence_downgrade'); + +INSERT INTO t_preserve_sequence_downgrade VALUES +('2024-02-09 00:02:00+0000', 'host_a', 3), +('2024-02-09 00:03:00+0000', 'host_c', 4); + +ADMIN FLUSH_TABLE('t_preserve_sequence_downgrade'); + +ADMIN COMPACT_TABLE('t_preserve_sequence_downgrade'); diff --git a/tests/compatibility/cases/downgrade_compatibility/verify.result b/tests/compatibility/cases/downgrade_compatibility/verify.result index e87b18b1bdf..3c052392927 100644 --- a/tests/compatibility/cases/downgrade_compatibility/verify.result +++ b/tests/compatibility/cases/downgrade_compatibility/verify.result @@ -6,3 +6,14 @@ SELECT ts, host, val FROM t_downgrade_compatibility ORDER BY ts, host; | 2024-02-09T00:00:00 | host_a | 1 | | 2024-02-09T00:01:00 | host_b | 2 | +---------------------+--------+-----+ + +SELECT ts, host, val FROM t_preserve_sequence_downgrade ORDER BY ts, host; + ++---------------------+--------+-----+ +| ts | host | val | ++---------------------+--------+-----+ +| 2024-02-09T00:00:00 | host_a | 1 | +| 2024-02-09T00:01:00 | host_b | 2 | +| 2024-02-09T00:02:00 | host_a | 3 | +| 2024-02-09T00:03:00 | host_c | 4 | ++---------------------+--------+-----+ diff --git a/tests/compatibility/cases/downgrade_compatibility/verify.sql b/tests/compatibility/cases/downgrade_compatibility/verify.sql index e369dafe669..764968bf1b8 100644 --- a/tests/compatibility/cases/downgrade_compatibility/verify.sql +++ b/tests/compatibility/cases/downgrade_compatibility/verify.sql @@ -1 +1,3 @@ SELECT ts, host, val FROM t_downgrade_compatibility ORDER BY ts, host; + +SELECT ts, host, val FROM t_preserve_sequence_downgrade ORDER BY ts, host;