feat(mito2): support exact sequence range reads

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

src/cmd/src/bin/query_perf_fixture/direct_sst.rs

@@ -310,6 +310,7 @@ fn file_meta_from_sst_info(
         num_series: info.num_series,
         primary_key_min: None,
         primary_key_max: None,
+        preserve_row_sequence: false,
     }
 }
 

src/cmd/src/datanode/objbench.rs

@@ -259,6 +259,7 @@ impl ObjbenchCommand {
             inverted_index_config: MitoConfig::default().inverted_index,
             fulltext_index_config,
             bloom_filter_index_config: MitoConfig::default().bloom_filter_index,
+            preserve_row_sequence: false,
             #[cfg(feature = "vector_index")]
             vector_index_config: Default::default(),
         };

src/cmd/src/datanode/parquetbench.rs

@@ -324,6 +324,7 @@ impl ParquetbenchCommand {
                 num_series: 0,
                 primary_key_min: None,
                 primary_key_max: None,
+                preserve_row_sequence: false,
             },
             Arc::new(NoopFilePurger),
         );

src/mito2/src/access_layer.rs

@@ -342,6 +342,11 @@ impl AccessLayer {
         let region_id = request.metadata.region_id;
         let region_metadata = request.metadata.clone();
         let cache_manager = request.cache_manager.clone();
+        let override_sequence = if request.preserve_row_sequence {
+            None
+        } else {
+            request.max_sequence
+        };
 
         let sst_info = if let Some(write_cache) = cache_manager.write_cache() {
             // Write to the write cache.
@@ -397,14 +402,14 @@ impl AccessLayer {
                     writer
                         .write_all_flat_as_primary_key(
                             request.source,
-                            request.max_sequence,
+                            override_sequence,
                             write_opts,
                         )
                         .await?
                 }
                 FormatType::Flat => {
                     writer
-                        .write_all_flat(request.source, request.max_sequence, write_opts)
+                        .write_all_flat(request.source, override_sequence, write_opts)
                         .await?
                 }
             }
@@ -565,6 +570,8 @@ pub struct SstWriteRequest {
     pub max_sequence: Option<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

@@ -2760,6 +2760,7 @@ mod tests {
                 append_mode: false,
                 filter_deleted: true,
                 merge_mode: crate::region::options::MergeMode::LastRow,
+                sequence_range: None,
                 partition_expr_version: 0,
             }
             .build(),

src/mito2/src/cache/write_cache.rs

@@ -211,6 +211,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());
@@ -248,14 +253,14 @@ impl WriteCache {
                 writer
                     .write_all_flat_as_primary_key(
                         write_request.source,
-                        write_request.max_sequence,
+                        override_sequence,
                         write_opts,
                     )
                     .await?
             }
             crate::sst::FormatType::Flat => {
                 writer
-                    .write_all_flat(write_request.source, write_request.max_sequence, write_opts)
+                    .write_all_flat(write_request.source, override_sequence, write_opts)
                     .await?
             }
         };
