fix(mito2): use target sequence for foreign SSTs (#8946)

* fix(mito2): use target sequence for foreign SSTs Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * chore(mito2): address foreign SST review feedback Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --------- Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
2026-09-05 21:18:57 +00:00 · 2026-08-28 03:45:41 +00:00
parent aaa843104b
commit bd7d2c1dfa
5 changed files with 466 additions and 84 deletions
@@ -12,20 +12,24 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use std::collections::HashSet;
 use std::sync::Arc;
 use std::{assert_matches, fs};

 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::datatypes::Float64Type;
 use datatypes::value::Value;
 use partition::expr::{PartitionExpr, col};
 use store_api::region_engine::{
    RegionEngine, RegionRole, RemapManifestsRequest, SettableRegionRoleState,
 };
 use store_api::region_request::{
-    ApplyStagingManifestRequest, EnterStagingRequest, RegionFlushRequest, RegionPutRequest,
-    RegionRequest, StagingPartitionDirective,
+    ApplyStagingManifestRequest, EnterStagingRequest, RegionCompactRequest, RegionFlushRequest,
+    RegionPutRequest, RegionRequest, StagingPartitionDirective,
 };
 use store_api::storage::{FileId, RegionId};

@@ -37,7 +41,9 @@ use crate::manifest::action::{
 };
 use crate::sst::FormatType;
 use crate::sst::file::FileMeta;
-use crate::test_util::{CreateRequestBuilder, TestEnv, build_rows, put_rows, rows_schema};
+use crate::test_util::{
+    CreateRequestBuilder, TestEnv, build_rows, build_rows_for_key, put_rows, rows_schema,
+};

 fn range_expr(col_name: &str, start: i64, end: i64) -> PartitionExpr {
    col(col_name)
@@ -45,6 +51,209 @@ fn range_expr(col_name: &str, start: i64, end: i64) -> PartitionExpr {
        .and(col(col_name).lt(Value::Int64(end)))
 }

+#[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;
+}
+
+async fn test_apply_staging_manifest_sequence_domain_with_format(flat_format: bool) {
+    let mut env = TestEnv::with_prefix("apply-staging-sequence-domain").await;
+    let engine = env
+        .create_engine(MitoConfig {
+            default_flat_format: flat_format,
+            ..Default::default()
+        })
+        .await;
+    let source = RegionId::new(1, 1);
+    let target = RegionId::new(1, 2);
+    let request = CreateRequestBuilder::new().build();
+    let schema = rows_schema(&request);
+
+    engine
+        .handle_request(source, RegionRequest::Create(request.clone()))
+        .await
+        .unwrap();
+    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()))
+        .await
+        .unwrap();
+    let source_manifest = engine
+        .get_region(source)
+        .unwrap()
+        .manifest_ctx
+        .manifest()
+        .await;
+    assert_eq!(source_manifest.files.len(), 1);
+    assert_eq!(
+        source_manifest.files.values().next().unwrap().sequence,
+        Some(std::num::NonZeroU64::new(3).unwrap())
+    );
+
+    engine
+        .set_region_role_state_gracefully(source, SettableRegionRoleState::StagingLeader)
+        .await
+        .unwrap();
+    let partition_expr = float_range_expr("field_0", 0.1, 100.1)
+        .as_json_str()
+        .unwrap();
+    let result = engine
+        .remap_manifests(RemapManifestsRequest {
+            region_id: source,
+            input_regions: vec![source],
+            region_mapping: [(source, vec![target])].into_iter().collect(),
+            new_partition_exprs: [(target, partition_expr.clone())].into_iter().collect(),
+        })
+        .await
+        .unwrap();
+    engine
+        .handle_request(target, RegionRequest::Create(request.clone()))
+        .await
+        .unwrap();
+    engine
+        .handle_request(
+            target,
+            RegionRequest::EnterStaging(EnterStagingRequest {
+                partition_directive: StagingPartitionDirective::UpdatePartitionExpr(
+                    partition_expr.clone(),
+                ),
+            }),
+        )
+        .await
+        .unwrap();
+    engine
+        .handle_request(
+            target,
+            RegionRequest::ApplyStagingManifest(ApplyStagingManifestRequest {
+                partition_expr,
+                central_region_id: source,
+                manifest_path: result.manifest_paths[&target].clone(),
+            }),
+        )
+        .await
+        .unwrap();
+
+    let manifest = engine
+        .get_region(target)
+        .unwrap()
+        .manifest_ctx
+        .manifest()
+        .await;
+    assert_eq!(manifest.files.len(), 1);
+    assert_eq!(manifest.committed_sequence, Some(1));
+    let imported_file = manifest.files.values().next().unwrap();
+    assert_eq!(imported_file.region_id, source);
+    assert_eq!(
+        imported_file.sequence,
+        Some(std::num::NonZeroU64::new(1).unwrap())
+    );
+
+    put_rows(
+        &engine,
+        target,
+        Rows {
+            schema: schema.clone(),
+            rows: build_rows_for_key("0", 0, 1, 99),
+        },
+    )
+    .await;
+    assert_target_value(&engine, target, 99.0).await;
+
+    engine
+        .handle_request(target, RegionRequest::Flush(RegionFlushRequest::default()))
+        .await
+        .unwrap();
+    let input_ids = current_file_ids(&engine, target);
+    assert_eq!(
+        input_ids.len(),
+        2,
+        "imported and target-write SSTs must both exist"
+    );
+
+    engine
+        .handle_request(
+            target,
+            RegionRequest::Compact(RegionCompactRequest {
+                options: compact_request::Options::StrictWindow(StrictWindow {
+                    window_seconds: 60,
+                }),
+                parallelism: None,
+                time_range: None,
+            }),
+        )
+        .await
+        .unwrap();
+    let output_version = engine.get_region(target).unwrap().version();
+    let output_files = output_version
+        .ssts
+        .levels()
+        .iter()
+        .flat_map(|level| level.files.values())
+        .collect::<Vec<_>>();
+    let output_ids = output_files
+        .iter()
+        .map(|file| file.meta_ref().file_id)
+        .collect::<HashSet<_>>();
+    assert_eq!(output_ids.len(), 1);
+    assert!(
+        output_files
+            .iter()
+            .all(|file| file.meta_ref().region_id == target)
+    );
+    assert!(
+        input_ids.iter().all(|id| !output_ids.contains(id)),
+        "real compaction must replace both input SSTs"
+    );
+    assert_target_value(&engine, target, 99.0).await;
+}
+
+fn current_file_ids(engine: &crate::engine::MitoEngine, region_id: RegionId) -> HashSet<FileId> {
+    engine
+        .get_region(region_id)
+        .unwrap()
+        .version()
+        .ssts
+        .levels()
+        .iter()
+        .flat_map(|level| level.files.values())
+        .map(|file| file.meta_ref().file_id)
+        .collect()
+}
+
+async fn assert_target_value(
+    engine: &crate::engine::MitoEngine,
+    region_id: RegionId,
+    expected: f64,
+) {
+    let scan = engine
+        .scan_to_stream(region_id, ScanRequest::default())
+        .await
+        .unwrap();
+    let batches = RecordBatches::try_collect(scan).await.unwrap();
+    assert_eq!(
+        batches.iter().map(|batch| batch.num_rows()).sum::<usize>(),
+        1
+    );
+    let batch = batches.iter().next().unwrap();
+    let values = batch
+        .column_by_name("field_0")
+        .unwrap()
+        .as_primitive::<Float64Type>();
+    assert_eq!(values.value(0), expected);
+}
+
 fn float_range_expr(col_name: &str, start: f64, end: f64) -> PartitionExpr {
    col(col_name)
        .gt_eq(Value::Float64(start.into()))
@@ -706,6 +706,14 @@ pub struct IndexBuildTask {
    pub region_id: RegionId,
    /// The SST file handle to build index for.
    pub file: FileHandle,
+    /// The target region metadata used to decode rows from the SST.
+    ///
+    /// An SST may originate in another region while being visible in the target
+    /// manifest. This metadata defines the target schema and sequence domain;
+    /// applying the staging manifest only makes imported files visible. Index
+    /// rebuild happens later when a flush, compaction, schema change, or manual
+    /// index build request schedules it.
+    pub(crate) target_region_metadata: RegionMetadataRef,
    /// The manifest state this build is based on.
    pub(crate) source: IndexBuildSource,
    pub reason: IndexBuildType,
@@ -846,6 +854,7 @@ impl IndexBuildTask {
        let mut parquet_reader = self
            .access_layer
            .read_sst(self.file.clone()) // use the latest file handle instead of creating a new one
+            .expected_metadata(Some(self.target_region_metadata.clone()))
            .build()
            .await?;

@@ -1494,8 +1503,12 @@ mod tests {
    use datatypes::schema::{
        ColumnSchema, FulltextOptions, SkippingIndexOptions, SkippingIndexType,
    };
+    use datatypes::value::Value;
+    use index::inverted_index::format::reader::InvertedIndexReader;
    use object_store::ObjectStore;
    use object_store::services::Memory;
+    use partition::expr::col;
+    use puffin::puffin_manager::{PuffinManager, PuffinReader};
    use puffin_manager::PuffinManagerFactory;
    use store_api::metadata::{ColumnMetadata, RegionMetadataBuilder};
    use tokio::sync::mpsc;
@@ -2088,6 +2101,7 @@ mod tests {
        let task = IndexBuildTask {
            region_id,
            file,
+            target_region_metadata: version_control.current().version.metadata.clone(),
            source: IndexBuildSource::new(
                file_meta,
                version_control.current().version.metadata.schema_version,
@@ -2120,16 +2134,34 @@ mod tests {
    }

    #[tokio::test]
-    async fn test_index_build_task_increments_legacy_index_version() {
+    async fn test_index_build_task_foreign_file_uses_target_metadata() {
        let env = SchedulerEnv::new().await;
        let mut scheduler = env.mock_index_build_scheduler(4);
-        let metadata = Arc::new(sst_region_metadata());
-        let manifest_ctx = env.mock_manifest_context(metadata.clone()).await;
-        let region_id = metadata.region_id;
+        let source_metadata = Arc::new(sst_region_metadata());
+        let mut target_metadata = (*source_metadata).clone();
+        target_metadata.region_id = RegionId::new(1, 3);
+        let mut target_builder = RegionMetadataBuilder::new(target_metadata.region_id);
+        for mut column_metadata in target_metadata.column_metadatas.clone() {
+            if column_metadata.column_id == 2 {
+                column_metadata.column_schema =
+                    column_metadata.column_schema.with_inverted_index(true);
+            }
+            target_builder.push_column_metadata(column_metadata);
+        }
+        let partition_expr = col("field_0")
+            .gt_eq(Value::UInt64(100))
+            .and(col("field_0").lt(Value::UInt64(200)));
+        target_builder
+            .primary_key(target_metadata.primary_key.clone())
+            .partition_expr_json(Some(partition_expr.as_json_str().unwrap()))
+            .bump_version();
+        let target_metadata = Arc::new(target_builder.build().unwrap());
+        let manifest_ctx = env.mock_manifest_context(target_metadata.clone()).await;
+        let region_id = target_metadata.region_id;
        let file_purger = Arc::new(NoopFilePurger {});
-        let sst_info = mock_sst_file(metadata.clone(), &env, IndexBuildMode::Async).await;
+        let sst_info = mock_sst_file(source_metadata.clone(), &env, IndexBuildMode::Async).await;
        let file_meta = FileMeta {
-            region_id,
+            region_id: source_metadata.region_id,
            file_id: sst_info.file_id,
            file_size: sst_info.file_size,
            max_row_group_uncompressed_size: sst_info.max_row_group_uncompressed_size,
@@ -2144,8 +2176,8 @@ mod tests {
        seed_manifest_file(&manifest_ctx, &file_meta).await;
        let files = HashMap::from([(file_meta.file_id, file_meta.clone())]);
        let version_control =
-            mock_version_control(metadata.clone(), file_purger.clone(), files).await;
-        let indexer_builder = mock_indexer_builder(metadata.clone(), &env).await;
+            mock_version_control(target_metadata.clone(), file_purger.clone(), files).await;
+        let indexer_builder = mock_indexer_builder(target_metadata.clone(), &env).await;

        let file = FileHandle::new(file_meta.clone(), file_purger.clone());

@@ -2155,6 +2187,7 @@ mod tests {
        let task = IndexBuildTask {
            region_id,
            file,
+            target_region_metadata: version_control.current().version.metadata.clone(),
            source: IndexBuildSource::new(
                file_meta.clone(),
                version_control.current().version.metadata.schema_version,
@@ -2199,9 +2232,40 @@ mod tests {
                assert!(updated_meta.index_file_size > 0);
                assert_eq!(updated_meta.file_id, file_meta.file_id);
                assert_eq!(updated_meta.index_version, 1);
+                let field_0_index = updated_meta
+                    .indexes
+                    .iter()
+                    .find(|index| index.column_id == 2)
+                    .expect("field_0 should have an inverted index");
+                assert_eq!(
+                    field_0_index.created_indexes.as_slice(),
+                    [IndexType::InvertedIndex]
+                );
            }
            _ => panic!("Unexpected worker request: {:?}", worker_req),
        }
+
+        let puffin_reader = env
+            .access_layer
+            .build_puffin_manager()
+            .reader(&RegionIndexId::new(
+                RegionFileId::new(source_metadata.region_id, file_meta.file_id),
+                1,
+            ))
+            .await
+            .unwrap();
+        let blob = puffin_reader
+            .blob(inverted_index::INDEX_BLOB_TYPE)
+            .await
+            .unwrap();
+        let blob_reader = blob.reader().await.unwrap();
+        let index_metadata =
+            index::inverted_index::format::reader::InvertedIndexBlobReader::new(blob_reader)
+                .metadata(None)
+                .await
+                .unwrap();
+        assert!(index_metadata.metas.contains_key("2"));
+        assert_eq!(index_metadata.total_row_count, 100);
    }

    async fn schedule_index_build_task_with_mode(build_mode: IndexBuildMode) {
@@ -2236,6 +2300,7 @@ mod tests {
        let task = IndexBuildTask {
            region_id,
            file,
+            target_region_metadata: version_control.current().version.metadata.clone(),
            source: IndexBuildSource::new(
                file_meta.clone(),
                version_control.current().version.metadata.schema_version,
@@ -2346,6 +2411,7 @@ mod tests {
        let task = IndexBuildTask {
            region_id,
            file,
+            target_region_metadata: version_control.current().version.metadata.clone(),
            source: IndexBuildSource::new(
                file_meta.clone(),
                version_control.current().version.metadata.schema_version,
@@ -2444,6 +2510,7 @@ mod tests {
        let task = IndexBuildTask {
            region_id,
            file,
+            target_region_metadata: version_control.current().version.metadata.clone(),
            source: IndexBuildSource::new(
                file_meta.clone(),
                version_control.current().version.metadata.schema_version,
@@ -2505,7 +2572,7 @@ mod tests {
        let schema_version = metadata.schema_version;
        let manifest_ctx = env.mock_manifest_context(metadata.clone()).await;
        let file_purger = Arc::new(NoopFilePurger {});
-        let indexer_builder = mock_indexer_builder(metadata, env).await;
+        let indexer_builder = mock_indexer_builder(metadata.clone(), env).await;
        let (tx, _rx) = mpsc::channel(4);
        let (result_tx, result_rx) = mpsc::channel::<Result<IndexBuildOutcome>>(4);

@@ -2521,6 +2588,7 @@ mod tests {
        let task = IndexBuildTask {
            region_id,
            file,
+            target_region_metadata: metadata,
            source: IndexBuildSource::new(file_meta, schema_version),
            reason,
            access_layer: env.access_layer.clone(),
@@ -118,8 +118,8 @@ mod tests {
    use datafusion_expr::{BinaryExpr, Expr, Literal, Operator, col, lit};
    use datatypes::arrow;
    use datatypes::arrow::array::{
-        ArrayRef, BinaryDictionaryBuilder, RecordBatch, StringArray, StringDictionaryBuilder,
-        TimestampMillisecondArray, UInt8Array, UInt64Array,
+        ArrayRef, AsArray, BinaryDictionaryBuilder, RecordBatch, StringArray,
+        StringDictionaryBuilder, TimestampMillisecondArray, UInt8Array, UInt64Array,
    };
    use datatypes::arrow::datatypes::{DataType, Field, Schema, UInt32Type};
    use datatypes::arrow::util::pretty::pretty_format_batches;
@@ -1625,91 +1625,178 @@ mod tests {

    #[tokio::test]
    async fn test_read_with_override_sequence() {
+        test_read_with_override_sequence_with_format(false).await;
+        test_read_with_override_sequence_with_format(true).await;
+    }
+
+    async fn test_read_with_override_sequence_with_format(flat_format: bool) {
        let mut env = TestEnv::new().await;
        let object_store = env.init_object_store_manager();
-        let handle = sst_file_handle(0, 1000);
-        let file_path = FixedPathProvider {
-            region_file_id: handle.file_id(),
-        };
        let metadata = Arc::new(sst_region_metadata());

-        // Create batches with sequence 0 to trigger override functionality.
-        let source = new_flat_source_from_record_batches(vec![
-            new_record_batch_with_custom_sequence(&["a", "d"], 0, 60, 0),
-            new_record_batch_with_custom_sequence(&["b", "f"], 0, 40, 0),
-        ]);
+        async fn read_sequences(builder: ParquetReaderBuilder) -> Vec<u64> {
+            let mut reader = builder.build().await.unwrap().unwrap();
+            let mut sequences = Vec::new();
+            while let Some(batch) = reader.next_record_batch().await.unwrap() {
+                let sequence = batch
+                    .column(batch.num_columns() - 2)
+                    .as_primitive::<datatypes::arrow::datatypes::UInt64Type>();
+                sequences.extend((0..sequence.len()).map(|idx| sequence.value(idx)));
+            }
+            sequences
+        }

-        let write_opts = WriteOptions {
-            row_group_size: 50,
-            ..Default::default()
-        };
+        async fn write_sst(
+            object_store: ObjectStore,
+            metadata: Arc<RegionMetadata>,
+            handle: FileHandle,
+            flat_format: bool,
+            sequence: u64,
+        ) {
+            let file_path = FixedPathProvider {
+                region_file_id: handle.file_id(),
+            };
+            let source = new_flat_source_from_record_batches(vec![
+                new_record_batch_with_custom_sequence(&["a", "d"], 0, 60, sequence),
+                new_record_batch_with_custom_sequence(&["b", "f"], 0, 40, sequence),
+            ]);
+            let write_opts = WriteOptions {
+                row_group_size: 50,
+                ..Default::default()
+            };
+            let mut metrics = Metrics::new(WriteType::Flush);
+            let mut writer = ParquetWriter::new_with_object_store(
+                object_store,
+                metadata,
+                IndexConfig::default(),
+                NoopIndexBuilder,
+                file_path,
+                &mut metrics,
+            )
+            .await;
+            if flat_format {
+                writer
+                    .write_all_flat(source, None, &write_opts)
+                    .await
+                    .unwrap();
+            } else {
+                writer
+                    .write_all_flat_as_primary_key(source, None, &write_opts)
+                    .await
+                    .unwrap();
+            }
+        }

-        let mut metrics = Metrics::new(WriteType::Flush);
-        let mut writer = ParquetWriter::new_with_object_store(
+        let custom_sequence = 12345u64;
+        let local_zero_handle = sst_file_handle(0, 1000);
+        write_sst(
            object_store.clone(),
            metadata.clone(),
-            IndexConfig::default(),
-            NoopIndexBuilder,
-            file_path,
-            &mut metrics,
+            local_zero_handle.clone(),
+            flat_format,
+            0,
        )
        .await;

-        writer
-            .write_all_flat_as_primary_key(source, None, &write_opts)
-            .await
-            .unwrap()
-            .remove(0);
+        // Local all-zero SSTs retain the compatibility override.
+        let local_zero_none = read_sequences(
+            ParquetReaderBuilder::new(
+                FILE_DIR.to_string(),
+                PathType::Bare,
+                local_zero_handle.clone(),
+                object_store.clone(),
+            )
+            .expected_metadata(Some(metadata.clone())),
+        )
+        .await;
+        assert!(local_zero_none.iter().all(|sequence| *sequence == 0));

-        // Read without override sequence (should read sequence 0)
-        let builder = ParquetReaderBuilder::new(
-            FILE_DIR.to_string(),
-            PathType::Bare,
-            handle.clone(),
-            object_store.clone(),
-        );
-        let mut reader = builder.build().await.unwrap().unwrap();
-        let mut normal_batches = Vec::new();
-        while let Some(batch) = reader.next_record_batch().await.unwrap() {
-            normal_batches.push(batch);
-        }
-
-        // Read with override sequence using FileMeta.sequence
-        let custom_sequence = 12345u64;
-        let file_meta = handle.meta_ref();
-        let mut override_file_meta = file_meta.clone();
-        override_file_meta.sequence = Some(std::num::NonZero::new(custom_sequence).unwrap());
-        let override_handle = FileHandle::new(
-            override_file_meta,
+        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),
        );
-
-        let builder = ParquetReaderBuilder::new(
-            FILE_DIR.to_string(),
-            PathType::Bare,
-            override_handle,
-            object_store.clone(),
+        let local_zero_override = read_sequences(
+            ParquetReaderBuilder::new(
+                FILE_DIR.to_string(),
+                PathType::Bare,
+                local_zero_override_handle,
+                object_store.clone(),
+            )
+            .expected_metadata(Some(metadata.clone())),
+        )
+        .await;
+        assert!(
+            local_zero_override
+                .iter()
+                .all(|sequence| *sequence == custom_sequence)
        );
-        let mut reader = builder.build().await.unwrap().unwrap();
-        let mut override_batches = Vec::new();
-        while let Some(batch) = reader.next_record_batch().await.unwrap() {
-            override_batches.push(batch);
-        }

-        // Compare the results
-        assert_eq!(normal_batches.len(), override_batches.len());
-        for (normal, override_batch) in normal_batches.into_iter().zip(override_batches.iter()) {
-            let expected_batch = {
-                let mut columns = normal.columns().to_vec();
-                let num_cols = columns.len();
-                columns[num_cols - 2] =
-                    Arc::new(UInt64Array::from_value(custom_sequence, normal.num_rows()));
-                RecordBatch::try_new(normal.schema(), columns).unwrap()
-            };
+        let local_nonzero_handle = sst_file_handle(0, 1000);
+        write_sst(
+            object_store.clone(),
+            metadata.clone(),
+            local_nonzero_handle.clone(),
+            flat_format,
+            7,
+        )
+        .await;

-            // Override batch should match expected batch
-            assert_eq!(*override_batch, expected_batch);
-        }
+        // 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),
+        );
+        let local_nonzero_override = read_sequences(
+            ParquetReaderBuilder::new(
+                FILE_DIR.to_string(),
+                PathType::Bare,
+                local_nonzero_override_handle,
+                object_store.clone(),
+            )
+            .expected_metadata(Some(metadata.clone())),
+        )
+        .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(),
+                PathType::Bare,
+                local_nonzero_handle.clone(),
+                object_store.clone(),
+            )
+            .expected_metadata(Some(metadata.clone())),
+        )
+        .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(
+            ParquetReaderBuilder::new(
+                FILE_DIR.to_string(),
+                PathType::Bare,
+                foreign_handle,
+                object_store,
+            )
+            .expected_metadata(Some(target_metadata)),
+        )
+        .await;
+        assert!(foreign.iter().all(|sequence| *sequence == custom_sequence));
    }

    #[tokio::test]
@@ -22,7 +22,7 @@ use std::time::{Duration, Instant};

 use api::v1::SemanticType;
 use common_recordbatch::filter::SimpleFilterEvaluator;
-use common_telemetry::{error, tracing, warn};
+use common_telemetry::{debug, error, tracing, warn};
 use datafusion::physical_plan::PhysicalExpr;
 use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion};
 use datafusion_expr::utils::expr_to_columns;
@@ -471,7 +471,24 @@ impl ParquetReaderBuilder {
            &file_path,
            skip_auto_convert,
        )?;
-        if need_override_sequence(&parquet_meta) {
+        // `region_meta` comes from the Parquet/source file and must not be used as the
+        // target identity. When the caller has no current metadata, the handle is the
+        // only local identity available and therefore denotes a local read.
+        let expected_region_id = self
+            .expected_metadata
+            .as_ref()
+            .map(|metadata| metadata.region_id)
+            .unwrap_or(self.file_handle.region_id());
+        let is_foreign = self.file_handle.region_id() != expected_region_id;
+        if is_foreign {
+            debug!(
+                "Reading foreign SST, file_id: {}, source_region_id: {}, expected_region_id: {}",
+                self.file_handle.file_id().file_id(),
+                self.file_handle.region_id(),
+                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()));
        }
@@ -79,6 +79,7 @@ impl<S> RegionWorkerLoop<S> {
        IndexBuildTask {
            region_id: region.region_id,
            file: file.clone(),
+            target_region_metadata: version.metadata.clone(),
            source: IndexBuildSource::new(file_meta, version.metadata.schema_version),
            reason: build_type,
            access_layer: access_layer.clone(),