From bd7d2c1dfa166f579bc33c496d50d41eb29b775f Mon Sep 17 00:00:00 2001 From: discord9 Date: Fri, 28 Aug 2026 03:45:41 +0000 Subject: [PATCH] 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> --- .../src/engine/apply_staging_manifest_test.rs | 215 ++++++++++++++++- src/mito2/src/sst/index.rs | 86 ++++++- src/mito2/src/sst/parquet.rs | 227 ++++++++++++------ src/mito2/src/sst/parquet/reader.rs | 21 +- src/mito2/src/worker/handle_rebuild_index.rs | 1 + 5 files changed, 466 insertions(+), 84 deletions(-) diff --git a/src/mito2/src/engine/apply_staging_manifest_test.rs b/src/mito2/src/engine/apply_staging_manifest_test.rs index e10ed0cbf0..49cc97c7d5 100644 --- a/src/mito2/src/engine/apply_staging_manifest_test.rs +++ b/src/mito2/src/engine/apply_staging_manifest_test.rs @@ -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::>(); + let output_ids = output_files + .iter() + .map(|file| file.meta_ref().file_id) + .collect::>(); + 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 { + 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::(), + 1 + ); + let batch = batches.iter().next().unwrap(); + let values = batch + .column_by_name("field_0") + .unwrap() + .as_primitive::(); + 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())) diff --git a/src/mito2/src/sst/index.rs b/src/mito2/src/sst/index.rs index 581d1edebd..7954af2b92 100644 --- a/src/mito2/src/sst/index.rs +++ b/src/mito2/src/sst/index.rs @@ -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::>(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(), diff --git a/src/mito2/src/sst/parquet.rs b/src/mito2/src/sst/parquet.rs index e67dcf2bee..8a05c099e9 100644 --- a/src/mito2/src/sst/parquet.rs +++ b/src/mito2/src/sst/parquet.rs @@ -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 { + 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::(); + 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, + 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] diff --git a/src/mito2/src/sst/parquet/reader.rs b/src/mito2/src/sst/parquet/reader.rs index 4f890cb83e..3adb335564 100644 --- a/src/mito2/src/sst/parquet/reader.rs +++ b/src/mito2/src/sst/parquet/reader.rs @@ -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())); } diff --git a/src/mito2/src/worker/handle_rebuild_index.rs b/src/mito2/src/worker/handle_rebuild_index.rs index ad0a09e1d3..3eb45b1ab2 100644 --- a/src/mito2/src/worker/handle_rebuild_index.rs +++ b/src/mito2/src/worker/handle_rebuild_index.rs @@ -79,6 +79,7 @@ impl RegionWorkerLoop { 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(),