@@ -553,6 +558,7 @@ mod tests {
             max_sequence: None,
             sst_write_format: Default::default(),
             cache_manager: Default::default(),
+            preserve_row_sequence: false,
             index_options: IndexOptions::default(),
             index_config: Default::default(),
             inverted_index_config: Default::default(),
@@ -657,6 +663,7 @@ mod tests {
             max_sequence: None,
             sst_write_format: Default::default(),
             cache_manager: cache_manager.clone(),
+            preserve_row_sequence: false,
             index_options: IndexOptions::default(),
             index_config: Default::default(),
             inverted_index_config: Default::default(),
@@ -751,6 +758,7 @@ mod tests {
             max_sequence: None,
             sst_write_format: Default::default(),
             cache_manager: cache_manager.clone(),
+            preserve_row_sequence: false,
             index_options: IndexOptions::default(),
             index_config: Default::default(),
             inverted_index_config: Default::default(),

src/mito2/src/compaction/compactor.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::num::NonZero;
+use std::num::{NonZero, NonZeroU64};
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -52,7 +52,7 @@ use crate::region::version::VersionRef;
 use crate::region::{ManifestContext, RegionLeaderState, RegionRoleState};
 use crate::schedule::scheduler::LocalScheduler;
 use crate::sst::FormatType;
-use crate::sst::file::{FileMeta, UncommittedSsts};
+use crate::sst::file::{FileHandle, FileMeta, UncommittedSsts};
 use crate::sst::file_purger::LocalFilePurger;
 use crate::sst::index::intermediate::IntermediateManager;
 use crate::sst::index::puffin_manager::PuffinManagerFactory;
@@ -404,6 +404,22 @@ pub trait SstMerger: Send + Sync + 'static {
 #[derive(Clone)]
 pub struct DefaultSstMerger;
 
+/// Computes the maximum sequence bound of the output of merging `inputs`.
+///
+/// Returns `None` if any input's sequence bound is unknown (`None`): a partial
+/// max over the known inputs would launder the unknown file's rows into a
+/// trusted sequence bound, letting exact sequence-range scans silently skip or
+/// wrongly filter rows from that file. Only when every input bound is known is
+/// the `max` computed.
+fn max_input_sequence(inputs: &[FileHandle]) -> Option<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 +455,17 @@ impl SstMerger for DefaultSstMerger {
             .iter()
             .map(|f| f.file_id().to_string())
             .join(",");
-        let max_sequence = output
-            .inputs
-            .iter()
-            .map(|f| f.meta_ref().sequence)
-            .max()
-            .flatten();
+        let max_sequence = max_input_sequence(&output.inputs);
+        // The compaction output can only preserve per-row sequences when the
+        // region option is enabled AND every input file carries the
+        // preserved-sequence marker. Otherwise a legacy (unmarked) input would
+        // be laundered into a trusted output and exact sequence-range scans
+        // would silently skip or wrongly filter its normalized sequences.
+        let output_preserves_sequence = compaction_region.region_options.preserve_row_sequence
+            && output
+                .inputs
+                .iter()
+                .all(|f| f.meta_ref().preserve_row_sequence);
         let builder = CompactionSstReaderBuilder {
             metadata: compaction_region.region_metadata.clone(),
             sst_layer: compaction_region.access_layer.clone(),
@@ -474,6 +495,7 @@ impl SstMerger for DefaultSstMerger {
                     } else {
                         FormatType::PrimaryKey
                     },
+                    preserve_row_sequence: output_preserves_sequence,
                     index_options,
                     index_config,
                     inverted_index_config,
@@ -527,6 +549,7 @@ impl SstMerger for DefaultSstMerger {
                     num_series: sst_info.num_series,
                     primary_key_min,
                     primary_key_max,
+                    preserve_row_sequence: output_preserves_sequence,
                 }
             })
             .collect::<Vec<_>>();
@@ -791,6 +814,33 @@ mod tests {
         FileHandle::new(meta, Arc::new(NoopFilePurger))
     }
 
+    #[test]
+    fn test_max_input_sequence() {
+        let meta_with = |sequence| FileMeta {
+            sequence,
+            ..dummy_file_meta()
+        };
+
+        // All inputs with known bounds yield their max.
+        let inputs = vec![
+            new_file_handle(meta_with(NonZeroU64::new(3))),
+            new_file_handle(meta_with(NonZeroU64::new(9))),
+            new_file_handle(meta_with(NonZeroU64::new(5))),
+        ];
+        assert_eq!(NonZeroU64::new(9), max_input_sequence(&inputs));
+
+        // A single unknown bound poisons the whole output: the known inputs'
+        // max must not be laundered onto the merged output.
+        let inputs = vec![
+            new_file_handle(meta_with(NonZeroU64::new(9))),
+            new_file_handle(meta_with(None)),
+        ];
+        assert_eq!(None, max_input_sequence(&inputs));
+
+        // No inputs at all yield `None`.
+        assert_eq!(None, max_input_sequence(&[]));
+    }
+
     /// Build a minimal [`CompactionRegion`] suitable for tests where the
     /// [`SstMerger`] is mocked and never touches the access layer.
     async fn new_test_compaction_region() -> CompactionRegion {

src/mito2/src/compaction/window.rs

@@ -336,6 +336,7 @@ mod tests {
                 max_row_group_row_count: None,
                 primary_key_encoding: None,
                 write_buffer_size: None,
+                preserve_row_sequence: false,
             },
             compaction_time_window: None,
         }

src/mito2/src/engine.rs

@@ -135,8 +135,8 @@ use crate::config::MitoConfig;
 use crate::engine::puffin_index::{IndexEntryContext, collect_index_entries_from_puffin};
 use crate::error::{
     IncrementalQueryStaleSnafu, InvalidRequestSnafu, JoinSnafu, MitoManifestInfoSnafu, RecvSnafu,
-    RegionNotFoundSnafu, Result, SerdeJsonSnafu, SerializeColumnMetadataSnafu,
-    SnapshotFenceStaleSnafu,
+    RegionNotFoundSnafu, Result, SequenceRangeUnsupportedSnafu, SerdeJsonSnafu,
+    SerializeColumnMetadataSnafu, SnapshotFenceStaleSnafu,
 };
 #[cfg(feature = "enterprise")]
 use crate::extension::BoxedExtensionRangeProviderFactory;
@@ -147,7 +147,7 @@ use crate::metrics::{
     HANDLE_REQUEST_ELAPSED, SCAN_MEMORY_EXHAUSTED_TOTAL, SCAN_MEMORY_USAGE_BYTES,
     SCAN_REQUESTS_REJECTED_TOTAL,
 };
-use crate::read::scan_region::{ScanRegion, Scanner};
+use crate::read::scan_region::{ScanRegion, Scanner, exact_sequence_range};
 use crate::read::stream::ScanBatchStream;
 use crate::region::MitoRegionRef;
 use crate::region::opener::PartitionExprFetcherRef;
@@ -1054,36 +1054,97 @@ impl EngineInner {
             request.memtable_max_sequence = Some(version_data.committed_sequence);
         }
 
-        if let Some(given_seq) = request.memtable_min_sequence {
-            let min_readable_seq = version.flushed_sequence;
-            ensure!(
-                given_seq >= min_readable_seq,
-                IncrementalQueryStaleSnafu {
-                    region_id,
-                    given_seq,
-                    min_readable_seq,
-                }
-            );
-        }
+        // Shared capability check used identically by the engine (fence
+        // relaxation) and the reader (row-level filtering) — see
+        // `read::scan_region::exact_sequence_range`. Only Flow's explicit
+        // `sequence_range` mode requests exactness; historical `memtable_only`
+        // scans never set `exact_sequence_range` and therefore keep every fence
+        // below regardless of region options.
+        let exact_sequence_range = exact_sequence_range(&request, &version);
 
-        if let Some(given_seq) = request.memtable_max_sequence
-            && !request.skip_sst_files
-        {
-            // Explicit snapshot fences that include SST reads are enforceable
-            // only while the requested upper bound is not older than the
-            // region's flushed frontier. If H has already been flushed into SST,
-            // mito cannot apply a memtable-only sequence upper bound to that
-            // SST scan, so fail and let Flow rebind the fenced repair instead
-            // of reading rows beyond H.
-            let min_enforceable_seq = version.flushed_sequence;
-            ensure!(
-                given_seq >= min_enforceable_seq,
-                SnapshotFenceStaleSnafu {
-                    region_id,
-                    given_seq,
-                    min_enforceable_seq,
+        if request.exact_sequence_range {
+            // Exact `sequence_range` mode: when the capability is available the
+            // requested (C, H] delta is enforced row-level on memtables and every
+            // SST, so the lower/upper fences can be relaxed. When it is not, we
+            // must fail with a structured stale/unsupported error so Flow falls
+            // back instead of silently reading an approximate superset.
+            if exact_sequence_range.is_none() {
+                if let Some(given_seq) = request.memtable_min_sequence {
+                    let min_readable_seq = version.flushed_sequence;
+                    ensure!(
+                        given_seq >= min_readable_seq,
+                        IncrementalQueryStaleSnafu {
+                            region_id,
+                            given_seq,
+                            min_readable_seq,
+                        }
+                    );
                 }
-            );
+
+                if let Some(given_seq) = request.memtable_max_sequence {
+                    let min_enforceable_seq = version.flushed_sequence;
+                    ensure!(
+                        given_seq >= min_enforceable_seq,
+                        SnapshotFenceStaleSnafu {
+                            region_id,
+                            given_seq,
+                            min_enforceable_seq,
+                        }
+                    );
+                }
+
+                // Both bounds are enforceable against the flushed frontier, yet the
+                // region cannot serve an exact row-level delta (preserve option off,
+                // or a file without the preserved-sequence marker). Main semantics
+                // would silently return rows outside (C, H], so fail instead.
+                return SequenceRangeUnsupportedSnafu {
+                    region_id,
+                    min_seq: request.memtable_min_sequence.unwrap_or_default(),
+                    max_seq: request.memtable_max_sequence.unwrap_or_default(),
+                    reason: if version.options.preserve_row_sequence {
+                        "region has files without preserved per-row sequences".to_string()
+                    } else {
+                        "region does not preserve per-row sequences (preserve_row_sequence is off)"
+                            .to_string()
+                    },
+                }
+                .fail();
+            }
+        } else {
+            // Non-exact mode (including historical `memtable_only` scans): keep
+            // main's fences exactly as-is. The preserve option never relaxes them
+            // because `exact_sequence_range` is the only explicit exact intent.
+            if let Some(given_seq) = request.memtable_min_sequence {
+                let min_readable_seq = version.flushed_sequence;
+                ensure!(
+                    given_seq >= min_readable_seq,
+                    IncrementalQueryStaleSnafu {
+                        region_id,
+                        given_seq,
+                        min_readable_seq,
+                    }
+                );
+            }
+
+            if let Some(given_seq) = request.memtable_max_sequence
+                && !request.skip_sst_files
+            {
+                // Explicit snapshot fences that include SST reads are enforceable
+                // only while the requested upper bound is not older than the
+                // region's flushed frontier. If H has already been flushed into SST,
+                // mito cannot apply a memtable-only sequence upper bound to that
+                // SST scan, so fail and let Flow rebind the fenced repair instead
+                // of reading rows beyond H.
+                let min_enforceable_seq = version.flushed_sequence;
+                ensure!(
+                    given_seq >= min_enforceable_seq,
+                    SnapshotFenceStaleSnafu {
+                        region_id,
+                        given_seq,
+                        min_enforceable_seq,
+                    }
+                );
+            }
         }
 
         // Get cache.

src/mito2/src/engine/alter_test.rs

@@ -22,6 +22,7 @@ use api::v1::helper::{row, tag_column_schema};
 use api::v1::value::ValueData;
 use api::v1::{ArrowIpc, BulkWalEntry, ColumnDataType, Row, Rows, SemanticType, Value, WalEntry};
 use common_error::ext::ErrorExt;
+use common_error::status_code::StatusCode;
 use common_meta::ddl::utils::{parse_column_metadatas, parse_manifest_infos_from_extensions};
 use common_recordbatch::{DfRecordBatch, RecordBatches};
 use common_test_util::flight::encode_to_flight_data;
@@ -37,6 +38,7 @@ use store_api::region_engine::{RegionEngine, RegionManifestInfo, RegionRole};
 use store_api::region_request::{
     AddColumn, AddColumnLocation, AlterKind, PathType, RegionAlterRequest,
     RegionBulkInsertsRequest, RegionOpenRequest, RegionRequest, SetIndexOption, SetRegionOption,
+    UnsetRegionOption,
 };
 use store_api::storage::{ColumnId, RegionId, ScanRequest};
 
@@ -1985,6 +1987,287 @@ async fn test_alter_region_append_mode_invalid() {
     check_append_mode(&engine, true);
 }
 
+#[tokio::test]
+async fn test_alter_region_preserve_row_sequence_lifecycle() {
+    common_telemetry::init_default_ut_logging();
+
+    let mut env = TestEnv::with_prefix("test_alter_region_preserve_row_sequence_lifecycle").await;
+    let engine = env.create_engine(MitoConfig::default()).await;
+
+    let region_id = RegionId::new(1, 1);
+    let request = CreateRequestBuilder::new()
+        .insert_option("append_mode", "true")
+        .build();
+    let column_schemas = rows_schema(&request);
+    engine
+        .handle_request(region_id, RegionRequest::Create(request))
+        .await
+        .unwrap();
+
+    let alter = |kind: AlterKind| {
+        engine.handle_request(region_id, RegionRequest::Alter(RegionAlterRequest { kind }))
+    };
+
+    // Fill the memtable while preserve_row_sequence is disabled.
+    put_rows(
+        &engine,
+        region_id,
+        Rows {
+            schema: column_schemas.clone(),
+            rows: build_rows(0, 3),
+        },
+    )
+    .await;
+    let version = engine.get_region(region_id).unwrap().version();
+    assert!(version.memtables.num_rows() > 0);
+    let flushed_sequence = version.flushed_sequence;
+
+    // Enable alone on the non-empty memtable: applies in memory through the
+    // fast path, neither flushing nor losing rows.
+    alter(AlterKind::SetRegionOptions {
+        options: vec![SetRegionOption::PreserveRowSequence(true)],
+    })
+    .await
+    .unwrap();
+    let version = engine.get_region(region_id).unwrap().version();
+    assert!(version.options.preserve_row_sequence);
+    assert_eq!(0, version.ssts.levels()[0].files.len());
+    assert!(version.memtables.num_rows() > 0);
+    assert_eq!(flushed_sequence, version.flushed_sequence);
+
+    // Rows written after the ALTER share the pre-ALTER memtable; a later flush
+    // writes both into one SST marked with preserve_row_sequence.
+    put_rows(
+        &engine,
+        region_id,
+        Rows {
+            schema: column_schemas.clone(),
+            rows: build_rows(3, 5),
+        },
+    )
+    .await;
+    flush_region(&engine, region_id, None).await;
+    let version = engine.get_region(region_id).unwrap().version();
+    assert!(version.options.preserve_row_sequence);
+    let level0 = &version.ssts.levels()[0].files;
+    assert_eq!(1, level0.len());
+    assert!(
+        level0
+            .values()
+            .next()
+            .unwrap()
+            .meta_ref()
+            .preserve_row_sequence
+    );
+    let flushed_sequence = version.flushed_sequence;
+
+    // Exact scan (2, 5] returns pre-ALTER row "2" (seq 3) and post-ALTER rows
+    // "3" and "4" (seqs 4 and 5): distinct row sequences survive the flush.
+    let scan_exact = async |min: Option<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/edit_region_test.rs

@@ -563,6 +563,42 @@ async fn test_region_edit_with_file_sequence_is_not_merged() {
     assert_eq!(Some(3), region_file_sequence(&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/error.rs

@@ -272,6 +272,22 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display(
+        "SEQUENCE_RANGE_UNSUPPORTED: exact sequence-range read unsupported, region: {}, min_seq: {}, max_seq: {}, reason: {}, retry_hint: FALLBACK_MEMTABLE_ONLY_OR_FULL_RECOMPUTE",
+        region_id,
+        min_seq,
+        max_seq,
+        reason
+    ))]
+    SequenceRangeUnsupported {
+        region_id: RegionId,
+        min_seq: u64,
+        max_seq: u64,
+        reason: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("Old manifest missing for region {}", region_id))]
     MissingOldManifest {
         region_id: RegionId,
@@ -1449,6 +1465,8 @@ impl ErrorExt for Error {
 
             IncrementalQueryStale { .. } | SnapshotFenceStale { .. } => StatusCode::RequestOutdated,
 
+            SequenceRangeUnsupported { .. } => StatusCode::Unsupported,
+
             RegionMetadataNotFound { .. }
             | Join { .. }
             | WorkerStopped { .. }

src/mito2/src/flush.rs

@@ -661,6 +661,7 @@ impl RegionFlushTask {
                         sst_info,
                         partition_expr.clone(),
                         pk_range,
+                        version.options.preserve_row_sequence,
                     ));
                 }
                 if hook.is_some() {
@@ -783,6 +784,7 @@ impl RegionFlushTask {
         sst_info: &SstInfo,
         partition_expr: Option<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)),
@@ -806,6 +808,7 @@ impl RegionFlushTask {
             num_series: sst_info.num_series,
             primary_key_min,
             primary_key_max,
+            preserve_row_sequence,
         }
     }
 
@@ -832,6 +835,7 @@ impl RegionFlushTask {
             } else {
                 FormatType::PrimaryKey
             },
+            preserve_row_sequence: version.options.preserve_row_sequence,
             index_options: self.index_options.clone(),
             index_config: self.engine_config.index.clone(),
             inverted_index_config: self.engine_config.inverted_index.clone(),

src/mito2/src/read/range_cache.rs

@@ -31,7 +31,7 @@ use datatypes::value::scalar_value_to_timestamp;
 use futures::TryStreamExt;
 use snafu::ResultExt;
 use store_api::region_engine::PartitionRange;
-use store_api::storage::{FileId, RegionId, TimeSeriesRowSelector};
+use store_api::storage::{FileId, RegionId, SequenceRange, TimeSeriesRowSelector};
 use table::predicate::is_string_timestamp_literal;
 use tokio::sync::{mpsc, oneshot};
 
@@ -63,6 +63,10 @@ pub(crate) struct ScanRequestFingerprint {
     append_mode: bool,
     filter_deleted: bool,
     merge_mode: MergeMode,
+    /// Exact sequence range applied row-level on SST reads, or `None` for scans
+    /// without row-level sequence filtering. Kept in the cache key so filtered
+    /// range results are never reused for scans with a different (or no) range.
+    sequence_range: Option<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,
         }
@@ -860,6 +870,7 @@ pub fn bench_cache_flat_range_stream(
         append_mode: false,
         filter_deleted: false,
         merge_mode: MergeMode::LastRow,
+        sequence_range: None,
         partition_expr_version: 0,
     }
     .build();
@@ -926,6 +937,7 @@ mod tests {
             append_mode: false,
             filter_deleted,
             merge_mode: MergeMode::LastRow,
+            sequence_range: None,
             partition_expr_version,
         }
         .build()

src/mito2/src/read/scan_region.rs

@@ -55,7 +55,9 @@ use tokio_stream::wrappers::ReceiverStream;
 use crate::access_layer::AccessLayerRef;
 use crate::cache::CacheStrategy;
 use crate::config::DEFAULT_MAX_CONCURRENT_SCAN_FILES;
-use crate::error::{InvalidPartitionExprSnafu, InvalidRequestSnafu, Result};
+use crate::error::{
+    InvalidPartitionExprSnafu, InvalidRequestSnafu, Result, SequenceRangeUnsupportedSnafu,
+};
 #[cfg(feature = "enterprise")]
 use crate::extension::{BoxedExtensionRange, BoxedExtensionRangeProvider};
 use crate::memtable::{MemtableRange, RangesOptions};
@@ -84,6 +86,7 @@ use crate::sst::index::inverted_index::applier::builder::InvertedIndexApplierBui
 use crate::sst::index::vector_index::applier::{VectorIndexApplier, VectorIndexApplierRef};
 use crate::sst::parquet::file_range::PreFilterMode;
 use crate::sst::parquet::reader::ReaderMetrics;
+use crate::sst::version::SstVersion;
 
 #[cfg(feature = "vector_index")]
 const VECTOR_INDEX_OVERFETCH_MULTIPLIER: usize = 2;
@@ -415,6 +418,23 @@ impl ScanRegion {
     async fn scan_input(self) -> Result<ScanInput> {
         let metadata = &self.version.metadata;
         let sst_min_sequence = self.request.sst_min_sequence.and_then(NonZeroU64::new);
+        // Exact `sequence_range` mode filters rows row-level on every
+        // time-matching SST. The legacy `sst_min_sequence` file-pruning hint
+        // would silently skip preserved files whose max sequence does not
+        // exceed it (e.g. `(5, 10]` with `sst_min_sequence = 8` drops a file
+        // with max = 8, losing sequences 6..8), so the conflicting combination
+        // is rejected explicitly instead of being silently approximated.
+        if sst_min_sequence.is_some() && self.exact_sequence_range().is_some() {
+            return SequenceRangeUnsupportedSnafu {
+                region_id: self.region_id(),
+                min_seq: self.request.memtable_min_sequence.unwrap_or_default(),
+                max_seq: self.request.memtable_max_sequence.unwrap_or_default(),
+                reason:
+                    "sst_min_sequence pruning hint is incompatible with exact sequence-range reads"
+                        .to_string(),
+            }
+            .fail();
+        }
         let time_range = self.build_time_range_predicate();
         let predicate = PredicateGroup::new(metadata, &self.request.filters)?;
 
@@ -468,10 +488,27 @@ impl ScanRegion {
         };
 
         let ssts = &self.version.ssts;
+        // Exact `(C, H]` scans skip files whose max sequence cannot reach the
+        // lower bound C: the whole file predates the range. This applies to
+        // marked files (no in-range rows) and to unmarked files already proven
+        // disjoint by `files_allow_exact_sequence_range` (their rows must not
+        // pass through unfiltered).
+        let sequence_range = self.exact_sequence_range();
+        let exact_min_sequence = match &sequence_range {
+            Some(SequenceRange::GtLtEq { min, .. }) => Some(*min),
+            _ => None,
+        };
         let mut files = Vec::new();
         if !self.request.skip_sst_files {
             for level in ssts.levels() {
                 for file in level.files.values() {
+                    if let Some(min) = exact_min_sequence
+                        && let Some(file_sequence) = file.meta_ref().sequence
+                        && file_sequence.get() <= min
+                    {
+                        continue;
+                    }
+
                     let exceed_min_sequence = match (sst_min_sequence, file.meta_ref().sequence) {
                         (Some(min_sequence), Some(file_sequence)) => file_sequence > min_sequence,
                         // If the file's sequence is None (or actually is zero), it could mean the file
@@ -588,6 +625,7 @@ impl ScanRegion {
                     .then_some(self.request.memtable_max_sequence)
                     .flatten(),
             )
+            .with_sequence_range(sequence_range)
             .with_query_stat_counters(self.query_stat_counters);
         #[cfg(feature = "vector_index")]
         let input = input
@@ -715,6 +753,14 @@ impl ScanRegion {
         self.version.metadata.region_id
     }
 
+    /// Returns the exact sequence range for row-level sequence filtering, or
+    /// `None` when the scan is not an exact `sequence_range` request or the
+    /// region cannot guarantee exactness (no per-row sequences preserved, or a
+    /// file in the version lacks the preserved-sequence marker).
+    fn exact_sequence_range(&self) -> Option<SequenceRange> {
+        exact_sequence_range(&self.request, &self.version)
+    }
+
     /// Build time range predicate from filters.
     fn build_time_range_predicate(&self) -> TimestampRange {
         let time_index = self.version.metadata.time_index_column();
@@ -933,6 +979,13 @@ pub struct ScanInput {
     explain_flat_format: bool,
     /// Snapshot upper bound bound at scan open and propagated back to the caller.
     pub(crate) snapshot_sequence: Option<SequenceNumber>,
+    /// Exact sequence range to filter rows row-level on SST reads.
+    ///
+    /// Set only when the region preserves per-row sequences
+    /// (`preserve_row_sequence` on an append-only table) and the scan
+    /// carries an explicit exact range. When set, SST readers apply the same
+    /// `(checkpoint, upper_bound]` row-level filter as memtables.
+    pub(crate) sequence_range: Option<SequenceRange>,
     /// Whether this scan is for compaction.
     pub(crate) compaction: bool,
     /// Counters that should receive query-load metrics.
@@ -973,6 +1026,7 @@ impl ScanInput {
             distribution: None,
             explain_flat_format: false,
             snapshot_sequence: None,
+            sequence_range: None,
             compaction: false,
             query_stat_counters: None,
             #[cfg(feature = "enterprise")]
@@ -1165,6 +1219,12 @@ impl ScanInput {
         self
     }
 
+    #[must_use]
+    pub(crate) fn with_sequence_range(mut self, sequence_range: Option<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 {
@@ -1583,6 +1643,52 @@ fn pre_filter_mode(append_mode: bool, merge_mode: MergeMode) -> PreFilterMode {
     }
 }
 
+/// Returns true if every unmarked SST in `ssts` can be excluded from an exact
+/// `(min_seq, max]` scan: it must carry a known max sequence not exceeding the
+/// lower bound `min_seq`, so none of its rows can fall inside the range.
+/// Marked files always qualify (the reader filters their rows row-level);
+/// unmarked files with an unknown max sequence fail closed.
+pub(crate) fn files_allow_exact_sequence_range(ssts: &SstVersion, min_seq: SequenceNumber) -> bool {
+    ssts.levels().iter().all(|level| {
+        level.files.values().all(|file| {
+            let meta = file.meta_ref();
+            meta.preserve_row_sequence || meta.sequence.is_some_and(|seq| seq.get() <= min_seq)
+        })
+    })
+}
+
+/// Single source of truth for the exact row-level sequence-range capability.
+///
+/// Returns the exact `GtLtEq { min, max }` range when **all** of the following
+/// hold — and only then — so the engine and the reader can never drift:
+/// - the scan explicitly requests `exact_sequence_range` (Flow's
+///   `sequence_range` mode; historical `memtable_only` scans never set it), and
+/// - the scan does not skip SST files (exactness requires reading them), and
+/// - the region preserves per-row sequences (`preserve_row_sequence`),
+///   and
+/// - every unmarked SST carries a known max sequence not exceeding the lower
+///   bound `min`, so it cannot contribute rows to `(min, max]`.
+///
+/// Returns `None` otherwise. Callers must keep main's fences/fallback semantics
+/// and never silently approximate.
+pub(crate) fn exact_sequence_range(
+    request: &ScanRequest,
+    version: &crate::region::version::Version,
+) -> Option<SequenceRange> {
+    if !request.exact_sequence_range
+        || request.skip_sst_files
+        || !version.options.preserve_row_sequence
+    {
+        return None;
+    }
+    let min = request.memtable_min_sequence?;
+    let max = request.memtable_max_sequence?;
+    if !files_allow_exact_sequence_range(&version.ssts, min) {
+        return None;
+    }
+    Some(SequenceRange::GtLtEq { min, max })
+}
+
 fn json_nested_paths(column_name: &str, json_type: &JsonNativeType) -> Vec<NestedPath> {
     let mut paths = Vec::new();
     let mut current = vec![column_name.to_string()];
@@ -1715,6 +1821,7 @@ pub(crate) fn build_scan_fingerprint(input: &ScanInput) -> Option<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();
@@ -2290,6 +2397,7 @@ mod tests {
             append_mode: false,
             filter_deleted: false,
             merge_mode: MergeMode::LastNonNull,
+            sequence_range: None,
             partition_expr_version: 0,
         }
         .build();
@@ -2363,6 +2471,7 @@ mod tests {
             append_mode: false,
             filter_deleted: true,
             merge_mode: MergeMode::LastRow,
+            sequence_range: None,
             partition_expr_version: metadata.partition_expr_version,
         }
         .build();
@@ -2579,4 +2688,39 @@ mod tests {
             assert_eq!(expected_mode, input.range_pre_filter_mode(source_count));
         }
     }
+
+    #[test]
+    fn test_files_allow_exact_sequence_range() {
+        use std::num::NonZeroU64;
+
+        use crate::test_util::new_noop_file_purger;
+
+        let purger = new_noop_file_purger();
+        let file = |sequence: Option<NonZeroU64>, marked: bool| FileMeta {
+            sequence,
+            preserve_row_sequence: marked,
+            ..Default::default()
+        };
+
+        // Marked file (always OK) + unmarked files with known max sequences.
+        let mut ssts = SstVersion::new();
+        ssts.add_files(
+            purger.clone(),
+            [
+                file(NonZeroU64::new(5), true),
+                file(NonZeroU64::new(5), false),
+                file(NonZeroU64::new(9), false),
+            ]
+            .into_iter(),
+        );
+        // C=5: the unmarked file with max 9 overlaps (5, H] -> not allowed.
+        assert!(!files_allow_exact_sequence_range(&ssts, 5));
+        // C=9: every unmarked file's max <= C -> allowed.
+        assert!(files_allow_exact_sequence_range(&ssts, 9));
+
+        // Unknown max sequence fails closed even when C is large.
+        let mut ssts = SstVersion::new();
+        ssts.add_files(purger, [file(None, false)].into_iter());
+        assert!(!files_allow_exact_sequence_range(&ssts, 100));
+    }
 }

src/mito2/src/read/scan_util.rs

@@ -29,9 +29,10 @@ use datatypes::timestamp::timestamp_array_to_primitive;
 use futures::Stream;
 use prometheus::IntGauge;
 use smallvec::SmallVec;
-use store_api::storage::RegionId;
+use snafu::ResultExt;
+use store_api::storage::{RegionId, SequenceRange};
 
-use crate::error::Result;
+use crate::error::{ComputeArrowSnafu, Result};
 use crate::memtable::MemScanMetrics;
 use crate::metrics::{
     IN_PROGRESS_SCAN, PRECISE_FILTER_ROWS_TOTAL, READ_BATCHES_RETURN, READ_ROW_GROUPS_TOTAL,
@@ -48,7 +49,7 @@ use crate::sst::index::bloom_filter::applier::BloomFilterIndexApplyMetrics;
 use crate::sst::index::fulltext_index::applier::FulltextIndexApplyMetrics;
 use crate::sst::index::inverted_index::applier::InvertedIndexApplyMetrics;
 use crate::sst::parquet::file_range::{FileRange, PreFilterMode};
-use crate::sst::parquet::flat_format::time_index_column_index;
+use crate::sst::parquet::flat_format::{sequence_column_index, time_index_column_index};
 use crate::sst::parquet::reader::{MetadataCacheMetrics, ReaderFilterMetrics, ReaderMetrics};
 use crate::sst::parquet::row_group::ParquetFetchMetrics;
 use crate::sst::parquet::{DEFAULT_READ_BATCH_SIZE, DEFAULT_ROW_GROUP_SIZE};
@@ -1501,6 +1502,49 @@ pub(crate) async fn scan_flat_file_ranges(
     ))
 }
 
+/// Filters a flat-format record batch by the exact sequence range.
+///
+/// Returns `None` when the entire batch is filtered out. The sequence column is
+/// the second-to-last internal column of the flat format
+/// (`__primary_key`, `__sequence`, `__op_type`), so it must be applied before any
+/// projection/compat conversion that drops internal columns.
+///
+/// `file_preserves_sequence` gates the filter per file: only files written with
+/// preserved per-row sequences (the `preserve_row_sequence` marker) may be
+/// row-level sequence filtered. Without the marker, rows are passed through
+/// untouched to keep main's file-level semantics.
+fn filter_flat_batch_by_sequence(
+    record_batch: RecordBatch,
+    sequence_range: Option<SequenceRange>,
+    file_preserves_sequence: bool,
+) -> Result<Option<RecordBatch>> {
+    let Some(sequence) = sequence_range else {
+        return Ok(Some(record_batch));
+    };
+    if !file_preserves_sequence {
+        return Ok(Some(record_batch));
+    }
+
+    let num_rows = record_batch.num_rows();
+    if num_rows == 0 {
+        return Ok(Some(record_batch));
+    }
+    let sequence_column = record_batch.column(sequence_column_index(record_batch.num_columns()));
+    let predicate = sequence
+        .filter(sequence_column)
+        .context(ComputeArrowSnafu)?;
+    let select_count = predicate.true_count();
+    if select_count == 0 {
+        return Ok(None);
+    }
+    if select_count == num_rows {
+        return Ok(Some(record_batch));
+    }
+    let filtered_batch = datatypes::arrow::compute::filter_record_batch(&record_batch, &predicate)
+        .context(ComputeArrowSnafu)?;
+    Ok(Some(filtered_batch))
+}
+
 /// Build the stream of scanning the input [`FileRange`]s using flat reader that returns RecordBatch.
 #[tracing::instrument(
     skip_all,
@@ -1527,7 +1571,19 @@ pub fn build_flat_file_range_scan_stream(
             let build_reader_start = Instant::now();
             let Some(mut reader) = range
                 .flat_reader(
-                    stream_ctx.input.series_row_selector,
+                    // In exact `sequence_range` mode the row-group-level LastRow
+                    // shortcut would reduce each row group to its last-timestamp
+                    // row *before* the row-level sequence filter runs, silently
+                    // dropping in-range rows (a series with seq 1 at t1 and seq 2
+                    // at t2 under `(0, 1]` keeps only seq 2 and then filters it
+                    // out). Bypass the shortcut so the final per-row selector
+                    // (`FlatLastRowReader`, applied after source merging and the
+                    // sequence filter) selects on the filtered rows instead.
+                    if stream_ctx.input.sequence_range.is_some() {
+                        None
+                    } else {
+                        stream_ctx.input.series_row_selector
+                    },
                     fetch_metrics.as_deref(),
                 )
                 .await?
@@ -1548,6 +1604,14 @@ pub fn build_flat_file_range_scan_stream(
                     record_batch
                 };
 
+                let Some(record_batch) = filter_flat_batch_by_sequence(
+                    record_batch,
+                    stream_ctx.input.sequence_range,
+                    range.file_handle().meta_ref().preserve_row_sequence,
+                )? else {
+                    continue;
+                };
+
                 if let Some(flat_compat) = may_compat {
                     let batch = flat_compat.compat(record_batch)?;
                     yield batch;
@@ -2033,3 +2097,108 @@ mod tests {
         assert_eq!(split_ts(&[42]), vec![vec![42]]);
     }
 }
+
+#[cfg(test)]
+mod sequence_filter_tests {
+    use std::sync::Arc;
+
+    use datatypes::arrow::array::{Int64Array, StringArray, UInt8Array, UInt64Array};
+    use datatypes::arrow::datatypes::{DataType, Field, Schema};
+    use datatypes::arrow::record_batch::RecordBatch;
+    use store_api::storage::SequenceRange;
+
+    use super::filter_flat_batch_by_sequence;
+
+    /// Builds a flat-format record batch: `(tag, field, ts, __primary_key, __sequence, __op_type)`.
+    fn batch(sequences: &[u64]) -> RecordBatch {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("tag_0", DataType::Utf8, false),
+            Field::new("field_0", DataType::Int64, false),
+            Field::new(
+                "ts",
+                DataType::Timestamp(datatypes::arrow::datatypes::TimeUnit::Millisecond, None),
+                false,
+            ),
+            Field::new("__primary_key", DataType::UInt8, false),
+            Field::new("__sequence", DataType::UInt64, false),
+            Field::new("__op_type", DataType::UInt8, false),
+        ]));
+        let tags = StringArray::from_iter_values((0..sequences.len()).map(|i| i.to_string()));
+        let fields = Int64Array::from_iter_values(0..sequences.len() as i64);
+        let ts = datatypes::arrow::array::TimestampMillisecondArray::from_iter_values(
+            (0..sequences.len()).map(|i| i as i64 * 1000),
+        );
+        let pk = UInt8Array::from(vec![0u8; sequences.len()]);
+        let seq = UInt64Array::from_iter_values(sequences.iter().copied());
+        let op = UInt8Array::from(vec![0u8; sequences.len()]);
+        RecordBatch::try_new(
+            schema,
+            vec![
+                Arc::new(tags),
+                Arc::new(fields),
+                Arc::new(ts),
+                Arc::new(pk),
+                Arc::new(seq),
+                Arc::new(op),
+            ],
+        )
+        .unwrap()
+    }
+
+    fn remaining_tags(batch: &RecordBatch) -> Vec<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);
+    }
+}

src/mito2/src/region/options.rs

@@ -118,6 +118,19 @@ pub struct RegionOptions {
     /// the configured size; zero disables both limits.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub write_buffer_size: Option<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,34 @@ mod tests {
             max_row_group_row_count: None,
             primary_key_encoding: None,
             write_buffer_size: None,
+            preserve_row_sequence: false,
         };
         assert_eq!(expect, options);
     }
 
+    #[test]
+    fn test_with_preserve_row_sequence() {
+        // Option requires append mode.
+        let map = make_map(&[("append_mode", "false"), ("preserve_row_sequence", "true")]);
+        let err = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap_err();
+        assert_eq!(StatusCode::InvalidArguments, err.status_code());
+        assert!(err.to_string().contains("preserve_row_sequence"));
+
+        let map = make_map(&[("append_mode", "true"), ("preserve_row_sequence", "true")]);
+        let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
+        assert!(options.append_mode);
+        assert!(options.preserve_row_sequence);
+
+        // Default is false and disabled keeps main behavior.
+        let map = make_map(&[("append_mode", "true")]);
+        let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
+        assert!(!options.preserve_row_sequence);
+
+        let map = make_map(&[]);
+        let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
+        assert_eq!(RegionOptions::default(), options);
+    }
+
     #[test]
     fn test_region_options_serde() {
         let options = RegionOptions {
@@ -1002,6 +1051,7 @@ mod tests {
             max_row_group_row_count: None,
             primary_key_encoding: None,
             write_buffer_size: Some(ReadableSize::mb(128)),
+            preserve_row_sequence: false,
         };
         let region_options_json_str = serde_json::to_string(&options).unwrap();
         let got: RegionOptions = serde_json::from_str(&region_options_json_str).unwrap();
@@ -1069,10 +1119,29 @@ mod tests {
             max_row_group_row_count: None,
             primary_key_encoding: None,
             write_buffer_size: None,
+            preserve_row_sequence: false,
         };
         assert_eq!(options, got);
     }
 
+    #[test]
+    fn test_region_options_serde_preserve_row_sequence_roundtrip() {
+        let options = RegionOptions {
+            append_mode: true,
+            preserve_row_sequence: true,
+            ..Default::default()
+        };
+        let json = serde_json::to_string(&options).unwrap();
+        assert!(json.contains("preserve_row_sequence"));
+        let got: RegionOptions = serde_json::from_str(&json).unwrap();
+        assert_eq!(options, got);
+
+        // Old manifests without the key default to false.
+        let old_json = r#"{"append_mode": true}"#;
+        let got: RegionOptions = serde_json::from_str(old_json).unwrap();
+        assert!(!got.preserve_row_sequence);
+    }
+
     #[test]
     fn test_max_row_group_row_count() {
         // Default falls back to DEFAULT_ROW_GROUP_SIZE.

src/mito2/src/sst/file.rs

@@ -282,6 +282,18 @@ 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.
+    ///
+    /// Set when the file is written by a flush or compaction of a region with
+    /// `preserve_row_sequence` enabled (append-only tables). Absent
+    /// or `false` in legacy files, which must not be row-level sequence filtered.
+    #[serde(default, skip_serializing_if = "is_false")]
+    pub preserve_row_sequence: bool,
+}
+
+fn is_false(value: &bool) -> bool {
+    !*value
 }
 
 impl Debug for FileMeta {
@@ -1116,6 +1128,30 @@ mod tests {
             deserialized_file_meta.file_id,
             FileId::from_str("bc5896ec-e4d8-4017-a80d-f2de73188d55").unwrap()
         );
+        assert!(!deserialized_file_meta.preserve_row_sequence);
+    }
+
+    #[test]
+    fn test_file_meta_preserve_row_sequence_serde() {
+        let file_meta = FileMeta {
+            preserve_row_sequence: true,
+            ..Default::default()
+        };
+
+        let serialized = serde_json::to_string(&file_meta).unwrap();
+        let value: serde_json::Value = serde_json::from_str(&serialized).unwrap();
+        assert_eq!(value["preserve_row_sequence"], true);
+
+        let deserialized: FileMeta = serde_json::from_str(&serialized).unwrap();
+        assert_eq!(file_meta, deserialized);
+
+        let file_meta_false = FileMeta {
+            preserve_row_sequence: false,
+            ..file_meta.clone()
+        };
+        let serialized_false = serde_json::to_string(&file_meta_false).unwrap();
+        let value_false: serde_json::Value = serde_json::from_str(&serialized_false).unwrap();
+        assert!(value_false.get("preserve_row_sequence").is_none());
     }
     #[test]
     fn test_is_index_consistent_with_region() {

src/mito2/src/sst/index.rs

@@ -1687,6 +1687,7 @@ mod tests {
             max_sequence: None,
             sst_write_format: Default::default(),
             cache_manager: Default::default(),
+            preserve_row_sequence: false,
             index_options: IndexOptions::default(),
             index_config,
             inverted_index_config: Default::default(),

src/mito2/src/worker/handle_alter.rs

@@ -196,7 +196,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(
@@ -300,6 +304,19 @@ impl<S: LogStore> RegionWorkerLoop<S> {
                         all_options_altered = false;
                     }
                 }
+                SetRegionOption::PreserveRowSequence(new_preserve) => {
+                    // Toggling preserve_row_sequence only affects in-memory
+                    // RegionOptions and how future flushes/compactions mark SST
+                    // metadata; it does not require an empty memtable, so it
+                    // never blocks the fast path.
+                    if new_preserve != current_options.preserve_row_sequence {
+                        info!(
+                            "Update region preserve_row_sequence: {}, previous: {:?} new: {:?}",
+                            region.region_id, current_options.preserve_row_sequence, new_preserve
+                        );
+                        current_options.preserve_row_sequence = new_preserve;
+                    }
+                }
                 SetRegionOption::SkipWal => {
                     if !current_options.skip_wal {
                         info!("Stop writing WAL for region: {}", region.region_id);
@@ -308,15 +325,33 @@ impl<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);
+            // No option independently requires an empty memtable
+            // (preserve_row_sequence toggles apply in memory): apply the staged
+            // options directly without flushing.
+            region.version_control.alter_options(candidate);
             Ok(None)
         } else {
-            let kind = AlterKind::SetRegionOptions { options };
-            Ok(new_region_options_on_empty_memtable(
-                &current_options,
-                &kind,
-            ))
+            // Some option independently requires an empty memtable (e.g.
+            // append_mode, sst_format or max_row_group_row_count): stage the
+            // complete final options and follow the existing flush path.
+            Ok(Some(candidate))
         }
     }
 }
@@ -368,6 +403,9 @@ fn new_region_options_on_empty_memtable(
             SetRegionOption::MaxRowGroupRowCount(new_row_count) => {
                 current_options.max_row_group_row_count = *new_row_count;
             }
+            SetRegionOption::PreserveRowSequence(new_preserve) => {
+                current_options.preserve_row_sequence = *new_preserve;
+            }
         }
     }
     Some(current_options)

src/mito2/src/worker/handle_manifest.rs

@@ -316,9 +316,15 @@ 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
+        // 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

@@ -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

@@ -56,7 +56,7 @@ use crate::metric_engine_consts::PHYSICAL_TABLE_METADATA_KEY;
 use crate::metrics;
 use crate::mito_engine_options::{
     APPEND_MODE_KEY, AUTO_FLUSH_INTERVAL_KEY, MAX_ROW_GROUP_ROW_COUNT,
-    MAX_ROW_GROUP_ROW_COUNT_LIMIT, SKIP_WAL_KEY, SST_FORMAT_KEY, TTL_KEY,
+    MAX_ROW_GROUP_ROW_COUNT_LIMIT, PRESERVE_ROW_SEQUENCE, SKIP_WAL_KEY, SST_FORMAT_KEY, TTL_KEY,
     TWCS_MAX_OUTPUT_FILE_SIZE, TWCS_TIME_WINDOW, TWCS_TRIGGER_FILE_NUM, WRITE_BUFFER_SIZE_KEY,
 };
 use crate::path_utils::table_dir;
@@ -1458,6 +1458,8 @@ pub enum SetRegionOption {
     AutoFlushInterval(Option<Duration>),
     // Modifying the max number of rows in a parquet row group.
     MaxRowGroupRowCount(Option<usize>),
+    /// Modifying the preserve per-row sequence numbers option.
+    PreserveRowSequence(bool),
     // Stops writing new WAL entries. This operation is irreversible.
     SkipWal,
 }
@@ -1517,6 +1519,12 @@ impl TryFrom<&PbOption> for SetRegionOption {
                     .ok_or_else(|| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
                 Ok(Self::MaxRowGroupRowCount(Some(row_count)))
             }
+            PRESERVE_ROW_SEQUENCE => {
+                let preserve = value
+                    .parse::<bool>()
+                    .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
+                Ok(Self::PreserveRowSequence(preserve))
+            }
             SKIP_WAL_KEY if value == "true" => Ok(Self::SkipWal),
             _ => InvalidSetRegionOptionRequestSnafu { key, value }.fail(),
         }
@@ -1538,6 +1546,7 @@ impl From<&UnsetRegionOption> for SetRegionOption {
             UnsetRegionOption::Ttl => SetRegionOption::Ttl(Default::default()),
             UnsetRegionOption::MaxRowGroupRowCount => SetRegionOption::MaxRowGroupRowCount(None),
             UnsetRegionOption::WriteBufferSize => SetRegionOption::WriteBufferSize(None),
+            UnsetRegionOption::PreserveRowSequence => SetRegionOption::PreserveRowSequence(false),
         }
     }
 }
@@ -1553,6 +1562,7 @@ impl TryFrom<&str> for UnsetRegionOption {
             TWCS_MAX_OUTPUT_FILE_SIZE => Ok(Self::TwcsMaxOutputFileSize),
             TWCS_TIME_WINDOW => Ok(Self::TwcsTimeWindow),
             MAX_ROW_GROUP_ROW_COUNT => Ok(Self::MaxRowGroupRowCount),
+            PRESERVE_ROW_SEQUENCE => Ok(Self::PreserveRowSequence),
             _ => InvalidUnsetRegionOptionRequestSnafu { key }.fail(),
         }
     }
@@ -1566,6 +1576,8 @@ pub enum UnsetRegionOption {
     Ttl,
     MaxRowGroupRowCount,
     WriteBufferSize,
+    /// Unset the preserve per-row sequence numbers option.
+    PreserveRowSequence,
 }
 
 impl UnsetRegionOption {
@@ -1577,6 +1589,7 @@ impl UnsetRegionOption {
             Self::TwcsMaxOutputFileSize => TWCS_MAX_OUTPUT_FILE_SIZE,
             Self::TwcsTimeWindow => TWCS_TIME_WINDOW,
             Self::MaxRowGroupRowCount => MAX_ROW_GROUP_ROW_COUNT,
+            Self::PreserveRowSequence => PRESERVE_ROW_SEQUENCE,
         }
     }
 }
@@ -1985,6 +1998,37 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_set_region_option_preserve_row_sequence_try_from() {
+        for (value, expected) in [("true", true), ("false", false)] {
+            let pb = PbOption {
+                key: PRESERVE_ROW_SEQUENCE.to_string(),
+                value: value.to_string(),
+            };
+            assert_eq!(
+                SetRegionOption::PreserveRowSequence(expected),
+                SetRegionOption::try_from(&pb).unwrap()
+            );
+        }
+
+        for value in ["1", "invalid"] {
+            let pb = PbOption {
+                key: PRESERVE_ROW_SEQUENCE.to_string(),
+                value: value.to_string(),
+            };
+            assert!(SetRegionOption::try_from(&pb).is_err());
+        }
+
+        assert_eq!(
+            UnsetRegionOption::PreserveRowSequence,
+            UnsetRegionOption::try_from(PRESERVE_ROW_SEQUENCE).unwrap()
+        );
+        assert_eq!(
+            SetRegionOption::PreserveRowSequence(false),
+            (&UnsetRegionOption::PreserveRowSequence).into()
+        );
+    }
+
     #[test]
     fn test_from_proto_alter_request() {
         RegionAlterRequest::try_from(AlterRequest {

src/store-api/src/storage/requests.rs

@@ -129,6 +129,15 @@ pub struct ScanRequest {
     pub skip_sst_files: bool,
     /// Whether to bind the effective snapshot upper bound when opening the scan.
     pub snapshot_on_scan: bool,
+    /// Explicit intent to read an exact row-level sequence delta `(min, max]`
+    /// across memtables and all SST files (Flow's `sequence_range` incremental
+    /// mode). The engine performs exact row-level filtering only when the region
+    /// preserves per-row sequences and every participating SST file is trusted;
+    /// otherwise it returns a structured stale/unsupported error so the caller
+    /// falls back instead of silently approximating.
+    ///
+    /// Historical `memtable_only` reads must never set this flag.
+    pub exact_sequence_range: bool,
     /// Optional hint for the distribution of time-series data.
     pub distribution: Option<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

@@ -30,7 +30,7 @@ use snafu::{OptionExt, ResultExt, ensure};
 use store_api::metric_engine_consts::PHYSICAL_TABLE_METADATA_KEY;
 use store_api::mito_engine_options::{
     APPEND_MODE_KEY, AUTO_FLUSH_INTERVAL_KEY, COMPACTION_TYPE, COMPACTION_TYPE_TWCS,
-    MAX_ROW_GROUP_ROW_COUNT, MERGE_MODE_KEY, SKIP_WAL_KEY, SST_FORMAT_KEY,
+    MAX_ROW_GROUP_ROW_COUNT, MERGE_MODE_KEY, PRESERVE_ROW_SEQUENCE, SKIP_WAL_KEY, SST_FORMAT_KEY,
 };
 use store_api::region_request::{SetRegionOption, UnsetRegionOption};
 use store_api::storage::{ColumnDescriptor, ColumnDescriptorBuilder, ColumnId};
@@ -401,6 +401,15 @@ impl TableMeta {
                         new_options.extra_options.remove(MAX_ROW_GROUP_ROW_COUNT);
                     }
                 }
+                SetRegionOption::PreserveRowSequence(preserve) => {
+                    if *preserve {
+                        new_options
+                            .extra_options
+                            .insert(PRESERVE_ROW_SEQUENCE.to_string(), preserve.to_string());
+                    } else {
+                        new_options.extra_options.remove(PRESERVE_ROW_SEQUENCE);
+                    }
+                }
                 SetRegionOption::SkipWal => {
                     new_options.skip_wal = true;
                     // Keep the explicit table option so it remains distinguishable
@@ -1695,6 +1704,66 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_set_preserve_row_sequence_option() {
+        let schema = Arc::new(new_test_schema());
+        let meta = TableMetaBuilder::empty()
+            .schema(schema)
+            .primary_key_indices(vec![0])
+            .engine("engine")
+            .next_column_id(3)
+            .build()
+            .unwrap();
+
+        let apply = |meta: &TableMeta, kind: &AlterKind| {
+            meta.builder_with_alter_kind("my_table", kind)
+                .unwrap()
+                .build()
+                .unwrap()
+        };
+
+        let with_true = apply(
+            &meta,
+            &AlterKind::SetTableOptions {
+                options: vec![SetRegionOption::PreserveRowSequence(true)],
+            },
+        );
+        assert_eq!(
+            Some("true"),
+            with_true
+                .options
+                .extra_options
+                .get(PRESERVE_ROW_SEQUENCE)
+                .map(String::as_str)
+        );
+
+        let set_false = apply(
+            &with_true,
+            &AlterKind::SetTableOptions {
+                options: vec![SetRegionOption::PreserveRowSequence(false)],
+            },
+        );
+        assert!(
+            !set_false
+                .options
+                .extra_options
+                .contains_key(PRESERVE_ROW_SEQUENCE)
+        );
+
+        let unset = apply(
+            &with_true,
+            &AlterKind::UnsetTableOptions {
+                keys: vec![UnsetRegionOption::PreserveRowSequence],
+            },
+        );
+        assert!(
+            !unset
+                .options
+                .extra_options
+                .contains_key(PRESERVE_ROW_SEQUENCE)
+        );
+    }
+
     #[test]
     fn test_set_skip_wal_updates_typed_and_extra_options() {
         let mut meta = TableMetaBuilder::empty()

tests/compatibility/cases/downgrade_compatibility/case.toml

@@ -1,8 +1,8 @@
 name = "downgrade_compatibility"
-reason = "Verify v1.1.4 can reopen a table whose region WAL options were written by the current binary."
-introduced_by = "fix: preserve legacy region WAL options format"
+reason = "Verify v1.1.4 can reopen a table whose region WAL options were written by the current binary, and can read all rows of an append-only table flushed and compacted with preserve_row_sequence enabled."
+introduced_by = "fix: preserve legacy region WAL options format; preserve_row_sequence"
 topologies = ["distributed", "standalone"]
 from_range = [">=v1.2.0"]
 to_range = ["=v1.1.4"]
-features = ["table", "wal", "downgrade"]
+features = ["table", "wal", "downgrade", "append", "preserve_row_sequence"]
 owner = "metasrv"

tests/compatibility/cases/downgrade_compatibility/setup.sql

@@ -9,3 +9,26 @@ INSERT INTO t_downgrade_compatibility VALUES
 ('2024-02-09 00:01:00+0000', 'host_b', 2);
 
 ADMIN FLUSH_TABLE('t_downgrade_compatibility');
+
+CREATE TABLE t_preserve_sequence_downgrade(
+    ts TIMESTAMP TIME INDEX,
+    host STRING,
+    val INT,
+    PRIMARY KEY(host)
+)
+ENGINE=mito
+WITH(append_mode='true', preserve_row_sequence='true');
+
+INSERT INTO t_preserve_sequence_downgrade VALUES
+('2024-02-09 00:00:00+0000', 'host_a', 1),
+('2024-02-09 00:01:00+0000', 'host_b', 2);
+
+ADMIN FLUSH_TABLE('t_preserve_sequence_downgrade');
+
+INSERT INTO t_preserve_sequence_downgrade VALUES
+('2024-02-09 00:02:00+0000', 'host_a', 3),
+('2024-02-09 00:03:00+0000', 'host_c', 4);
+
+ADMIN FLUSH_TABLE('t_preserve_sequence_downgrade');
+
+ADMIN COMPACT_TABLE('t_preserve_sequence_downgrade');

tests/compatibility/cases/downgrade_compatibility/verify.result

@@ -6,3 +6,14 @@ SELECT ts, host, val FROM t_downgrade_compatibility ORDER BY ts, host;
 | 2024-02-09T00:00:00 | host_a | 1   |
 | 2024-02-09T00:01:00 | host_b | 2   |
 +---------------------+--------+-----+
+
+SELECT ts, host, val FROM t_preserve_sequence_downgrade ORDER BY ts, host;
+
++---------------------+--------+-----+
+| ts                  | host   | val |
++---------------------+--------+-----+
+| 2024-02-09T00:00:00 | host_a | 1   |
+| 2024-02-09T00:01:00 | host_b | 2   |
+| 2024-02-09T00:02:00 | host_a | 3   |
+| 2024-02-09T00:03:00 | host_c | 4   |
++---------------------+--------+-----+

tests/compatibility/cases/downgrade_compatibility/verify.sql

@@ -1 +1,3 @@
 SELECT ts, host, val FROM t_downgrade_compatibility ORDER BY ts, host;
+
+SELECT ts, host, val FROM t_preserve_sequence_downgrade ORDER BY ts, host;