rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-09-05 13:08:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: preserve row sequences and support exact sequence-range reads (#8865)

Browse Source 
* feat(mito2): support exact sequence range reads

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* test(mito2): cover preserve row sequence table alter

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(mito2): clear preserve_row_sequence marker on copy_region_from

copy_region_from copies source FileMeta into the target region, which has an
independent sequence domain. The physical per-row sequences in the copied
file belong to the source region only; trusting them in the target would let
an exact sequence-range request replay source-domain rows as if they were
target sequences. Clear the preserve_row_sequence marker on copied files so
the target fails closed with SequenceRangeUnsupported until the scan provably
cannot intersect the copied rows.

Add a regression test: copying from a preserve-enabled source into a
preserve-enabled target clears the marker, and an exact (2, 7] request on the
target returns SequenceRangeUnsupported instead of replaying source rows.

Fixes #8865

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* style: remove redundant doc comments for exact sequence range options

Approved comment-cleanup-only changes for #8865: drop outdated doc
summaries duplicated on the exact_sequence_range wrapper and the
preserve_row_sequence field, drop pure-restatement doc comments on the
SetRegionOption/UnsetRegionOption PreserveRowSequence variants, and
remove the four structural SQL comments from the alter_preserve_row_sequence
case. No behavior changes; .result regenerated by the sqlness runner.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(mito2): fail closed exact reads on copied files and extension ranges

Address review feedback on #8865:

- copy_region_from: clear the source-domain FileMeta::sequence along with
  the preserve_row_sequence marker. An unmarked file retaining a stale
  source-domain max sequence could be silently skipped by
  files_allow_exact_sequence_range() as 'proven disjoint' in the target's
  independent sequence domain, dropping rows on exact (C, H] reads. With
  sequence=None the capability check fails closed (SequenceRangeUnsupported)
  until the copied rows are provably disjoint.
- Engine/reader: reject exact sequence-range reads whenever a follower
  region has an extension range provider attached. Extension streams are
  returned without a row-level sequence filter, so exactness cannot be
  proven; treat the capability as missing (fail closed) instead of emitting
  out-of-range rows. The reader also fails closed as defense in depth.

Tests: extend copy_region_from regression to assert the copied file's
sequence hint is cleared; mito2 suite 1148/1148 passing.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* style(mito2): use doc comments for test function descriptions

Elevate the block comments describing test functions (in scan_test and
copy_region_from_test) to /// doc comments, matching the convention used
elsewhere in the exact sequence range change. No logic change.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* refactor(mito2): extract helpers and trim comment noise in exact sequence reads

PR finalization for #8865 (zero behavior change, full mito2 suite green):

- engine: extract validate_sequence_fences and
  sequence_range_unsupported_reason, keeping error variants, check order
  and reason strings identical; OSS binds the extension blocker to false.
- handle_copy_region: extract remap_copied_file_meta and
  file_descriptors_for_meta; rename file_ids -> source_file_ids and
  files_to_copy -> new_file_metas.
- compactor: rename max_input_sequence -> known_max_input_sequence,
  document the None semantics (empty input vs unknown sequence).
- Remove restating/outdated comments (ScanInput::sequence_range doc
  first line, outdated file-pruning note, options test restatements),
  compress verbatim comments while keeping why/invariants/contracts.

Verified: cargo check -p mito2 (+ --features enterprise), cargo fmt,
git diff --check, mito2 suite 1148/1148 passing.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(mito2): reject foreign-region SSTs in exact sequence reads

Reading an SST whose FileMeta.region_id differs from the scanned region
means the region's sequence domain is broken (manifest corruption or a
repartition/copy path that leaked a source-domain file). Treat this as
an explicit RegionSequenceDomainBroken error instead of silently
ignoring the file's sequence or falling back to a full scan: the region
is unusable for exact sequence-range reads until the foreign lineage is
compacted away or repaired.

- files_allow_exact_sequence_range / exact_sequence_range now return
  Result and propagate the error through engine fence validation and
  scan construction (StatusCode::Internal, distinct from the
  fallback-capable SequenceRangeUnsupported).
- Row-level flat-batch sequence filtering rejects foreign-region files
  as defense in depth.
- Engine test asserts the broken-domain error rather than
  Unsupported/fallback.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(mito2): never trust unmarked SSTs for exact-range disjoint skipping

An unmarked file's FileMeta.sequence may be synthesized by the
region-edit or repartition paths (committed+1 import barrier), not a
physical max of its rows. Treating it as a whole-file disjoint proof
could permanently skip rows that were never incrementally consumed
once the flow checkpoint passes that value.

Exact sequence-range capability now requires every SST in the region to
carry the preserve_row_sequence marker; any unmarked file disables
exactness (fallback), and the (C, H] file-selection skip also only
applies to marked files. Foreign-region files still raise
RegionSequenceDomainBroken as before.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(mito2): scope exact sequence-range capability to the time-selected read set

The exact capability check used to walk the entire SstVersion, so a
single unmarked or foreign-region SST anywhere in the region disabled
exact reads or raised RegionSequenceDomainBroken even when the
request's time range could never touch that file.

Both the engine fence and the scan builder now derive the read set with
shared time-pruning + exact-min/sst-min selection and validate
capability only over the files actually selected: a time-pruned file
cannot contribute a row to (C, H], so it cannot affect exactness. The
existing fail-loud semantics are unchanged for every selected file
(foreign region id -> RegionSequenceDomainBroken; unmarked -> exact
unavailable).

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(mito2): wash untrusted sequences in compaction and restore barrier skipping

Compaction with any non-preserved input now writes a sequence-less
output: the physical __sequence column is zeroed (the flat format
requires the internal columns) and FileMeta.sequence records the
region-local admission barrier committed_sequence + 1 (falling back to
the flushed frontier). preserve_row_sequence stays false.

Exact sequence-range scans interpret an unmarked file's sequence as an
admission barrier: barrier <= C means flow has already consumed the
whole file, so it is skipped at file level; a missing or newer barrier
fails closed. Foreign-region files stay in the selected read set so the
capability fence still raises RegionSequenceDomainBroken.

This closes the recovery loop: after a region repartition, one
time-scoped fallback consumes the migrated rows, then compaction washes
the untrusted per-row sequences away and exact incremental reads resume
via file-level barrier skipping.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* chore(mito2): drop restating comments in known_max_input_sequence tests

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* test(mito2): trim SQLness result EOF whitespace

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* refactor(mito2): reuse exact scan file selection

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* test(mito2): strengthen sequence scan coverage

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* style(mito2): trim ALTER option comments

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* test(mito2): remove no-op bulk compaction check

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(mito2): preserve trusted row sequences when reading SSTs

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(mito2): preserve target sequence domain for imported SSTs

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* test(mito2): add trailing blank line to SQLness result EOF

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* chore(mito2): trim exact sequence scan plumbing

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* refactor(mito2): fold exact SST selection checks

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* test(mito2): make legacy compaction rewrite deterministic

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* test(mito2): make PK compaction rewrite deterministic

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

---------

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-09-03 06:20:09 +00:00
committed by GitHub
parent 25d49ba092
commit 27a7047f31
37 changed files with 3757 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cmd/src/bin/query_perf_fixture/direct_sst.rs
+1
View File
@@ -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,
}
}
src/cmd/src/datanode/objbench.rs
+1
View File
@@ -208,6 +208,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(),
};
src/cmd/src/datanode/parquetbench.rs
+1
View File
@@ -356,6 +356,7 @@ impl ParquetbenchCommand {
num_series: 0,
primary_key_min: None,
primary_key_max: None,
preserve_row_sequence: false,
},
Arc::new(NoopFilePurger),
);
src/mito2/src/access_layer.rs
+9 -2
View File
@@ -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<SequenceNumber>,
pub sst_write_format: FormatType,
pub preserve_row_sequence: bool,
/// Configs for index
pub index_options: IndexOptions,
pub index_config: IndexConfig,
src/mito2/src/cache.rs
+1
View File
@@ -2773,6 +2773,7 @@ mod tests {
append_mode: false,
filter_deleted: true,
merge_mode: crate::region::options::MergeMode::LastRow,
sequence_range: None,
partition_expr_version: 0,
}
.build(),
src/mito2/src/cache/write_cache.rs
Vendored
+10 -2
View File
@@ -231,6 +231,11 @@ impl WriteCache {
metrics: &mut Metrics,
) -> Result<SstInfoArray> {
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());
@@ -268,14 +273,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?
}
};
@@ -631,6 +636,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(),
@@ -735,6 +741,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(),
@@ -829,6 +836,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(),
src/mito2/src/compaction/compactor.rs
+106 -10
View File
@@ -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::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,19 @@ pub trait SstMerger: Send + Sync + 'static {
#[derive(Clone)]
pub struct DefaultSstMerger;
/// Computes the maximum target-domain sequence bound of the output of merging
/// `inputs`. Foreign files are described by their target-local barrier; local
/// trusted files use their physical sequence bound. Unknown bounds never become
/// trusted through a partial maximum.
fn known_max_input_sequence(inputs: &[FileHandle]) -> Option<NonZeroU64> {
let mut max: Option<NonZeroU64> = 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 +452,40 @@ 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 input_max_sequence = known_max_input_sequence(&output.inputs);
// The output is trusted only when every input is trusted in the target
// domain. Foreign marker values describe the source and are ignored;
// their present FileMeta.sequence is the target-local barrier.
let output_preserves_sequence = compaction_region.region_options.preserve_row_sequence
&& output
.inputs
.iter()
.all(|f| f.is_effective_target_sequence_trusted(region_id));
let output_sequence = if output_preserves_sequence {
input_max_sequence
} else {
// The manifest records the region's latest accepted sequence for
// edit paths and its flushed frontier otherwise. Compaction only
// rewrites the immutable SST snapshot, so this is the same
// region-local admission barrier used when accepting files.
let manifest = compaction_region
.manifest_ctx
.manifest_manager
.read()
.await
.manifest();
NonZeroU64::new(
manifest
.committed_sequence
.unwrap_or(manifest.flushed_sequence)
+ 1,
)
};
// For untrusted output, keep the physical sequence override compatible
// with the main write path: use the known maximum sequence of the
// inputs. The FileMeta sequence remains the admission barrier, but it
// must not be written into the rows (or replaced with zero).
let write_max_sequence = input_max_sequence.map(NonZeroU64::get);
let builder = CompactionSstReaderBuilder {
metadata: compaction_region.region_metadata.clone(),
sst_layer: compaction_region.access_layer.clone(),
@@ -468,12 +509,13 @@ impl SstMerger for DefaultSstMerger {
source,
cache_manager: compaction_region.cache_manager.clone(),
storage,
max_sequence: max_sequence.map(NonZero::get),
max_sequence: write_max_sequence,
sst_write_format: if flat_format {
FormatType::Flat
} else {
FormatType::PrimaryKey
},
preserve_row_sequence: output_preserves_sequence,
index_options,
index_config,
inverted_index_config,
@@ -522,11 +564,12 @@ impl SstMerger for DefaultSstMerger {
index_version: 0,
num_rows: sst_info.num_rows as u64,
num_row_groups: sst_info.num_row_groups,
sequence: max_sequence,
sequence: output_sequence,
partition_expr: partition_expr.clone(),
num_series: sst_info.num_series,
primary_key_min,
primary_key_max,
preserve_row_sequence: output_preserves_sequence,
}
})
.collect::<Vec<_>>();
@@ -791,6 +834,59 @@ mod tests {
FileHandle::new(meta, Arc::new(NoopFilePurger))
}
#[test]
fn test_effective_target_trust_and_max_sequence() {
let target = RegionId::new(1, 1);
let local_trusted = new_file_handle(FileMeta {
region_id: target,
sequence: NonZeroU64::new(3),
preserve_row_sequence: true,
..dummy_file_meta()
});
let foreign_untrusted_marker = new_file_handle(FileMeta {
region_id: RegionId::new(1, 2),
sequence: NonZeroU64::new(9),
preserve_row_sequence: false,
..dummy_file_meta()
});
let local_untrusted = new_file_handle(FileMeta {
region_id: target,
sequence: NonZeroU64::new(11),
preserve_row_sequence: false,
..dummy_file_meta()
});
assert!(local_trusted.is_effective_target_sequence_trusted(target));
assert!(foreign_untrusted_marker.is_effective_target_sequence_trusted(target));
assert!(!local_untrusted.is_effective_target_sequence_trusted(target));
assert_eq!(
NonZeroU64::new(9),
known_max_input_sequence(&[local_trusted, foreign_untrusted_marker])
);
}
#[test]
fn test_known_max_input_sequence() {
let meta_with = |sequence| FileMeta {
sequence,
..dummy_file_meta()
};
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), known_max_input_sequence(&inputs));
let inputs = vec![
new_file_handle(meta_with(NonZeroU64::new(9))),
new_file_handle(meta_with(None)),
];
assert_eq!(None, known_max_input_sequence(&inputs));
assert_eq!(None, known_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 {
src/mito2/src/compaction/window.rs
+1
View File
@@ -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,
}
src/mito2/src/engine.rs
+187 -33
View File
@@ -100,7 +100,7 @@ use common_meta::error::UnexpectedSnafu;
use common_meta::key::SchemaMetadataManagerRef;
use common_recordbatch::{QueryMemoryTracker, SendableRecordBatchStream};
use common_stat::get_total_memory_bytes;
use common_telemetry::{info, tracing, warn};
use common_telemetry::{debug, info, tracing, warn};
use common_wal::options::WalOptions;
use datafusion::execution::memory_pool::{GreedyMemoryPool, MemoryPool, UnboundedMemoryPool};
use futures::future::{join_all, try_join_all};
@@ -136,8 +136,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;
@@ -148,7 +148,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;
@@ -1059,37 +1059,82 @@ 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,
// Select the files and decide the exact capability from the same version
// snapshot. Keep them together so the reader cannot recompute either
// side from a different file set.
let exact_selection = if request.exact_sequence_range {
match exact_sequence_range(&request, &version) {
Ok((files, sequence_range)) => Some((files, sequence_range)),
Err(err) => {
debug!(
"Scan region {} exact sequence range denied: min={:?}, max={:?}, denial_reason=foreign_file_missing_barrier",
region_id, request.memtable_min_sequence, request.memtable_max_sequence,
);
return Err(err);
}
);
}
}
} else {
None
};
let exact_sequence_range = exact_selection
.as_ref()
.and_then(|(_, sequence_range)| *sequence_range);
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,
}
);
}
// An extension range provider may contribute ranges on a follower
// region, and extension streams are returned without a row-level
// sequence filter (`scan_flat_extension_range`), so exactness cannot
// be proven when one is attached. Treat the capability as missing so
// the request fails closed below instead of emitting out-of-range
// rows. See the injection point in `ScanRegion::scan_input`.
#[cfg(feature = "enterprise")]
let extension_provider_blocks_exact = region.is_follower()
&& self.extension_range_provider_factory.is_some()
&& exact_sequence_range.is_some();
#[cfg(not(feature = "enterprise"))]
let extension_provider_blocks_exact = false;
let exact_sequence_range = if extension_provider_blocks_exact {
None
} else {
exact_sequence_range
};
let exact_selection = exact_selection.map(|(files, _)| (files, exact_sequence_range));
let exact_denial_reason = if !request.exact_sequence_range {
"not_requested"
} else if request.skip_sst_files {
"sst_files_skipped"
} else if request.memtable_min_sequence.is_none() {
"missing_lower_bound"
} else if request.memtable_max_sequence.is_none() {
"missing_upper_bound"
} else if !version.options.preserve_row_sequence {
"preserve_row_sequence_disabled"
} else if extension_provider_blocks_exact {
"extension_provider"
} else if exact_sequence_range.is_none() {
"selected_file_barrier_not_admitted"
} else {
"none"
};
debug!(
"Scan region {} exact sequence range: requested={}, min={:?}, max={:?}, committed={}, flushed={}, available={}, selected_files={}, denial_reason={}",
region_id,
request.exact_sequence_range,
request.memtable_min_sequence,
request.memtable_max_sequence,
version_data.committed_sequence,
version.flushed_sequence,
exact_sequence_range.is_some(),
exact_selection.as_ref().map_or(0, |(files, _)| files.len()),
exact_denial_reason,
);
validate_sequence_fences(
&request,
&version,
region_id,
exact_sequence_range.is_some(),
extension_provider_blocks_exact,
)?;
// Get cache.
let cache_manager = self.workers.cache_manager();
@@ -1108,6 +1153,11 @@ impl EngineInner {
.with_ignore_fulltext_index(self.config.fulltext_index.apply_on_query.disabled())
.with_ignore_bloom_filter(self.config.bloom_filter_index.apply_on_query.disabled())
.with_start_time(query_start);
let scan_region = if let Some(selection) = exact_selection {
scan_region.with_exact_selection(selection)
} else {
scan_region
};
#[cfg(feature = "enterprise")]
let scan_region = self.maybe_fill_extension_range_provider(scan_region, region);
@@ -1196,6 +1246,110 @@ impl EngineInner {
}
}
fn validate_sequence_fences(
request: &ScanRequest,
version: &crate::region::version::Version,
region_id: RegionId,
exact_sequence_range_available: bool,
extension_provider_blocks_exact: bool,
) -> Result<()> {
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_available {
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,
// a file without the preserved-sequence marker, or a follower with
// an extension range provider). 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: sequence_range_unsupported_reason(version, extension_provider_blocks_exact),
}
.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,
}
);
}
}
Ok(())
}
fn sequence_range_unsupported_reason(
version: &crate::region::version::Version,
extension_provider_blocks_exact: bool,
) -> String {
if extension_provider_blocks_exact {
"region is a follower with an extension range provider, whose streams cannot be filtered by sequence".to_string()
} else 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()
}
}
fn map_batch_responses(responses: Vec<(RegionId, Result<AffectedRows>)>) -> BatchResponses {
responses
.into_iter()
src/mito2/src/engine/alter_test.rs
+283 -1
View File
@@ -25,6 +25,7 @@ use api::v1::{
ArrowIpc, BulkWalEntry, ColumnDataType, Row, Rows, SemanticType, Value, WalEntry, WriteHint,
};
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;
@@ -42,7 +43,7 @@ use store_api::region_engine::{RegionEngine, RegionManifestInfo, RegionRole};
use store_api::region_request::{
AddColumn, AddColumnLocation, AlterKind, ModifyColumnType, PathType, RegionAlterRequest,
RegionBulkInsertsRequest, RegionCompactRequest, RegionOpenRequest, RegionPutRequest,
RegionRequest, SetIndexOption, SetRegionOption,
RegionRequest, SetIndexOption, SetRegionOption, UnsetRegionOption,
};
use store_api::storage::consts::PRIMARY_KEY_COLUMN_NAME;
use store_api::storage::{ColumnId, RegionId, ScanRequest};
@@ -2613,6 +2614,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<u64>, max: Option<u64>| -> 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::<usize>());
}
}
#[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<SetRegionOption>| {
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::<error::Error>().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 {
src/mito2/src/engine/apply_staging_manifest_test.rs
+106 -10
View File
@@ -20,7 +20,7 @@ use api::v1::Rows;
use api::v1::region::{StrictWindow, compact_request};
use common_function::utils::partition_expr_version;
use common_recordbatch::RecordBatches;
use datatypes::arrow::array::AsArray;
use datatypes::arrow::array::{AsArray, UInt64Array};
use datatypes::arrow::datatypes::Float64Type;
use datatypes::value::Value;
use partition::expr::{PartitionExpr, col};
@@ -54,11 +54,16 @@ fn range_expr(col_name: &str, start: i64, end: i64) -> PartitionExpr {
#[tokio::test]
async fn test_apply_staging_manifest_sequence_domain() {
common_telemetry::init_default_ut_logging();
test_apply_staging_manifest_sequence_domain_with_format(false).await;
test_apply_staging_manifest_sequence_domain_with_format(true).await;
for flat_format in [false, true] {
test_apply_staging_manifest_sequence_domain_with_format(flat_format, false).await;
test_apply_staging_manifest_sequence_domain_with_format(flat_format, true).await;
}
}
async fn test_apply_staging_manifest_sequence_domain_with_format(flat_format: bool) {
async fn test_apply_staging_manifest_sequence_domain_with_format(
flat_format: bool,
preserve_row_sequence: bool,
) {
let mut env = TestEnv::with_prefix("apply-staging-sequence-domain").await;
let engine = env
.create_engine(MitoConfig {
@@ -68,23 +73,41 @@ async fn test_apply_staging_manifest_sequence_domain_with_format(flat_format: bo
.await;
let source = RegionId::new(1, 1);
let target = RegionId::new(1, 2);
let request = CreateRequestBuilder::new().build();
let mut request_builder = CreateRequestBuilder::new();
if preserve_row_sequence {
request_builder = request_builder
.insert_option("append_mode", "true")
.insert_option("preserve_row_sequence", "true");
}
let request = request_builder.build();
let schema = rows_schema(&request);
engine
.handle_request(source, RegionRequest::Create(request.clone()))
.await
.unwrap();
for value in 0..3 {
if preserve_row_sequence {
put_rows(
&engine,
source,
Rows {
schema: schema.clone(),
rows: build_rows_for_key("0", 0, 1, value),
rows: build_rows_for_key("0", 0, 3, 0),
},
)
.await;
} else {
for value in 0..3 {
put_rows(
&engine,
source,
Rows {
schema: schema.clone(),
rows: build_rows_for_key("0", 0, 1, value),
},
)
.await;
}
}
engine
.handle_request(source, RegionRequest::Flush(RegionFlushRequest::default()))
@@ -165,11 +188,18 @@ async fn test_apply_staging_manifest_sequence_domain_with_format(flat_format: bo
target,
Rows {
schema: schema.clone(),
rows: build_rows_for_key("0", 0, 1, 99),
rows: build_rows_for_key(
if preserve_row_sequence { "1" } else { "0" },
if preserve_row_sequence { 3 } else { 0 },
if preserve_row_sequence { 4 } else { 1 },
99,
),
},
)
.await;
assert_target_value(&engine, target, 99.0).await;
if !preserve_row_sequence {
assert_target_value(&engine, target, 99.0).await;
}
engine
.handle_request(target, RegionRequest::Flush(RegionFlushRequest::default()))
@@ -216,7 +246,73 @@ async fn test_apply_staging_manifest_sequence_domain_with_format(flat_format: bo
input_ids.iter().all(|id| !output_ids.contains(id)),
"real compaction must replace both input SSTs"
);
assert_target_value(&engine, target, 99.0).await;
if !preserve_row_sequence {
assert_target_value(&engine, target, 99.0).await;
}
if preserve_row_sequence {
let output = output_files
.iter()
.find(|file| output_ids.contains(&file.meta_ref().file_id))
.expect("target-owned compaction output");
assert!(output.meta_ref().preserve_row_sequence);
// The imported rows are virtualized at target barrier 1 and the local
// row has physical sequence 2, so the output frontier is 2.
assert_eq!(
Some(std::num::NonZeroU64::new(2).unwrap()),
output.meta_ref().sequence
);
let mut reader = engine
.get_region(target)
.unwrap()
.access_layer
.read_sst((**output).clone())
.build()
.await
.unwrap()
.expect("compaction output reader");
let batch = reader
.next_record_batch()
.await
.unwrap()
.expect("output batch");
let sequences = batch
.column(batch.num_columns() - 2)
.as_any()
.downcast_ref::<UInt64Array>()
.expect("sequence column");
// The target partition excludes source field_0 = 0, leaving two
// imported rows at barrier 1 and the local row at sequence 2.
assert_eq!(
2,
sequences.values().iter().filter(|&&seq| seq == 1).count()
);
assert_eq!(
1,
sequences.values().iter().filter(|&&seq| seq == 2).count()
);
for (min, max, expected_rows) in [(0, 1, 2), (1, 2, 1)] {
let stream = engine
.scan_to_stream(
target,
ScanRequest {
exact_sequence_range: true,
memtable_min_sequence: Some(min),
memtable_max_sequence: Some(max),
..Default::default()
},
)
.await
.unwrap();
let batches = RecordBatches::try_collect(stream).await.unwrap();
assert_eq!(
expected_rows,
batches.iter().map(|batch| batch.num_rows()).sum::<usize>()
);
}
}
}
fn current_file_ids(engine: &crate::engine::MitoEngine, region_id: RegionId) -> HashSet<FileId> {
src/mito2/src/engine/copy_region_from_test.rs
+145 -1
View File
@@ -18,10 +18,11 @@ use std::{assert_matches, fs};
use api::v1::Rows;
use common_error::ext::ErrorExt;
use common_error::status_code::StatusCode;
use common_recordbatch::RecordBatches;
use object_store::layers::mock::{Error as MockError, ErrorKind, MockLayerBuilder};
use store_api::region_engine::{MitoCopyRegionFromRequest, RegionEngine, RegionRole};
use store_api::region_request::{RegionFlushRequest, RegionRequest};
use store_api::storage::RegionId;
use store_api::storage::{RegionId, ScanRequest};
use crate::config::MitoConfig;
use crate::error::Error;
@@ -353,3 +354,146 @@ async fn test_engine_copy_region_unexpected_state_with_format(flat_format: bool)
Error::RegionState { .. }
)
}
/// Regression for #8865: `copy_region_from` must not carry the source region's
/// `FileMeta::preserve_row_sequence` trust marker into the target region, whose
/// sequence domain is independent. The copied file's physical per-row sequences
/// belong to the source region only; trusting them in the target would let an
/// exact sequence-range request replay source rows as if they were target
/// sequences. The marker must be cleared so the target fails closed with
/// `SequenceRangeUnsupported` instead.
#[tokio::test]
async fn test_copy_region_from_clears_preserve_row_sequence_marker() {
common_telemetry::init_default_ut_logging();
let mut env = TestEnv::with_prefix("copy-region-from-preserve-sequence").await;
let engine = env.create_engine(MitoConfig::default()).await;
// Source and target regions both preserve per-row sequences.
let source_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 = rows_schema(&request);
engine
.handle_request(source_region_id, RegionRequest::Create(request.clone()))
.await
.unwrap();
let rows = Rows {
schema: column_schemas,
rows: build_rows(0, 42),
};
put_rows(&engine, source_region_id, rows).await;
engine
.handle_request(
source_region_id,
RegionRequest::Flush(RegionFlushRequest::default()),
)
.await
.unwrap();
// The flushed source file keeps the trust marker: the source region's own
// sequence domain is intact.
let source_manifest = engine
.get_region(source_region_id)
.unwrap()
.manifest_ctx
.manifest()
.await;
assert_eq!(1, source_manifest.files.len());
assert!(
source_manifest
.files
.values()
.all(|meta| meta.preserve_row_sequence),
"source files must keep the preserve_row_sequence marker"
);
let target_region_id = RegionId::new(1, 2);
engine
.handle_request(target_region_id, RegionRequest::Create(request))
.await
.unwrap();
engine
.copy_region_from(
target_region_id,
MitoCopyRegionFromRequest {
source_region_id,
parallelism: 1,
},
)
.await
.unwrap();
// The copied file in the target must NOT be trusted: its physical per-row
// sequences belong to the source region's independent sequence domain.
// Both the `preserve_row_sequence` marker and the source-domain max
// `sequence` hint must be cleared, so exact reads fail closed instead of
// silently skipping the file as "proven disjoint".
let target_manifest = engine
.get_region(target_region_id)
.unwrap()
.manifest_ctx
.manifest()
.await;
assert_eq!(1, target_manifest.files.len());
assert!(
target_manifest
.files
.values()
.all(|meta| !meta.preserve_row_sequence),
"copied files must have the preserve_row_sequence marker cleared"
);
assert!(
target_manifest
.files
.values()
.all(|meta| meta.sequence.is_none()),
"copied files must have their source-domain sequence hint cleared"
);
// An exact sequence-range request intersecting the copied rows' sequences
// (rows 3..=7 in the source domain) must fail closed with
// `SequenceRangeUnsupported` instead of replaying source-domain rows.
let err = engine
.scanner(
target_region_id,
ScanRequest {
memtable_min_sequence: Some(2),
memtable_max_sequence: Some(7),
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);
// Sanity: the source region keeps the marker and still serves the same
// exact range with its own row-level sequences (3..=7, 5 rows).
let stream = engine
.scan_to_stream(
source_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();
let row_count: usize = batches.iter().map(|b| b.num_rows()).sum();
assert_eq!(
5, row_count,
"source exact range should return rows with sequence 3..=7"
);
}
src/mito2/src/engine/edit_region_test.rs
+36
View File
@@ -563,6 +563,42 @@ async fn test_region_edit_with_file_sequence_is_not_merged() {
assert_eq!(Some(3), region_file_sequence(&region, 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) {
src/mito2/src/engine/scan_test.rs
+1693 -12
View File
File diff suppressed because it is too large Load Diff
src/mito2/src/error.rs
+34 -1
View File
@@ -286,6 +286,36 @@ 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(
"region {} is unusable for sequence reads: file {} declares region {}",
region_id,
file_id,
file_region_id
))]
RegionSequenceDomainBroken {
region_id: RegionId,
file_region_id: RegionId,
file_id: FileId,
#[snafu(implicit)]
location: Location,
},
#[snafu(display("Old manifest missing for region {}", region_id))]
MissingOldManifest {
region_id: RegionId,
@@ -1466,7 +1496,10 @@ impl ErrorExt for Error {
IncrementalQueryStale { .. } | SnapshotFenceStale { .. } => StatusCode::RequestOutdated,
RegionMetadataNotFound { .. }
SequenceRangeUnsupported { .. } => StatusCode::Unsupported,
RegionSequenceDomainBroken { .. }
| RegionMetadataNotFound { .. }
| Join { .. }
| WorkerStopped { .. }
| Recv { .. }
src/mito2/src/flush.rs
+4
View File
@@ -661,6 +661,7 @@ impl RegionFlushTask {
sst_info,
partition_expr.clone(),
pk_range,
version.options.preserve_row_sequence,
));
}
if hook.is_some() {
@@ -784,6 +785,7 @@ impl RegionFlushTask {
sst_info: &SstInfo,
partition_expr: Option<PartitionExpr>,
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)),
@@ -807,6 +809,7 @@ impl RegionFlushTask {
num_series: sst_info.num_series,
primary_key_min,
primary_key_max,
preserve_row_sequence,
}
}
@@ -833,6 +836,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(),
src/mito2/src/read/range_cache.rs
+13 -1
View File
@@ -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<SequenceRange>,
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<SequenceRange>,
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,
}
@@ -855,6 +865,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();
@@ -921,6 +932,7 @@ mod tests {
append_mode: false,
filter_deleted,
merge_mode: MergeMode::LastRow,
sequence_range: None,
partition_expr_version,
}
.build()
src/mito2/src/read/scan_region.rs
+310 -27
View File
@@ -56,7 +56,10 @@ 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, RegionSequenceDomainBrokenSnafu, Result,
SequenceRangeUnsupportedSnafu,
};
#[cfg(feature = "enterprise")]
use crate::extension::{BoxedExtensionRange, BoxedExtensionRangeProvider};
use crate::memtable::{MemtableRange, RangesOptions};
@@ -255,6 +258,8 @@ pub(crate) struct ScanRegion {
/// Whether to filter out the deleted rows.
/// Usually true for normal read, and false for scan for compaction.
filter_deleted: bool,
/// Files and exact sequence capability selected together by the engine.
exact_selection: Option<(Vec<FileHandle>, Option<SequenceRange>)>,
/// Counters that should receive query-load metrics.
query_stat_counters: Option<RegionQueryStatCounters>,
#[cfg(feature = "enterprise")]
@@ -282,6 +287,7 @@ impl ScanRegion {
ignore_bloom_filter: false,
start_time: None,
filter_deleted: true,
exact_selection: None,
query_stat_counters: None,
#[cfg(feature = "enterprise")]
extension_range_provider: None,
@@ -350,6 +356,14 @@ impl ScanRegion {
self.filter_deleted = filter_deleted;
}
pub(crate) fn with_exact_selection(
mut self,
exact_selection: (Vec<FileHandle>, Option<SequenceRange>),
) -> Self {
self.exact_selection = Some(exact_selection);
self
}
#[cfg(feature = "enterprise")]
pub(crate) fn set_extension_range_provider(
&mut self,
@@ -435,7 +449,7 @@ impl ScanRegion {
/// Creates a scan input.
#[tracing::instrument(skip_all, fields(region_id = %self.region_id()))]
async fn scan_input(self) -> Result<ScanInput> {
async fn scan_input(mut self) -> Result<ScanInput> {
let metadata = &self.version.metadata;
let sst_min_sequence = self.request.sst_min_sequence.and_then(NonZeroU64::new);
let time_range = self.build_time_range_predicate();
@@ -458,33 +472,23 @@ impl ScanRegion {
mapper
};
let ssts = &self.version.ssts;
let mut files = Vec::new();
if !self.request.skip_sst_files {
for level in ssts.levels() {
for file in level.files.values() {
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
// is generated and added to the region "directly". In this case, its data should
// be considered as fresh as the memtable. So its sequence is treated greater than
// the min_sequence, whatever the value of min_sequence is. Hence the default
// "true" in this arm.
(Some(_), None) => true,
(None, _) => true,
};
// Finds SST files in range.
if exceed_min_sequence && file_in_range(file, &time_range) {
files.push(file.clone());
}
// There is no need to check and prune for file's sequence here as the sequence number is usually very new,
// unless the timing is too good, or the sequence number wouldn't be in file.
// and the batch will be filtered out by tree reader anyway.
}
let (files, sequence_range) =
if let Some((files, sequence_range)) = self.exact_selection.take() {
(files, sequence_range)
} else {
exact_sequence_range(&self.request, &self.version)?
};
if sst_min_sequence.is_some() && 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 memtables = self.version.memtables.list_memtables();
// Skip empty memtables and memtables out of time range.
let mut mem_range_builders = Vec::new();
@@ -580,6 +584,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
@@ -590,6 +595,22 @@ impl ScanRegion {
let input = if !self.request.skip_sst_files
&& let Some(provider) = self.extension_range_provider
{
if sequence_range.is_some() {
// Defense in depth: the engine already rejects exact
// sequence-range reads on follower regions with an extension
// provider, but if one ever reaches the reader, fail closed
// here rather than letting unfiltered extension streams bypass
// the row-level sequence filter and emit out-of-range rows.
return SequenceRangeUnsupportedSnafu {
region_id,
min_seq: self.request.memtable_min_sequence.unwrap_or_default(),
max_seq: self.request.memtable_max_sequence.unwrap_or_default(),
reason:
"exact sequence-range reads are unsupported when an extension range provider is present"
.to_string(),
}
.fail();
}
let ranges = provider
.find_extension_ranges(self.version.flushed_sequence, time_range, &self.request)
.await?;
@@ -951,6 +972,11 @@ 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<SequenceNumber>,
/// 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<SequenceRange>,
/// Whether this scan is for compaction.
pub(crate) compaction: bool,
/// Compaction-only JSON2 physical rewrite targets.
@@ -994,6 +1020,7 @@ impl ScanInput {
distribution: None,
explain_flat_format: false,
snapshot_sequence: None,
sequence_range: None,
compaction: false,
json2_rewrite_targets: Arc::default(),
query_stat_counters: None,
@@ -1194,6 +1221,12 @@ impl ScanInput {
self
}
#[must_use]
pub(crate) fn with_sequence_range(mut self, sequence_range: Option<SequenceRange>) -> 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 {
@@ -1620,6 +1653,105 @@ fn pre_filter_mode(append_mode: bool, merge_mode: MergeMode) -> PreFilterMode {
}
}
/// Selects the SST files for a scan and, when requested, determines whether
/// the selected files support an exact sequence-range scan.
///
/// Files excluded by the request time range are not selected: `(C, H]` rows are
/// a subset of the query's time-range rows, so a time-pruned file cannot
/// contribute a row to `(C, H]`.
///
/// Unmarked local files use `FileMeta.sequence` as an admission barrier rather
/// than a row maximum. Compaction and edit assign it as `committed_sequence + 1`;
/// `C >= barrier` proves Flow has already consumed the entire file, so such a
/// file is excluded before the capability check.
///
/// A foreign file is different: the parquet reader virtualizes every row to its
/// target-local `FileMeta.sequence`. Consequently, a present sequence is the
/// only trust requirement for a foreign file; its source marker is irrelevant.
/// A foreign file without that barrier is retained as a failed-closed error so
/// it cannot be mistaken for a local sequence domain. This check is performed
/// even when another exact-range capability condition would return `None`.
pub(crate) fn exact_sequence_range(
request: &ScanRequest,
version: &crate::region::version::Version,
) -> Result<(Vec<FileHandle>, Option<SequenceRange>)> {
if request.skip_sst_files {
return Ok((Vec::new(), None));
}
let time_index = version.metadata.time_index_column();
let unit = time_index
.column_schema
.data_type
.as_timestamp()
.expect("Time index must have timestamp-compatible type")
.unit();
let time_range =
build_time_range_predicate(&time_index.column_schema.name, unit, &request.filters);
let min = request.memtable_min_sequence;
let mut check_capability = request.exact_sequence_range && min.is_some();
let sst_min_sequence = request.sst_min_sequence.and_then(NonZeroU64::new);
let mut files = Vec::new();
let mut files_allow_exact_range = true;
for file in version
.ssts
.levels()
.iter()
.flat_map(|level| level.files.values())
.filter(|file| file_in_range(file, &time_range))
{
let meta = file.meta_ref();
let selected = (!request.exact_sequence_range
|| min.is_none_or(|min| meta.sequence.is_none_or(|sequence| sequence.get() > min)))
&& match (sst_min_sequence, meta.sequence) {
(Some(min_sequence), Some(file_sequence)) => file_sequence > min_sequence,
// A missing file sequence is treated as newer than the SST
// pruning hint, matching scan input construction.
(Some(_), None) | (None, _) => true,
};
if !selected {
continue;
}
if let Some(min) = min
&& check_capability
{
if meta.region_id != version.metadata.region_id && meta.sequence.is_none() {
return RegionSequenceDomainBrokenSnafu {
region_id: version.metadata.region_id,
file_region_id: meta.region_id,
file_id: meta.file_id,
}
.fail();
}
if meta.region_id == version.metadata.region_id
&& !file.is_effective_target_sequence_trusted(version.metadata.region_id)
&& meta.sequence.is_none_or(|barrier| barrier.get() > min)
{
// Match the capability helper's short-circuit behavior: once
// an unadmitted local file is found, later files cannot change
// the result or expose a foreign-domain error.
files_allow_exact_range = false;
check_capability = false;
}
}
files.push(file.clone());
}
let sequence_range = match (
request.exact_sequence_range,
min,
version.options.preserve_row_sequence,
request.memtable_max_sequence,
files_allow_exact_range,
) {
(true, Some(min), true, Some(max), true) => Some(SequenceRange::GtLtEq { min, max }),
_ => None,
};
Ok((files, sequence_range))
}
/// Output of [build_scan_fingerprint]: the cache fingerprint plus the derived
/// implied time range used to decide whether the cache key can drop the time
/// predicates for a given partition (see `build_range_cache_key`).
@@ -1734,6 +1866,7 @@ pub(crate) fn build_scan_fingerprint(input: &ScanInput) -> Option<ScanFingerprin
append_mode: input.append_mode,
filter_deleted: input.filter_deleted,
merge_mode: input.merge_mode,
sequence_range: input.sequence_range,
partition_expr_version: metadata.partition_expr_version,
}
.build();
@@ -2286,6 +2419,7 @@ mod tests {
append_mode: false,
filter_deleted: false,
merge_mode: MergeMode::LastNonNull,
sequence_range: None,
partition_expr_version: 0,
}
.build();
@@ -2359,6 +2493,7 @@ mod tests {
append_mode: false,
filter_deleted: true,
merge_mode: MergeMode::LastRow,
sequence_range: None,
partition_expr_version: metadata.partition_expr_version,
}
.build();
@@ -2727,4 +2862,152 @@ mod tests {
assert_eq!(expected_mode, input.range_pre_filter_mode(source_count));
}
}
#[test]
fn test_exact_sequence_selection_preserves_range_and_fallback() {
let request = ScanRequest {
exact_sequence_range: true,
memtable_min_sequence: Some(1),
memtable_max_sequence: Some(2),
..Default::default()
};
let metadata = Arc::new(metadata_with_primary_key(vec![0, 1], false));
let mutable = Arc::new(crate::memtable::time_partition::TimePartitions::new(
metadata.clone(),
Arc::new(crate::test_util::memtable_util::EmptyMemtableBuilder::default()),
0,
None,
));
let version = Arc::new(
crate::region::version::VersionBuilder::new(metadata, mutable)
.options(crate::region::options::RegionOptions {
preserve_row_sequence: true,
..Default::default()
})
.build(),
);
let (files, range) = exact_sequence_range(&request, &version).unwrap();
assert!(files.is_empty());
assert_eq!(Some(SequenceRange::GtLtEq { min: 1, max: 2 }), range);
let skip_sst_request = ScanRequest {
skip_sst_files: true,
..request
};
let (files, range) = exact_sequence_range(&skip_sst_request, &version).unwrap();
assert!(files.is_empty());
assert_eq!(None, range);
}
#[test]
fn test_exact_sequence_selection_checks_selected_files() {
use std::num::NonZeroU64;
use crate::test_util::new_noop_file_purger;
let request = ScanRequest {
exact_sequence_range: true,
memtable_min_sequence: Some(5),
memtable_max_sequence: Some(10),
..Default::default()
};
let metadata = Arc::new(metadata_with_primary_key(vec![0, 1], false));
let mutable = Arc::new(crate::memtable::time_partition::TimePartitions::new(
metadata.clone(),
Arc::new(crate::test_util::memtable_util::EmptyMemtableBuilder::default()),
0,
None,
));
let target_region_id = metadata.region_id;
// Files at or below the cursor are excluded. Among the remaining files,
// marked locals are exact-capable while unmarked locals deny exactness.
for (sequence, marked, expected_files, expected_range) in [
(NonZeroU64::new(5), false, false, true),
(NonZeroU64::new(5), true, false, true),
(NonZeroU64::new(100), false, true, false),
(None, false, true, false),
(None, true, true, true),
] {
let version =
crate::region::version::VersionBuilder::new(metadata.clone(), mutable.clone())
.options(crate::region::options::RegionOptions {
preserve_row_sequence: true,
..Default::default()
})
.add_files(
new_noop_file_purger(),
[FileMeta {
region_id: target_region_id,
sequence,
preserve_row_sequence: marked,
..Default::default()
}]
.into_iter(),
)
.build();
let (files, range) = exact_sequence_range(&request, &version).unwrap();
assert_eq!(expected_files, !files.is_empty());
assert_eq!(expected_range, range.is_some());
}
// Foreign files use the target-local barrier, regardless of their
// source marker. A missing barrier fails closed.
for (sequence, marked, expected_count, expected_error) in [
(NonZeroU64::new(5), false, 0, false),
(NonZeroU64::new(6), false, 1, false),
(NonZeroU64::new(11), true, 1, false),
(None, true, 1, true),
] {
let file_id = store_api::storage::FileId::random();
let version =
crate::region::version::VersionBuilder::new(metadata.clone(), mutable.clone())
.options(crate::region::options::RegionOptions {
preserve_row_sequence: true,
..Default::default()
})
.add_files(
new_noop_file_purger(),
[FileMeta {
region_id: RegionId::new(1, 2),
file_id,
sequence,
preserve_row_sequence: marked,
..Default::default()
}]
.into_iter(),
)
.build();
let result = exact_sequence_range(&request, &version);
if expected_error {
assert!(result.is_err());
} else {
let (files, range) = result.unwrap();
assert_eq!(expected_count, files.len());
assert!(range.is_some());
}
}
// Foreign-domain validation still takes precedence over preserve mode
// and a missing upper bound once the file is selected.
let request = ScanRequest {
memtable_max_sequence: None,
..request
};
let version = crate::region::version::VersionBuilder::new(metadata, mutable)
.options(crate::region::options::RegionOptions {
preserve_row_sequence: false,
..Default::default()
})
.add_files(
new_noop_file_purger(),
[FileMeta {
region_id: RegionId::new(1, 2),
..Default::default()
}]
.into_iter(),
)
.build();
assert!(exact_sequence_range(&request, &version).is_err());
}
}
src/mito2/src/read/scan_util.rs
+185 -4
View File
@@ -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_sequence_trusted` is the per-file trust decision. Foreign reader
/// batches have already been virtualized to the target-local file barrier, so
/// they are filtered using the effective batch sequence. Local untrusted files
/// pass through because exact capability excludes them before row filtering.
fn filter_flat_batch_by_sequence(
record_batch: RecordBatch,
sequence_range: Option<SequenceRange>,
file_sequence_trusted: bool,
) -> Result<Option<RecordBatch>> {
let Some(sequence) = sequence_range else {
return Ok(Some(record_batch));
};
if !file_sequence_trusted {
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?
@@ -1538,6 +1594,9 @@ pub fn build_flat_file_range_scan_stream(
part_metrics.inc_build_reader_cost(build_cost);
let may_compat = range.compat_batch();
let file_sequence_trusted = range
.file_handle()
.is_effective_target_sequence_trusted(stream_ctx.input.region_metadata().region_id);
let mapper = range.compaction_projection_mapper();
while let Some(record_batch) = reader.next_batch().await? {
@@ -1548,6 +1607,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,
file_sequence_trusted,
)? else {
continue;
};
if let Some(flat_compat) = may_compat {
let batch = flat_compat.compat(record_batch)?;
yield batch;
@@ -2033,3 +2100,117 @@ 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<String> {
batch
.column(0)
.as_any()
.downcast_ref::<StringArray>()
.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);
// Foreign batches are already virtualized by the reader. Their source
// marker is irrelevant, and filtering uses the effective batch values.
let out = filter_flat_batch_by_sequence(
batch(&[1, 5, 9]),
Some(SequenceRange::GtLtEq { min: 2, max: 8 }),
true,
)
.unwrap();
assert_eq!(remaining_tags(&out.unwrap()), vec!["1"]);
}
}
src/mito2/src/region/options.rs
+52
View File
@@ -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<ReadableSize>,
/// 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<FormatType>,
#[serde_as(as = "NoneAsEmptyString")]
max_row_group_row_count: Option<usize>,
#[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,32 @@ 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() {
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);
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,14 +1049,18 @@ mod tests {
max_row_group_row_count: None,
primary_key_encoding: None,
write_buffer_size: Some(ReadableSize::mb(128)),
preserve_row_sequence: true,
};
let region_options_json_str = serde_json::to_string(&options).unwrap();
assert!(region_options_json_str.contains("preserve_row_sequence"));
let got: RegionOptions = serde_json::from_str(&region_options_json_str).unwrap();
assert_eq!(options, got);
// Old manifests without the key default to false.
let old_region_options_json_str = r#"{"ttl":null}"#;
let got: RegionOptions = serde_json::from_str(old_region_options_json_str).unwrap();
assert_eq!(None, got.write_buffer_size);
assert!(!got.preserve_row_sequence);
let default_json = serde_json::to_value(RegionOptions::default()).unwrap();
assert!(default_json.get(WRITE_BUFFER_SIZE_KEY).is_none());
@@ -1069,6 +1120,7 @@ mod tests {
max_row_group_row_count: None,
primary_key_encoding: None,
write_buffer_size: None,
preserve_row_sequence: false,
};
assert_eq!(options, got);
}
src/mito2/src/sst/file.rs
+44
View File
@@ -282,6 +282,14 @@ pub struct FileMeta {
deserialize_with = "deserialize_bytes_option"
)]
pub primary_key_max: Option<Bytes>,
/// Whether the file preserves per-row sequence numbers usable for exact
/// row-level sequence filtering.
#[serde(default, skip_serializing_if = "is_false")]
pub preserve_row_sequence: bool,
}
fn is_false(value: &bool) -> bool {
!*value
}
impl Debug for FileMeta {
@@ -507,6 +515,18 @@ impl FileHandle {
self.inner.meta.region_id
}
/// Returns whether this file's row sequences are trusted in the target region.
///
/// Foreign files use their target-local sequence barrier; local files require
/// the preserve marker because their physical sequences belong to this region.
pub(crate) fn is_effective_target_sequence_trusted(&self, target_region_id: RegionId) -> bool {
if self.region_id() != target_region_id {
self.meta_ref().sequence.is_some()
} else {
self.meta_ref().preserve_row_sequence
}
}
/// Returns the cross-region file id.
pub fn file_id(&self) -> RegionFileId {
RegionFileId::new(self.inner.meta.region_id, self.inner.meta.file_id)
@@ -1116,6 +1136,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() {
src/mito2/src/sst/index.rs
+1
View File
@@ -1700,6 +1700,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(),
src/mito2/src/sst/parquet.rs
+91 -42
View File
@@ -1703,8 +1703,20 @@ mod tests {
}
}
let custom_sequence = 12345u64;
fn handle_with_meta(
handle: &FileHandle,
sequence: Option<u64>,
preserve_row_sequence: bool,
) -> FileHandle {
let mut file_meta = handle.meta_ref().clone();
file_meta.sequence = sequence.and_then(std::num::NonZeroU64::new);
file_meta.preserve_row_sequence = preserve_row_sequence;
FileHandle::new(file_meta, Arc::new(NoopFilePurger))
}
let custom_sequence = 12345;
let local_zero_handle = sst_file_handle(0, 1000);
let local_nonzero_handle = sst_file_handle(0, 1000);
write_sst(
object_store.clone(),
metadata.clone(),
@@ -1713,8 +1725,15 @@ mod tests {
0,
)
.await;
write_sst(
object_store.clone(),
metadata.clone(),
local_nonzero_handle.clone(),
flat_format,
7,
)
.await;
// Local all-zero SSTs retain the compatibility override.
let local_zero_none = read_sequences(
ParquetReaderBuilder::new(
FILE_DIR.to_string(),
@@ -1727,17 +1746,12 @@ mod tests {
.await;
assert!(local_zero_none.iter().all(|sequence| *sequence == 0));
let mut local_zero_meta = local_zero_handle.meta_ref().clone();
local_zero_meta.sequence = Some(std::num::NonZeroU64::new(custom_sequence).unwrap());
let local_zero_override_handle = FileHandle::new(
local_zero_meta,
Arc::new(crate::sst::file_purger::NoopFilePurger),
);
// Legacy local all-zero files use the FileMeta sequence compatibility override.
let local_zero_override = read_sequences(
ParquetReaderBuilder::new(
FILE_DIR.to_string(),
PathType::Bare,
local_zero_override_handle,
handle_with_meta(&local_zero_handle, Some(custom_sequence), false),
object_store.clone(),
)
.expected_metadata(Some(metadata.clone())),
@@ -1749,28 +1763,12 @@ mod tests {
.all(|sequence| *sequence == custom_sequence)
);
let local_nonzero_handle = sst_file_handle(0, 1000);
write_sst(
object_store.clone(),
metadata.clone(),
local_nonzero_handle.clone(),
flat_format,
7,
)
.await;
// Local nonzero SSTs retain physical per-row sequences, even with FileMeta.sequence.
let mut local_nonzero_meta = local_nonzero_handle.meta_ref().clone();
local_nonzero_meta.sequence = Some(std::num::NonZeroU64::new(custom_sequence).unwrap());
let local_nonzero_override_handle = FileHandle::new(
local_nonzero_meta,
Arc::new(crate::sst::file_purger::NoopFilePurger),
);
// Local nonzero physical sequences must not be replaced by a legacy barrier.
let local_nonzero_override = read_sequences(
ParquetReaderBuilder::new(
FILE_DIR.to_string(),
PathType::Bare,
local_nonzero_override_handle,
handle_with_meta(&local_nonzero_handle, Some(custom_sequence), false),
object_store.clone(),
)
.expected_metadata(Some(metadata.clone())),
@@ -1778,7 +1776,6 @@ mod tests {
.await;
assert!(local_nonzero_override.iter().all(|sequence| *sequence == 7));
// None never overrides a local nonzero physical sequence.
let local_nonzero_none = read_sequences(
ParquetReaderBuilder::new(
FILE_DIR.to_string(),
@@ -1791,28 +1788,80 @@ mod tests {
.await;
assert!(local_nonzero_none.iter().all(|sequence| *sequence == 7));
// A source-owned handle is foreign when read against target metadata, so the
// target-local manifest barrier is applied even for nonzero physical sequences.
let mut target_metadata = (*metadata).clone();
target_metadata.region_id = RegionId::new(0, 1);
let target_metadata = Arc::new(target_metadata);
let mut foreign_meta = local_nonzero_handle.meta_ref().clone();
foreign_meta.sequence = Some(std::num::NonZeroU64::new(custom_sequence).unwrap());
let foreign_handle = FileHandle::new(
foreign_meta,
Arc::new(crate::sst::file_purger::NoopFilePurger),
);
let foreign = read_sequences(
// The trusted marker preserves physical sequences, including explicit all-zero data.
let local_trusted_nonzero = read_sequences(
ParquetReaderBuilder::new(
FILE_DIR.to_string(),
PathType::Bare,
foreign_handle,
handle_with_meta(&local_nonzero_handle, Some(custom_sequence), true),
object_store.clone(),
)
.expected_metadata(Some(metadata.clone())),
)
.await;
assert!(local_trusted_nonzero.iter().all(|sequence| *sequence == 7));
let local_trusted_zero = read_sequences(
ParquetReaderBuilder::new(
FILE_DIR.to_string(),
PathType::Bare,
handle_with_meta(&local_zero_handle, Some(custom_sequence), true),
object_store.clone(),
)
.expected_metadata(Some(metadata.clone())),
)
.await;
assert!(local_trusted_zero.iter().all(|sequence| *sequence == 0));
let mut target_metadata = (*metadata).clone();
target_metadata.region_id = RegionId::new(0, 1);
let target_metadata = Arc::new(target_metadata);
// A foreign file always uses the target-local barrier, regardless of its marker.
let foreign_marked = read_sequences(
ParquetReaderBuilder::new(
FILE_DIR.to_string(),
PathType::Bare,
handle_with_meta(&local_nonzero_handle, Some(custom_sequence), true),
object_store.clone(),
)
.expected_metadata(Some(target_metadata.clone())),
)
.await;
assert!(
foreign_marked
.iter()
.all(|sequence| *sequence == custom_sequence)
);
let foreign_unmarked = read_sequences(
ParquetReaderBuilder::new(
FILE_DIR.to_string(),
PathType::Bare,
handle_with_meta(&local_nonzero_handle, Some(custom_sequence), false),
object_store.clone(),
)
.expected_metadata(Some(target_metadata.clone())),
)
.await;
assert!(
foreign_unmarked
.iter()
.all(|sequence| *sequence == custom_sequence)
);
// A foreign handle without a barrier leaves physical sequences untouched.
let foreign_none = read_sequences(
ParquetReaderBuilder::new(
FILE_DIR.to_string(),
PathType::Bare,
handle_with_meta(&local_nonzero_handle, None, false),
object_store,
)
.expected_metadata(Some(target_metadata)),
)
.await;
assert!(foreign.iter().all(|sequence| *sequence == custom_sequence));
assert!(foreign_none.iter().all(|sequence| *sequence == 7));
}
#[tokio::test]
src/mito2/src/sst/parquet/reader.rs
+12 -3
View File
@@ -539,9 +539,18 @@ impl ParquetReaderBuilder {
expected_region_id,
);
}
if is_foreign || need_override_sequence(&parquet_meta) {
read_format
.set_override_sequence(self.file_handle.meta_ref().sequence.map(|x| x.get()));
let file_meta = self.file_handle.meta_ref();
let override_sequence = if is_foreign {
file_meta.sequence.map(|sequence| sequence.get())
} else if file_meta.preserve_row_sequence {
None
} else if need_override_sequence(&parquet_meta) {
file_meta.sequence.map(|sequence| sequence.get())
} else {
None
};
if let Some(sequence) = override_sequence {
read_format.set_override_sequence(Some(sequence));
}
// Computes the projection mask.
src/mito2/src/worker/handle_alter.rs
+37 -7
View File
@@ -197,7 +197,11 @@ impl<S: LogStore> RegionWorkerLoop<S> {
/// 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(
@@ -301,6 +305,15 @@ impl<S: LogStore> RegionWorkerLoop<S> {
all_options_altered = false;
}
}
SetRegionOption::PreserveRowSequence(new_preserve) => {
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);
@@ -309,15 +322,29 @@ impl<S: LogStore> RegionWorkerLoop<S> {
}
}
}
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(&current_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);
region.version_control.alter_options(candidate);
Ok(None)
} else {
let kind = AlterKind::SetRegionOptions { options };
Ok(new_region_options_on_empty_memtable(
&current_options,
&kind,
))
// Some options require an empty memtable (e.g. append_mode,
// sst_format, or max_row_group_row_count).
Ok(Some(candidate))
}
}
}
@@ -369,6 +396,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)
src/mito2/src/worker/handle_copy_region.rs
+50 -39
View File
@@ -25,6 +25,7 @@ use crate::region::{
use crate::request::{
BackgroundNotify, CopyRegionFromFinished, CopyRegionFromRequest, WorkerRequest,
};
use crate::sst::file::FileMeta;
use crate::sst::location::region_dir_from_table_dir;
use crate::worker::{RegionWorkerLoop, WorkerRequestWithTime};
@@ -68,7 +69,7 @@ impl<S> RegionWorkerLoop<S> {
let worker_sender = self.sender.clone();
common_runtime::spawn_global(async move {
let (region_edit, file_ids) = match Self::copy_region_from(
let (region_edit, source_file_ids) = match Self::copy_region_from(
&region,
region_metadata_loader,
source_region_id,
@@ -104,7 +105,7 @@ impl<S> RegionWorkerLoop<S> {
}
None => {
let _ = sender.send(Ok(MitoCopyRegionFromResponse {
copied_file_ids: file_ids,
copied_file_ids: source_file_ids,
}));
}
}
@@ -138,7 +139,7 @@ impl<S> RegionWorkerLoop<S> {
/// Returns the region edit and the file ids that were copied from the source region to the target region.
///
/// If no need to copy files, returns (None, file_ids).
/// If no need to copy files, returns (None, source_file_ids).
async fn copy_region_from(
region: &MitoRegionRef,
region_metadata_loader: RegionMetadataLoader,
@@ -158,9 +159,9 @@ impl<S> RegionWorkerLoop<S> {
.context(MissingManifestSnafu {
region_id: source_region_id,
})?;
let mut files_to_copy = vec![];
let mut new_file_metas = vec![];
let target_region_manifest = region.manifest_ctx.manifest().await;
let file_ids = source_region_manifest
let source_file_ids = source_region_manifest
.files
.keys()
.cloned()
@@ -171,43 +172,16 @@ impl<S> RegionWorkerLoop<S> {
);
for (file_id, file_meta) in &source_region_manifest.files {
if !target_region_manifest.files.contains_key(file_id) {
let mut new_file_meta = file_meta.clone();
new_file_meta.region_id = target_region_id;
files_to_copy.push(new_file_meta);
new_file_metas.push(remap_copied_file_meta(file_meta, target_region_id));
}
}
if files_to_copy.is_empty() {
return Ok((None, file_ids));
if new_file_metas.is_empty() {
return Ok((None, source_file_ids));
}
let file_descriptors = files_to_copy
let file_descriptors = new_file_metas
.iter()
.flat_map(|file_meta| {
if file_meta.exists_index() {
let region_index_id = file_meta.index_id();
let file_id = region_index_id.file_id.file_id();
let version = region_index_id.version;
let file_size = file_meta.file_size;
let index_file_size = file_meta.index_file_size();
vec![
FileDescriptor::Data {
file_id: file_meta.file_id,
size: file_size,
},
FileDescriptor::Index {
file_id,
version,
size: index_file_size,
},
]
} else {
let file_size = file_meta.file_size;
vec![FileDescriptor::Data {
file_id: file_meta.file_id,
size: file_size,
}]
}
})
.flat_map(file_descriptors_for_meta)
.collect();
debug!("File descriptors to copy: {:?}", file_descriptors);
let copier = RegionFileCopier::new(region.access_layer());
@@ -221,7 +195,7 @@ impl<S> RegionWorkerLoop<S> {
)
.await?;
let edit = RegionEdit {
files_to_add: files_to_copy,
files_to_add: new_file_metas,
files_to_remove: vec![],
timestamp_ms: Some(chrono::Utc::now().timestamp_millis()),
compaction_time_window: None,
@@ -242,6 +216,43 @@ impl<S> RegionWorkerLoop<S> {
"Successfully update manifest version to {version}, region: {target_region_id}, reason: CopyRegionFrom"
);
Ok((Some(edit), file_ids))
Ok((Some(edit), source_file_ids))
}
}
fn remap_copied_file_meta(file_meta: &FileMeta, target_region_id: RegionId) -> FileMeta {
let mut new_file_meta = file_meta.clone();
new_file_meta.region_id = target_region_id;
// The target region has an independent sequence domain: the physical
// per-row sequences in the copied file belong to the source region, so they
// must not be trusted for exact sequence-range reads on the target. Clear
// the `preserve_row_sequence` marker and source-domain max sequence to fail
// closed until the scan provably cannot intersect the copied rows (see
// `files_allow_exact_sequence_range`); otherwise an exact request would
// replay source-domain rows as target sequences, and an unmarked file with
// a stale source-domain `sequence` hint could be silently skipped as
// "proven disjoint".
new_file_meta.preserve_row_sequence = false;
new_file_meta.sequence = None;
new_file_meta
}
fn file_descriptors_for_meta(file_meta: &FileMeta) -> Vec<FileDescriptor> {
let data = FileDescriptor::Data {
file_id: file_meta.file_id,
size: file_meta.file_size,
};
if file_meta.exists_index() {
let region_index_id = file_meta.index_id();
vec![
data,
FileDescriptor::Index {
file_id: region_index_id.file_id.file_id(),
version: region_index_id.version,
size: file_meta.index_file_size(),
},
]
} else {
vec![data]
}
}
src/mito2/src/worker/handle_manifest.rs
+6 -1
View File
@@ -316,9 +316,14 @@ impl<S: LogStore> RegionWorkerLoop<S> {
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
// 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.
src/store-api/src/mito_engine_options.rs
+4
View File
@@ -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"));
}
}
src/store-api/src/region_request.rs
+43 -1
View File
@@ -57,7 +57,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;
@@ -1494,6 +1494,7 @@ pub enum SetRegionOption {
AutoFlushInterval(Option<Duration>),
// Modifying the max number of rows in a parquet row group.
MaxRowGroupRowCount(Option<usize>),
PreserveRowSequence(bool),
// Stops writing new WAL entries. This operation is irreversible.
SkipWal,
}
@@ -1553,6 +1554,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::<bool>()
.map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
Ok(Self::PreserveRowSequence(preserve))
}
SKIP_WAL_KEY if value == "true" => Ok(Self::SkipWal),
_ => InvalidSetRegionOptionRequestSnafu { key, value }.fail(),
}
@@ -1574,6 +1581,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),
}
}
}
@@ -1589,6 +1597,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(),
}
}
@@ -1602,6 +1611,7 @@ pub enum UnsetRegionOption {
Ttl,
MaxRowGroupRowCount,
WriteBufferSize,
PreserveRowSequence,
}
impl UnsetRegionOption {
@@ -1613,6 +1623,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,
}
}
}
@@ -2021,6 +2032,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 {
src/store-api/src/storage/requests.rs
+14 -1
View File
@@ -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<TimeSeriesDistribution>,
/// 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 {
src/table/src/metadata.rs
+70 -1
View File
@@ -31,7 +31,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};
@@ -406,6 +406,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
@@ -1826,6 +1835,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()
tests/cases/standalone/common/alter/alter_preserve_row_sequence.result
+122
View File
@@ -0,0 +1,122 @@
-- Test altering preserve_row_sequence on an append-only table, including
-- persistence across a restart.
CREATE TABLE test_alter_preserve_row_sequence(
host STRING,
ts TIMESTAMP TIME INDEX,
cpu DOUBLE,
PRIMARY KEY(host)
) ENGINE=mito WITH('append_mode'='true');
Affected Rows: 0
INSERT INTO test_alter_preserve_row_sequence VALUES ('host1', 0, 1.0), ('host2', 1, 2.0);
Affected Rows: 2
-- SET preserve_row_sequence on the append-only table should succeed
ALTER TABLE test_alter_preserve_row_sequence SET 'preserve_row_sequence' = 'true';
Affected Rows: 0
-- SHOW CREATE TABLE should show both append_mode and preserve_row_sequence
SHOW CREATE TABLE test_alter_preserve_row_sequence;
+----------------------------------+-----------------------------------------------------------------+
| Table | Create Table |
+----------------------------------+-----------------------------------------------------------------+
| test_alter_preserve_row_sequence | CREATE TABLE IF NOT EXISTS "test_alter_preserve_row_sequence" ( |
| | "host" STRING NULL, |
| | "ts" TIMESTAMP(3) NOT NULL, |
| | "cpu" DOUBLE NULL, |
| | TIME INDEX ("ts"), |
| | PRIMARY KEY ("host") |
| | ) |
| | |
| | ENGINE=mito |
| | WITH( |
| | append_mode = 'true', |
| | preserve_row_sequence = 'true' |
| | ) |
+----------------------------------+-----------------------------------------------------------------+
-- Restart the server: the option and the inserted data must survive
-- SQLNESS ARG restart=true
SHOW CREATE TABLE test_alter_preserve_row_sequence;
+----------------------------------+-----------------------------------------------------------------+
| Table | Create Table |
+----------------------------------+-----------------------------------------------------------------+
| test_alter_preserve_row_sequence | CREATE TABLE IF NOT EXISTS "test_alter_preserve_row_sequence" ( |
| | "host" STRING NULL, |
| | "ts" TIMESTAMP(3) NOT NULL, |
| | "cpu" DOUBLE NULL, |
| | TIME INDEX ("ts"), |
| | PRIMARY KEY ("host") |
| | ) |
| | |
| | ENGINE=mito |
| | WITH( |
| | append_mode = 'true', |
| | preserve_row_sequence = 'true' |
| | ) |
+----------------------------------+-----------------------------------------------------------------+
-- Ordinary query should still see the inserted rows after restart
SELECT * FROM test_alter_preserve_row_sequence ORDER BY host, ts;
+-------+-------------------------+-----+
| host | ts | cpu |
+-------+-------------------------+-----+
| host1 | 1970-01-01T00:00:00 | 1.0 |
| host2 | 1970-01-01T00:00:00.001 | 2.0 |
+-------+-------------------------+-----+
-- UNSET preserve_row_sequence should succeed
ALTER TABLE test_alter_preserve_row_sequence UNSET 'preserve_row_sequence';
Affected Rows: 0
-- SHOW CREATE TABLE should keep append_mode but drop preserve_row_sequence
SHOW CREATE TABLE test_alter_preserve_row_sequence;
+----------------------------------+-----------------------------------------------------------------+
| Table | Create Table |
+----------------------------------+-----------------------------------------------------------------+
| test_alter_preserve_row_sequence | CREATE TABLE IF NOT EXISTS "test_alter_preserve_row_sequence" ( |
| | "host" STRING NULL, |
| | "ts" TIMESTAMP(3) NOT NULL, |
| | "cpu" DOUBLE NULL, |
| | TIME INDEX ("ts"), |
| | PRIMARY KEY ("host") |
| | ) |
| | |
| | ENGINE=mito |
| | WITH( |
| | append_mode = 'true' |
| | ) |
+----------------------------------+-----------------------------------------------------------------+
DROP TABLE test_alter_preserve_row_sequence;
Affected Rows: 0
-- Test that preserve_row_sequence requires append_mode=true
CREATE TABLE test_alter_preserve_no_append(
host STRING,
ts TIMESTAMP TIME INDEX,
cpu DOUBLE,
PRIMARY KEY(host)
) ENGINE=mito;
Affected Rows: 0
-- Setting preserve_row_sequence on a non-append-only table should fail
-- SQLNESS REPLACE \d+\(\d+,\s+\d+\) REDACTED
ALTER TABLE test_alter_preserve_no_append SET 'preserve_row_sequence' = 'true';
Error: 1004(InvalidArguments), Invalid region request, region_id: REDACTED, err: Invalid region options, preserve_row_sequence is only supported for append-only tables (append_mode must be true)
DROP TABLE test_alter_preserve_no_append;
Affected Rows: 0
tests/cases/standalone/common/alter/alter_preserve_row_sequence.sql
+46
View File
@@ -0,0 +1,46 @@
-- Test altering preserve_row_sequence on an append-only table, including
-- persistence across a restart.
CREATE TABLE test_alter_preserve_row_sequence(
host STRING,
ts TIMESTAMP TIME INDEX,
cpu DOUBLE,
PRIMARY KEY(host)
) ENGINE=mito WITH('append_mode'='true');
INSERT INTO test_alter_preserve_row_sequence VALUES ('host1', 0, 1.0), ('host2', 1, 2.0);
-- SET preserve_row_sequence on the append-only table should succeed
ALTER TABLE test_alter_preserve_row_sequence SET 'preserve_row_sequence' = 'true';
-- SHOW CREATE TABLE should show both append_mode and preserve_row_sequence
SHOW CREATE TABLE test_alter_preserve_row_sequence;
-- Restart the server: the option and the inserted data must survive
-- SQLNESS ARG restart=true
SHOW CREATE TABLE test_alter_preserve_row_sequence;
-- Ordinary query should still see the inserted rows after restart
SELECT * FROM test_alter_preserve_row_sequence ORDER BY host, ts;
-- UNSET preserve_row_sequence should succeed
ALTER TABLE test_alter_preserve_row_sequence UNSET 'preserve_row_sequence';
-- SHOW CREATE TABLE should keep append_mode but drop preserve_row_sequence
SHOW CREATE TABLE test_alter_preserve_row_sequence;
DROP TABLE test_alter_preserve_row_sequence;
-- Test that preserve_row_sequence requires append_mode=true
CREATE TABLE test_alter_preserve_no_append(
host STRING,
ts TIMESTAMP TIME INDEX,
cpu DOUBLE,
PRIMARY KEY(host)
) ENGINE=mito;
-- Setting preserve_row_sequence on a non-append-only table should fail
-- SQLNESS REPLACE \d+\(\d+,\s+\d+\) REDACTED
ALTER TABLE test_alter_preserve_no_append SET 'preserve_row_sequence' = 'true';
DROP TABLE test_alter_preserve_no_append;
tests/compatibility/cases/downgrade_compatibility/case.toml
+3 -3
View File
@@ -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"
tests/compatibility/cases/downgrade_compatibility/setup.sql
+23
View File
@@ -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');
tests/compatibility/cases/downgrade_compatibility/verify.result
+11
View File
@@ -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 |
+---------------------+--------+-----+
tests/compatibility/cases/downgrade_compatibility/verify.sql
+2
View File
@@ -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;