diff --git a/bindings/c/src/tests.rs b/bindings/c/src/tests.rs index 37fa0c91..db8d52d5 100644 --- a/bindings/c/src/tests.rs +++ b/bindings/c/src/tests.rs @@ -2461,6 +2461,7 @@ fn build_pk_vector_table(path: &str, vectors: &[[f32; PK_DIM]]) -> Table { file_size: i64::try_from(index_file_size).unwrap(), row_count, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: 0, row_range_end: row_count - 1, diff --git a/crates/integrations/datafusion/src/system_tables/table_indexes.rs b/crates/integrations/datafusion/src/system_tables/table_indexes.rs index cff4796a..2533b828 100644 --- a/crates/integrations/datafusion/src/system_tables/table_indexes.rs +++ b/crates/integrations/datafusion/src/system_tables/table_indexes.rs @@ -289,6 +289,7 @@ mod tests { file_size: 1, row_count: 1, deletion_vectors_ranges: None, + external_path: None, global_index_meta: None, }, version: 1, diff --git a/crates/integrations/datafusion/tests/read_tables.rs b/crates/integrations/datafusion/tests/read_tables.rs index e4fc4721..8d770d35 100644 --- a/crates/integrations/datafusion/tests/read_tables.rs +++ b/crates/integrations/datafusion/tests/read_tables.rs @@ -1568,6 +1568,7 @@ mod fulltext_tests { file_size: i64::try_from(index_bytes.len()).unwrap(), row_count: 5, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: 0, row_range_end: 4, diff --git a/crates/paimon/src/catalog/filesystem.rs b/crates/paimon/src/catalog/filesystem.rs index 45628ad6..681bc3a6 100644 --- a/crates/paimon/src/catalog/filesystem.rs +++ b/crates/paimon/src/catalog/filesystem.rs @@ -26,7 +26,10 @@ use crate::common::{CatalogOptions, Options}; use crate::error::{ConfigInvalidSnafu, Error, Result}; use crate::io::cache::{create_local_cache, LocalCache}; use crate::io::FileIO; -use crate::spec::{CoreOptions, Schema, TableSchema, TableType, TABLE_TYPE_OPTION}; +use crate::spec::{ + CoreOptions, Schema, TableSchema, TableType, INDEX_FILE_IN_DATA_FILE_DIR_OPTION, + TABLE_TYPE_OPTION, +}; use crate::table::{ObjectTable, SchemaManager, Table}; use async_trait::async_trait; use bytes::Bytes; @@ -539,7 +542,7 @@ impl Catalog for FileSystemCatalog { full_name: identifier.full_name(), })?; - reject_table_type_changes(current.options(), &changes)?; + reject_immutable_option_changes(current.options(), &changes)?; let new_schema = current .apply_changes(changes) @@ -548,10 +551,19 @@ impl Catalog for FileSystemCatalog { } } -/// The declared type picks the reader, so it is fixed at creation: flipping -/// it strands a populated table behind a reader that cannot see its data. -/// Only case-insensitive no-ops pass, matching Java `SchemaManager`. -fn reject_table_type_changes( +/// Options whose value is baked into the on-disk layout, so changing one on a +/// populated table strands the files already written under the old value. +/// +/// Java rejects any alteration of an option annotated `@Immutable` +/// (`SchemaManager.checkAlterTableOption` against `CoreOptions.IMMUTABLE_OPTIONS`); +/// this mirrors the subset this crate acts on: +/// +/// * `type` picks the reader, so flipping it strands a populated table behind a +/// reader that cannot see its data. Only case-insensitive no-ops pass. +/// * `index-file-in-data-file-dir` picks the directory every bucket-local index +/// file is written to and read from, so flipping it hides every index file the +/// table already has. +fn reject_immutable_option_changes( current_options: &HashMap, changes: &[crate::spec::SchemaChange], ) -> Result<()> { @@ -575,6 +587,18 @@ fn reject_table_type_changes( message: format!("removing '{TABLE_TYPE_OPTION}' is not supported"), }); } + crate::spec::SchemaChange::SetOption { key, .. } + | crate::spec::SchemaChange::RemoveOption { key } + if key == INDEX_FILE_IN_DATA_FILE_DIR_OPTION => + { + return Err(Error::Unsupported { + message: format!( + "changing '{INDEX_FILE_IN_DATA_FILE_DIR_OPTION}' is not supported: \ + it selects the directory index files are written to, so the files \ + already written would no longer be found" + ), + }); + } _ => {} } } @@ -828,6 +852,51 @@ mod tests { catalog.get_table(&identifier).await.unwrap(); } + #[tokio::test] + async fn test_alter_table_cannot_change_where_index_files_live() { + use crate::spec::SchemaChange; + + // The option selects the directory every bucket-local index file is written + // to and read from. Flipping it on a populated table would hide every index + // file already written, so it is fixed at creation, as in Java where it is + // annotated `@Immutable`. + let (_temp_dir, catalog) = create_test_catalog(); + catalog + .create_database("db1", false, HashMap::new()) + .await + .unwrap(); + let schema = Schema::builder() + .column( + "id", + crate::spec::DataType::Int(crate::spec::IntType::new()), + ) + .build() + .unwrap(); + let identifier = Identifier::new("db1", "t"); + catalog + .create_table(&identifier, schema, false) + .await + .unwrap(); + + for change in [ + SchemaChange::set_option( + "index-file-in-data-file-dir".to_string(), + "true".to_string(), + ), + SchemaChange::set_option( + "index-file-in-data-file-dir".to_string(), + "false".to_string(), + ), + SchemaChange::remove_option("index-file-in-data-file-dir".to_string()), + ] { + let err = catalog + .alter_table(&identifier, vec![change], false) + .await + .unwrap_err(); + assert!(matches!(err, Error::Unsupported { .. }), "{err:?}"); + } + } + #[tokio::test] async fn test_create_table_rejects_an_unknown_type() { let (_temp_dir, catalog) = create_test_catalog(); diff --git a/crates/paimon/src/spec/avro/decode_helpers.rs b/crates/paimon/src/spec/avro/decode_helpers.rs index b36aec15..38165329 100644 --- a/crates/paimon/src/spec/avro/decode_helpers.rs +++ b/crates/paimon/src/spec/avro/decode_helpers.rs @@ -58,13 +58,23 @@ pub(crate) fn read_bytes_field(cursor: &mut AvroCursor, nullable: bool) -> crate } pub(crate) fn read_string_field(cursor: &mut AvroCursor, nullable: bool) -> crate::Result { + Ok(read_nullable_string_field(cursor, nullable)?.unwrap_or_default()) +} + +/// Reads a nullable string field, preserving the null/present distinction. +/// Returns `None` for the null branch of a `["null", "string"]` union (a +/// non-nullable field is always `Some`). +pub(crate) fn read_nullable_string_field( + cursor: &mut AvroCursor, + nullable: bool, +) -> crate::Result> { if nullable { let idx = cursor.read_union_index()?; if idx == 0 { - return Ok(String::new()); + return Ok(None); } } - Ok(cursor.read_string()?.to_string()) + Ok(Some(cursor.read_string()?.to_string())) } const EMPTY_PARTITION: [u8; 4] = [0, 0, 0, 0]; diff --git a/crates/paimon/src/spec/avro/index_manifest_entry_decode.rs b/crates/paimon/src/spec/avro/index_manifest_entry_decode.rs index 38bf0c71..50cfece1 100644 --- a/crates/paimon/src/spec/avro/index_manifest_entry_decode.rs +++ b/crates/paimon/src/spec/avro/index_manifest_entry_decode.rs @@ -19,7 +19,7 @@ use super::cursor::AvroCursor; use super::decode::{neg_count_to_usize, AvroRecordDecode}; use super::decode_helpers::{ extract_record_schema, normalize_partition, read_bytes_field, read_int_field, read_long_field, - read_string_field, + read_nullable_string_field, read_string_field, }; use super::schema::{skip_nullable_field, WriterSchema}; use crate::spec::index_manifest::IndexManifestEntry; @@ -38,6 +38,7 @@ impl AvroRecordDecode for IndexManifestEntry { let mut file_size: Option = None; let mut row_count: Option = None; let mut deletion_vectors_ranges: Option> = None; + let mut external_path: Option = None; let mut global_index_meta: Option = None; for field in &writer_schema.fields { @@ -65,6 +66,9 @@ impl AvroRecordDecode for IndexManifestEntry { "_DELETIONS_VECTORS_RANGES" | "_DELETION_VECTORS_RANGES" => { deletion_vectors_ranges = decode_nullable_dv_ranges(cursor, field.nullable)?; } + "_EXTERNAL_PATH" => { + external_path = read_nullable_string_field(cursor, field.nullable)?; + } "_GLOBAL_INDEX" => { global_index_meta = decode_nullable_global_index(cursor, field.nullable, &field.schema)?; @@ -84,6 +88,7 @@ impl AvroRecordDecode for IndexManifestEntry { file_size: file_size.unwrap_or(0), row_count: row_count.unwrap_or(0), deletion_vectors_ranges, + external_path, global_index_meta, }, }) diff --git a/crates/paimon/src/spec/core_options.rs b/crates/paimon/src/spec/core_options.rs index b17d3168..7afff6f6 100644 --- a/crates/paimon/src/spec/core_options.rs +++ b/crates/paimon/src/spec/core_options.rs @@ -32,6 +32,7 @@ const GLOBAL_INDEX_ROW_COUNT_PER_SHARD_OPTION: &str = "global-index.row-count-pe const GLOBAL_INDEX_THREAD_NUM_OPTION: &str = "global-index.thread-num"; const GLOBAL_INDEX_VINDEX_READ_THREAD_NUM_OPTION: &str = "global-index.vindex.read-thread-num"; const GLOBAL_INDEX_COLUMN_UPDATE_ACTION_OPTION: &str = "global-index.column-update-action"; +pub(crate) const INDEX_FILE_IN_DATA_FILE_DIR_OPTION: &str = "index-file-in-data-file-dir"; const SORTED_INDEX_RECORDS_PER_RANGE_OPTION: &str = "sorted-index.records-per-range"; const BTREE_INDEX_RECORDS_PER_RANGE_OPTION: &str = "btree-index.records-per-range"; const BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE_OPTION: &str = "btree-index.fallback-scan-max-size"; @@ -686,6 +687,16 @@ impl<'a> CoreOptions<'a> { .unwrap_or(true) } + /// Whether index files are stored in the bucket data-file directory rather + /// than the table `index/` directory (option `index-file-in-data-file-dir`, + /// default false). + pub fn index_file_in_data_file_dir(&self) -> bool { + self.options + .get(INDEX_FILE_IN_DATA_FILE_DIR_OPTION) + .map(|value| value.eq_ignore_ascii_case("true")) + .unwrap_or(false) + } + pub fn global_index_search_mode(&self) -> crate::Result { self.index_search_mode(GLOBAL_INDEX_SEARCH_MODE_OPTION) } diff --git a/crates/paimon/src/spec/index_file_meta.rs b/crates/paimon/src/spec/index_file_meta.rs index 3b1b2f56..7af98412 100644 --- a/crates/paimon/src/spec/index_file_meta.rs +++ b/crates/paimon/src/spec/index_file_meta.rs @@ -77,6 +77,15 @@ pub struct IndexFileMeta { )] pub deletion_vectors_ranges: Option>, + /// Absolute path of an externally-stored index file. `None` when the file + /// lives under the table's index directory (or bucket data-file directory). + #[serde( + default, + rename = "_EXTERNAL_PATH", + skip_serializing_if = "Option::is_none" + )] + pub external_path: Option, + #[serde( default, rename = "_GLOBAL_INDEX", diff --git a/crates/paimon/src/spec/index_manifest.rs b/crates/paimon/src/spec/index_manifest.rs index 24cac0f2..3cb3ead4 100644 --- a/crates/paimon/src/spec/index_manifest.rs +++ b/crates/paimon/src/spec/index_manifest.rs @@ -58,6 +58,7 @@ pub const INDEX_MANIFEST_ENTRY_SCHEMA: &str = r#"{ }] }] }, + {"name": "_EXTERNAL_PATH", "type": ["null", "string"], "default": null}, { "default": null, "name": "_GLOBAL_INDEX", @@ -202,6 +203,7 @@ mod tests { cardinality: Some(3), } )])), + external_path: None, global_index_meta: None, } }] @@ -228,6 +230,7 @@ mod tests { cardinality: Some(7), }, )])), + external_path: None, global_index_meta: None, }, }; @@ -288,6 +291,7 @@ mod tests { file_size: 42, row_count: 7, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: 10, row_range_end: 20, @@ -399,9 +403,122 @@ mod tests { } #[test] - fn legacy_five_field_global_index_decodes_without_source_meta() { - // 5-field _GLOBAL_INDEX schema (pre-#8549): no _SOURCE_META. Identical to - // INDEX_MANIFEST_ENTRY_SCHEMA with the trailing _SOURCE_META line removed. + fn decodes_index_external_path_at_its_schema_position() { + // `_EXTERNAL_PATH` (nullable string) is field 5 of Java `IndexFileMeta.SCHEMA`, + // between `_DELETIONS_VECTORS_RANGES` and `_GLOBAL_INDEX`. A manifest written + // for an externally-stored index file records the absolute path there; the + // decoder must read it from that position rather than skip it. + const SCHEMA_WITH_EXTERNAL_PATH: &str = r#"{ + "type": "record", + "name": "org.apache.paimon.avro.generated.record", + "fields": [ + {"name": "_VERSION", "type": "int"}, + {"name": "_KIND", "type": "int"}, + {"name": "_PARTITION", "type": "bytes"}, + {"name": "_BUCKET", "type": "int"}, + {"name": "_INDEX_TYPE", "type": "string"}, + {"name": "_FILE_NAME", "type": "string"}, + {"name": "_FILE_SIZE", "type": "long"}, + {"name": "_ROW_COUNT", "type": "long"}, + { + "default": null, + "name": "_DELETIONS_VECTORS_RANGES", + "type": ["null", { + "type": "array", + "items": ["null", { + "type": "record", + "name": "org.apache.paimon.avro.generated.record__DELETIONS_VECTORS_RANGES", + "fields": [ + {"name": "f0", "type": "string"}, + {"name": "f1", "type": "int"}, + {"name": "f2", "type": "int"}, + {"name": "_CARDINALITY", "type": ["null", "long"], "default": null} + ] + }] + }] + }, + {"name": "_EXTERNAL_PATH", "type": ["null", "string"], "default": null}, + { + "default": null, + "name": "_GLOBAL_INDEX", + "type": ["null", { + "type": "record", + "name": "org.apache.paimon.avro.generated.record__GLOBAL_INDEX", + "fields": [ + {"name": "_ROW_RANGE_START", "type": "long"}, + {"name": "_ROW_RANGE_END", "type": "long"}, + {"name": "_INDEX_FIELD_ID", "type": "int"}, + {"name": "_EXTRA_FIELD_IDS", "type": ["null", {"type": "array", "items": "int"}], "default": null}, + {"name": "_INDEX_META", "type": ["null", "bytes"], "default": null}, + {"name": "_SOURCE_META", "type": ["null", "bytes"], "default": null} + ] + }] + } + ] +}"#; + + let external = "s3://bucket/warehouse/db/tbl/index/idx-external-0"; + let entry: IndexManifestEntry = serde_json::from_value(serde_json::json!({ + "_VERSION": 1, + "_KIND": 0, + "_PARTITION": [0, 0, 0, 0], + "_BUCKET": 0, + "_INDEX_TYPE": "TEST", + "_FILE_NAME": "idx-external-0", + "_FILE_SIZE": 42, + "_ROW_COUNT": 7, + "_EXTERNAL_PATH": external + })) + .unwrap(); + + let bytes = crate::spec::to_avro_bytes_with_compression( + SCHEMA_WITH_EXTERNAL_PATH, + std::slice::from_ref(&entry), + crate::spec::DEFAULT_AVRO_COMPRESSION, + ) + .unwrap(); + + let decoded = IndexManifest::read_from_bytes(&bytes).unwrap(); + assert_eq!( + decoded[0].index_file.external_path.as_deref(), + Some(external) + ); + } + + #[test] + fn decodes_null_index_external_path_as_none() { + // A manifest whose `_EXTERNAL_PATH` is present but null decodes to `None`. + // The field being absent from the writer schema entirely is covered by + // `legacy_schema_without_external_path_field_decodes_as_none`. + let entry: IndexManifestEntry = serde_json::from_value(serde_json::json!({ + "_VERSION": 1, + "_KIND": 0, + "_PARTITION": [0, 0, 0, 0], + "_BUCKET": 0, + "_INDEX_TYPE": "TEST", + "_FILE_NAME": "idx-local-0", + "_FILE_SIZE": 42, + "_ROW_COUNT": 7 + })) + .unwrap(); + + let bytes = crate::spec::to_avro_bytes_with_compression( + INDEX_MANIFEST_ENTRY_SCHEMA, + std::slice::from_ref(&entry), + crate::spec::DEFAULT_AVRO_COMPRESSION, + ) + .unwrap(); + + let decoded = IndexManifest::read_from_bytes(&bytes).unwrap(); + assert_eq!(decoded[0].index_file.external_path, None); + } + + #[test] + fn legacy_schema_without_external_path_field_decodes_as_none() { + // A writer schema from before either field existed: no `_SOURCE_META` inside + // `_GLOBAL_INDEX` (pre-#8549), and no `_EXTERNAL_PATH` at all. The decoder + // walks the writer's own field list, so both must simply be absent from the + // decoded entry without misaligning the stream. const LEGACY_SCHEMA: &str = r#"{ "type": "record", "name": "org.apache.paimon.avro.generated.record", @@ -457,7 +574,7 @@ mod tests { crate::spec::DEFAULT_AVRO_COMPRESSION, ) .unwrap(); - // Decoding with the current 6-field reader must not misalign the stream. + // Decoding with the current reader must not misalign the stream. let decoded = IndexManifest::read_from_bytes(&bytes).unwrap(); assert_eq!(decoded[0], entry); assert_eq!( @@ -469,5 +586,6 @@ mod tests { .source_meta, None ); + assert_eq!(decoded[0].index_file.external_path, None); } } diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs index a2613fec..a165ffe0 100644 --- a/crates/paimon/src/spec/mod.rs +++ b/crates/paimon/src/spec/mod.rs @@ -98,7 +98,7 @@ pub use types::*; mod partition; pub use partition::Partition; mod partition_utils; -pub(crate) use partition_utils::PartitionComputer; +pub(crate) use partition_utils::{bucket_path, bucket_path_under, PartitionComputer}; mod predicate; pub(crate) use predicate::datum_cmp; pub(crate) use predicate::eval_row; diff --git a/crates/paimon/src/spec/partition_utils.rs b/crates/paimon/src/spec/partition_utils.rs index 3db1884c..e4245c31 100644 --- a/crates/paimon/src/spec/partition_utils.rs +++ b/crates/paimon/src/spec/partition_utils.rs @@ -31,6 +31,40 @@ use chrono::{Local, NaiveDate, NaiveDateTime, TimeZone, Timelike}; const MILLIS_PER_DAY: i64 = 86_400_000; +/// The directory holding one bucket's data files, `/[/]bucket-N`. +/// +/// Mirrors Java `FileStorePathFactory.bucketPath`. Every consumer that needs a +/// bucket directory — data files, and index files kept beside them — must derive +/// it here: a writer and a reader that disagree by one segment silently lose the +/// file, and there is no compile error to catch that. +/// +/// `partition_computer` is `None` for an unpartitioned table, whose buckets sit +/// directly under the table path. +pub(crate) fn bucket_path( + table_path: &str, + partition_computer: Option<&PartitionComputer>, + partition: &BinaryRow, + bucket: i32, +) -> crate::Result { + let partition_path = match partition_computer { + Some(computer) => computer.generate_partition_path(partition)?, + None => String::new(), + }; + Ok(bucket_path_under(table_path, &partition_path, bucket)) +} + +/// [`bucket_path`] for a partition directory that is already computed. +/// +/// `partition_path` is empty for an unpartitioned table and otherwise ends with `/`, +/// matching [`PartitionComputer::generate_partition_path`]. +pub(crate) fn bucket_path_under(table_path: &str, partition_path: &str, bucket: i32) -> String { + format!( + "{}/{partition_path}{}", + table_path.trim_end_matches('/'), + crate::spec::bucket_dir_name(bucket) + ) +} + /// Computes partition string values and directory paths from a partition `BinaryRow`. /// /// Mirrors Java `InternalRowPartitionComputer` — holds resolved partition field metadata @@ -38,7 +72,7 @@ const MILLIS_PER_DAY: i64 = 86_400_000; /// (escaped directory path). /// /// Reference: `org.apache.paimon.utils.InternalRowPartitionComputer` in Java Paimon. -#[derive(Debug)] +#[derive(Debug, Clone)] pub(crate) struct PartitionComputer { partition_keys: Vec, partition_fields: Vec, diff --git a/crates/paimon/src/table/bucket_assigner.rs b/crates/paimon/src/table/bucket_assigner.rs index a6291e4d..64abf810 100644 --- a/crates/paimon/src/table/bucket_assigner.rs +++ b/crates/paimon/src/table/bucket_assigner.rs @@ -59,7 +59,6 @@ pub(crate) trait BucketAssigner: Send { fn prepare_commit_index( &mut self, file_io: &FileIO, - index_dir: &str, ) -> impl std::future::Future>>> + Send; } @@ -67,7 +66,7 @@ pub(crate) trait BucketAssigner: Send { pub(crate) enum BucketAssignerEnum { Constant(ConstantBucketAssigner), Fixed(FixedBucketAssigner), - Dynamic(DynamicBucketAssigner), + Dynamic(Box), CrossPartition(Box), } @@ -88,13 +87,12 @@ impl BucketAssignerEnum { pub async fn prepare_commit_index( &mut self, file_io: &FileIO, - index_dir: &str, ) -> Result>> { match self { - Self::Constant(a) => a.prepare_commit_index(file_io, index_dir).await, - Self::Fixed(a) => a.prepare_commit_index(file_io, index_dir).await, - Self::Dynamic(a) => a.prepare_commit_index(file_io, index_dir).await, - Self::CrossPartition(a) => a.prepare_commit_index(file_io, index_dir).await, + Self::Constant(a) => a.prepare_commit_index(file_io).await, + Self::Fixed(a) => a.prepare_commit_index(file_io).await, + Self::Dynamic(a) => a.prepare_commit_index(file_io).await, + Self::CrossPartition(a) => a.prepare_commit_index(file_io).await, } } diff --git a/crates/paimon/src/table/bucket_assigner_constant.rs b/crates/paimon/src/table/bucket_assigner_constant.rs index 5d0f2a4e..843d2464 100644 --- a/crates/paimon/src/table/bucket_assigner_constant.rs +++ b/crates/paimon/src/table/bucket_assigner_constant.rs @@ -71,7 +71,6 @@ impl BucketAssigner for ConstantBucketAssigner { async fn prepare_commit_index( &mut self, _file_io: &FileIO, - _index_dir: &str, ) -> Result>> { Ok(HashMap::new()) } diff --git a/crates/paimon/src/table/bucket_assigner_cross.rs b/crates/paimon/src/table/bucket_assigner_cross.rs index 5a1d3398..4ae18157 100644 --- a/crates/paimon/src/table/bucket_assigner_cross.rs +++ b/crates/paimon/src/table/bucket_assigner_cross.rs @@ -354,7 +354,6 @@ impl BucketAssigner for CrossPartitionAssigner { async fn prepare_commit_index( &mut self, _file_io: &FileIO, - _index_dir: &str, ) -> Result>> { Ok(HashMap::new()) } diff --git a/crates/paimon/src/table/bucket_assigner_dynamic.rs b/crates/paimon/src/table/bucket_assigner_dynamic.rs index 030385d9..57fbc48e 100644 --- a/crates/paimon/src/table/bucket_assigner_dynamic.rs +++ b/crates/paimon/src/table/bucket_assigner_dynamic.rs @@ -22,10 +22,11 @@ use crate::io::FileIO; use crate::spec::{ - batch_hash_codes, batch_to_serialized_bytes, DataField, IndexFileMeta, IndexManifest, - IndexManifestEntry, EMPTY_SERIALIZED_ROW, + batch_hash_codes, batch_to_serialized_bytes, bucket_path_under, BinaryRow, DataField, + IndexFileMeta, IndexManifest, IndexManifestEntry, PartitionComputer, EMPTY_SERIALIZED_ROW, }; use crate::table::bucket_assigner::{BatchAssignOutput, BucketAssigner, PartitionBucketKey}; +use crate::table::index_file_path::IndexFileLocation; use crate::table::SnapshotManager; use crate::Result; use arrow_array::RecordBatch; @@ -96,6 +97,7 @@ impl HashIndexFile { .try_into() .expect("hash index row count exceeds i32::MAX"), deletion_vectors_ranges: None, + external_path: None, global_index_meta: None, }) } @@ -155,6 +157,48 @@ impl DynamicBucketIndexMaintainer { // PartitionIndex // --------------------------------------------------------------------------- +/// Where one partition's hash index files live. +/// +/// A hash index is an index file, so it sits beside its bucket's data files when +/// the table keeps index files in the data-file directory, and under the table +/// `index/` directory otherwise. Reads and writes resolve through the same value +/// so a file written here is found again. +struct HashIndexLayout<'a> { + table_path: &'a str, + /// Partition directory, already terminated by `/`, or empty when unpartitioned. + partition_path: &'a str, + index_file_in_data_file_dir: bool, +} + +impl HashIndexLayout<'_> { + /// This layout as the shared resolver's bucket-local mode. The bucket + /// directory is passed in so the resolver can borrow it. + fn location<'b>(&'b self, bucket_path: &'b str) -> IndexFileLocation<'b> { + IndexFileLocation::BucketLocal { + table_path: self.table_path, + bucket_path, + index_file_in_data_file_dir: self.index_file_in_data_file_dir, + } + } + + fn bucket_path(&self, bucket: i32) -> String { + bucket_path_under(self.table_path, self.partition_path, bucket) + } + + /// The directory a new hash index file for `bucket` is written into. + fn directory(&self, bucket: i32) -> String { + let bucket_path = self.bucket_path(bucket); + self.location(&bucket_path).directory() + } + + /// The path of an existing hash index file recorded for `bucket`. + fn resolve(&self, bucket: i32, file_name: &str, external_path: Option<&str>) -> String { + let bucket_path = self.bucket_path(bucket); + self.location(&bucket_path) + .resolve(file_name, external_path) + } +} + /// Per-partition index that maps key hashes to bucket ids. /// /// Also maintains per-bucket index files via embedded `DynamicBucketIndexMaintainer`s, @@ -194,7 +238,7 @@ impl PartitionIndex { /// the hash→bucket mapping and bucket row counts. async fn load( file_io: &FileIO, - index_dir: &str, + layout: &HashIndexLayout<'_>, entries: &[IndexManifestEntry], target_bucket_row_number: i64, ) -> Result { @@ -207,7 +251,11 @@ impl PartitionIndex { continue; } let bucket = entry.bucket; - let path = format!("{index_dir}/{}", entry.index_file.file_name); + let path = layout.resolve( + bucket, + &entry.index_file.file_name, + entry.index_file.external_path.as_deref(), + ); let hashes = HashIndexFile::read(file_io, &path).await?; let count = hashes.len() as i64; for &h in &hashes { @@ -292,13 +340,14 @@ impl PartitionIndex { async fn prepare_commit( &mut self, file_io: &FileIO, - index_dir: &str, + layout: &HashIndexLayout<'_>, ) -> Result)>> { let mut result = Vec::new(); let buckets: Vec = self.bucket_maintainers.keys().copied().collect(); for bucket in buckets { if let Some(maintainer) = self.bucket_maintainers.get_mut(&bucket) { - let files = maintainer.prepare_commit(file_io, index_dir).await?; + let index_dir = layout.directory(bucket); + let files = maintainer.prepare_commit(file_io, &index_dir).await?; if !files.is_empty() { result.push((bucket, files)); } @@ -328,9 +377,16 @@ pub(crate) struct DynamicBucketAssigner { cached_index_entries: Option>, /// Overwrite mode: skip loading existing index entries. is_overwrite: bool, + /// Builds the partition directory of a bucket, so a hash index kept in the + /// data-file directory is written and read in the same place. Yields an empty + /// path for an unpartitioned table. + partition_computer: PartitionComputer, + /// Whether the table stores index files in the data-file (bucket) directory. + index_file_in_data_file_dir: bool, } impl DynamicBucketAssigner { + #[allow(clippy::too_many_arguments)] pub fn new( partition_field_indices: Vec, primary_key_indices: Vec, @@ -339,6 +395,8 @@ impl DynamicBucketAssigner { file_io: FileIO, table_location: String, is_overwrite: bool, + partition_computer: PartitionComputer, + index_file_in_data_file_dir: bool, ) -> Self { Self { partition_field_indices, @@ -350,6 +408,8 @@ impl DynamicBucketAssigner { table_location, cached_index_entries: None, is_overwrite, + partition_computer, + index_file_in_data_file_dir, } } @@ -386,6 +446,14 @@ impl DynamicBucketAssigner { Ok(()) } + /// The partition directory of a bucket, terminated by `/`, or empty when the + /// table is unpartitioned. + fn partition_path(&self, partition_bytes: &[u8]) -> Result { + let partition_row = BinaryRow::from_serialized_bytes(partition_bytes)?; + self.partition_computer + .generate_partition_path(&partition_row) + } + /// Load partition index from cached index manifest entries. async fn load_partition_index(&self, partition_bytes: &[u8]) -> Result { let entries = self.cached_index_entries.as_deref().unwrap_or(&[]); @@ -396,10 +464,15 @@ impl DynamicBucketAssigner { .collect(); if !partition_entries.is_empty() { - let index_dir = format!("{}/index", self.table_location); + let partition_path = self.partition_path(partition_bytes)?; + let layout = HashIndexLayout { + table_path: self.table_location.trim_end_matches('/'), + partition_path: &partition_path, + index_file_in_data_file_dir: self.index_file_in_data_file_dir, + }; return PartitionIndex::load( &self.file_io, - &index_dir, + &layout, &partition_entries, self.target_bucket_row_number, ) @@ -458,13 +531,23 @@ impl BucketAssigner for DynamicBucketAssigner { async fn prepare_commit_index( &mut self, file_io: &FileIO, - index_dir: &str, ) -> Result>> { let mut result = HashMap::new(); + let table_path = self.table_location.trim_end_matches('/').to_string(); + let index_file_in_data_file_dir = self.index_file_in_data_file_dir; let partition_keys: Vec> = self.partition_indexes.keys().cloned().collect(); - for partition_bytes in partition_keys { + let mut partition_paths = Vec::with_capacity(partition_keys.len()); + for partition_bytes in &partition_keys { + partition_paths.push(self.partition_path(partition_bytes)?); + } + for (partition_bytes, partition_path) in partition_keys.into_iter().zip(partition_paths) { + let layout = HashIndexLayout { + table_path: &table_path, + partition_path: &partition_path, + index_file_in_data_file_dir, + }; if let Some(partition_index) = self.partition_indexes.get_mut(&partition_bytes) { - let bucket_files = partition_index.prepare_commit(file_io, index_dir).await?; + let bucket_files = partition_index.prepare_commit(file_io, &layout).await?; for (bucket, idx_files) in bucket_files { result.insert((partition_bytes.clone(), bucket), idx_files); } @@ -557,6 +640,81 @@ mod tests { // -- HashIndexFile tests -- + /// Reads and writes resolve through the same layout, so a hash index written + /// under one configuration is found again; an explicit external path wins. + #[tokio::test] + async fn test_hash_index_layout_round_trips_read_and_write() { + for index_file_in_data_file_dir in [false, true] { + let tmp = tempfile::tempdir().unwrap(); + let table_path = format!("file://{}", tmp.path().display()); + let file_io = FileIO::from_url(&table_path).unwrap().build().unwrap(); + let layout = super::HashIndexLayout { + table_path: &table_path, + partition_path: "pt=1/", + index_file_in_data_file_dir, + }; + + // Write where this layout says, then read it back through the same layout. + let dir = layout.directory(3); + file_io.mkdirs(&dir).await.unwrap(); + let hashes = vec![7i32, 8, 9]; + let meta = HashIndexFile::write(&file_io, &dir, &hashes).await.unwrap(); + let entries = vec![IndexManifestEntry { + version: 1, + kind: crate::spec::FileKind::Add, + partition: EMPTY_SERIALIZED_ROW.to_vec(), + bucket: 3, + index_file: meta, + }]; + let loaded = PartitionIndex::load(&file_io, &layout, &entries, 100) + .await + .unwrap(); + for hash in &hashes { + assert_eq!(loaded.hash_to_bucket.get(hash), Some(&3)); + } + + let expected_dir = if index_file_in_data_file_dir { + format!("{table_path}/pt=1/bucket-3") + } else { + format!("{table_path}/index") + }; + assert_eq!(dir, expected_dir); + } + } + + /// An external path wins over both layouts. + #[tokio::test] + async fn test_hash_index_external_path_wins() { + let tmp = tempfile::tempdir().unwrap(); + let table_path = format!("file://{}", tmp.path().display()); + let file_io = FileIO::from_url(&table_path).unwrap().build().unwrap(); + let external_dir = format!("{table_path}/elsewhere"); + file_io.mkdirs(&external_dir).await.unwrap(); + let mut index_file = HashIndexFile::write(&file_io, &external_dir, &[42i32]) + .await + .unwrap(); + index_file.external_path = Some(format!("{external_dir}/{}", index_file.file_name)); + + for index_file_in_data_file_dir in [false, true] { + let layout = super::HashIndexLayout { + table_path: &table_path, + partition_path: "pt=1/", + index_file_in_data_file_dir, + }; + let entries = vec![IndexManifestEntry { + version: 1, + kind: crate::spec::FileKind::Add, + partition: EMPTY_SERIALIZED_ROW.to_vec(), + bucket: 5, + index_file: index_file.clone(), + }]; + let loaded = PartitionIndex::load(&file_io, &layout, &entries, 100) + .await + .unwrap(); + assert_eq!(loaded.hash_to_bucket.get(&42), Some(&5)); + } + } + #[tokio::test] async fn test_hash_index_roundtrip() { let tmp = tempfile::TempDir::new().unwrap(); diff --git a/crates/paimon/src/table/bucket_assigner_fixed.rs b/crates/paimon/src/table/bucket_assigner_fixed.rs index 428b07dd..684a98a8 100644 --- a/crates/paimon/src/table/bucket_assigner_fixed.rs +++ b/crates/paimon/src/table/bucket_assigner_fixed.rs @@ -86,7 +86,6 @@ impl BucketAssigner for FixedBucketAssigner { async fn prepare_commit_index( &mut self, _file_io: &FileIO, - _index_dir: &str, ) -> Result>> { Ok(HashMap::new()) } diff --git a/crates/paimon/src/table/data_evolution_writer.rs b/crates/paimon/src/table/data_evolution_writer.rs index 1736626c..d2855348 100644 --- a/crates/paimon/src/table/data_evolution_writer.rs +++ b/crates/paimon/src/table/data_evolution_writer.rs @@ -29,11 +29,12 @@ use crate::deletion_vector::{DeletionVector, DeletionVectorFactory}; use crate::io::FileIO; use crate::spec::{ - BinaryRow, CoreOptions, DataField, DataFileMeta, DataType, DeletionVectorMeta, FileKind, - IndexFileMeta, IndexManifest, PartitionComputer, + bucket_path, BinaryRow, CoreOptions, DataField, DataFileMeta, DataType, DeletionVectorMeta, + FileKind, IndexFileMeta, IndexManifest, PartitionComputer, EMPTY_BINARY_ROW, }; use crate::table::commit_message::CommitMessage; use crate::table::data_file_writer::DataFileWriter; +use crate::table::index_file_path::IndexFileLocation; use crate::table::source::data_evolution_anchor_file; use crate::table::stats_filter::group_by_overlapping_row_id; use crate::table::DataSplitBuilder; @@ -53,7 +54,6 @@ use uuid::Uuid; const DELETION_VECTORS_INDEX_TYPE: &str = "DELETION_VECTORS"; const DELETION_VECTORS_INDEX_VERSION_V1: u8 = 1; -const INDEX_DIR: &str = "index"; const MANIFEST_DIR: &str = "manifest"; /// Engine-agnostic writer for partial-column updates via `_ROW_ID`. @@ -388,6 +388,25 @@ impl DataEvolutionWriter { } } +/// One bucket's deletion-vector location, owning the strings the shared resolver +/// borrows. Built once per bucket so the read that merges existing vectors and the +/// write that follows it cannot disagree about where the file goes. +struct DeletionVectorLayout { + table_path: String, + bucket_path: String, + index_file_in_data_file_dir: bool, +} + +impl DeletionVectorLayout { + fn location(&self) -> IndexFileLocation<'_> { + IndexFileLocation::BucketLocal { + table_path: &self.table_path, + bucket_path: &self.bucket_path, + index_file_in_data_file_dir: self.index_file_in_data_file_dir, + } + } +} + /// Engine-agnostic DELETE writer for data evolution tables. /// /// DELETE is represented by a deletion-vector index file keyed by the normal @@ -589,7 +608,9 @@ impl DataEvolutionDeleteWriter { return Ok(None); } - let new_index_file = self.write_deletion_vector_index_file(bitmaps).await?; + let new_index_file = self + .write_deletion_vector_index_file(&partition, bucket, bitmaps) + .await?; let mut message = CommitMessage::new(partition, bucket, vec![]); message.check_from_snapshot = Some(delete_plan.check_from_snapshot); message.new_index_files = vec![new_index_file]; @@ -597,6 +618,45 @@ impl DataEvolutionDeleteWriter { Ok(Some(message)) } + /// Where this bucket's deletion vectors live. A deletion vector is an index + /// file, so it sits beside the bucket's data files when the table keeps index + /// files there, and under the table `index/` directory otherwise. Reads and + /// writes both go through this, so a file written here is found again. + fn deletion_vector_layout( + &self, + partition: &[u8], + bucket: i32, + ) -> Result { + let schema = self.table.schema(); + let partition_keys = schema.partition_keys(); + let core_options = CoreOptions::new(schema.options()); + let computer = if partition_keys.is_empty() { + None + } else { + Some(PartitionComputer::new( + partition_keys, + schema.fields(), + core_options.partition_default_name(), + core_options.legacy_partition_name(), + )?) + }; + let partition_row = if computer.is_some() { + BinaryRow::from_serialized_bytes(partition)? + } else { + EMPTY_BINARY_ROW + }; + Ok(DeletionVectorLayout { + table_path: self.table.location().trim_end_matches('/').to_string(), + bucket_path: bucket_path( + self.table.location(), + computer.as_ref(), + &partition_row, + bucket, + )?, + index_file_in_data_file_dir: core_options.index_file_in_data_file_dir(), + }) + } + async fn read_existing_bucket_deletion_vectors( &self, partition: &[u8], @@ -611,6 +671,7 @@ impl DataEvolutionDeleteWriter { let Some(index_manifest_name) = snapshot.index_manifest() else { return Ok((IndexMap::new(), Vec::new())); }; + let layout = self.deletion_vector_layout(partition, bucket)?; let manifest_path = format!( "{}/{MANIFEST_DIR}/{}", @@ -633,10 +694,9 @@ impl DataEvolutionDeleteWriter { let Some(ranges) = entry.index_file.deletion_vectors_ranges.as_ref() else { continue; }; - let index_path = format!( - "{}/{INDEX_DIR}/{}", - self.table.location().trim_end_matches('/'), - entry.index_file.file_name + let index_path = layout.location().resolve( + &entry.index_file.file_name, + entry.index_file.external_path.as_deref(), ); for (data_file_name, meta) in ranges { let deletion_file = crate::DeletionFile::new( @@ -657,15 +717,19 @@ impl DataEvolutionDeleteWriter { async fn write_deletion_vector_index_file( &self, + partition: &[u8], + bucket: i32, mut bitmaps: IndexMap, ) -> Result { bitmaps.sort_keys(); let file_name = format!("index-{}-1", Uuid::new_v4()); - let table_path = self.table.location().trim_end_matches('/'); - let index_dir = format!("{table_path}/{INDEX_DIR}"); - self.table.file_io().mkdirs(&index_dir).await?; - let path = format!("{index_dir}/{file_name}"); + // Write where the reader resolves it, so a deletion vector written here is + // found again on the next scan. + let layout = self.deletion_vector_layout(partition, bucket)?; + let location = layout.location(); + let path = location.resolve(&file_name, None); + self.table.file_io().mkdirs(&location.directory()).await?; let mut bytes = vec![DELETION_VECTORS_INDEX_VERSION_V1]; let mut ranges = IndexMap::new(); @@ -715,6 +779,7 @@ impl DataEvolutionDeleteWriter { file_size, row_count: i64::from(row_count), deletion_vectors_ranges: Some(ranges), + external_path: None, global_index_meta: None, }) } diff --git a/crates/paimon/src/table/full_text_search_builder.rs b/crates/paimon/src/table/full_text_search_builder.rs index 81beacc2..fac4db4a 100644 --- a/crates/paimon/src/table/full_text_search_builder.rs +++ b/crates/paimon/src/table/full_text_search_builder.rs @@ -31,6 +31,7 @@ use crate::table::global_index_scanner::{ deleted_row_ranges_for_data_evolution_dvs, search_limit_with_deleted_rows, unindexed_ranges_for_global_index_entries, RowRangeIndex, }; +use crate::table::index_file_path::IndexFileLocation; use crate::table::pk_full_text_read::PrimaryKeyFullTextRead; use crate::table::pk_full_text_scan::PrimaryKeyFullTextScan; use crate::table::{ @@ -44,7 +45,6 @@ use roaring::RoaringTreemap; use serde_json::json; use std::collections::{HashMap, HashSet}; -const INDEX_DIR: &str = "index"; const FULL_TEXT_INDEX_TYPE: &str = "full-text"; const FULL_TEXT_INDEX_SEARCH_CONCURRENCY: usize = 8; @@ -278,6 +278,10 @@ impl<'a> FullTextSearchBuilder<'a> { self.table.file_io().clone(), materialize_reader, self.table.location().trim_end_matches('/').to_string(), + self.table + .schema() + .core_options() + .index_file_in_data_file_dir(), ); read.read(&plan, query_text, limit).await } @@ -359,7 +363,10 @@ async fn evaluate_full_text_search( .map(|plan| { let entry = plan.entry; let global_meta = entry.index_file.global_index_meta.as_ref().unwrap(); - let path = format!("{table_path}/{INDEX_DIR}/{}", entry.index_file.file_name); + let path = IndexFileLocation::Global { table_path }.resolve( + &entry.index_file.file_name, + entry.index_file.external_path.as_deref(), + ); let file_name = entry.index_file.file_name.clone(); let query_text = search.query_text.clone(); let local_filter = plan.local_filter; @@ -1014,6 +1021,7 @@ mod tests { file_size: i64::try_from(index_bytes.len()).unwrap(), row_count: 2, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: 100, row_range_end: 101, @@ -1095,6 +1103,7 @@ mod tests { file_size: i64::try_from(index_bytes.len()).unwrap(), row_count: 2, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: 100, row_range_end: 101, @@ -1272,6 +1281,7 @@ mod tests { file_size: 0, row_count: end - start + 1, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: start, row_range_end: end, diff --git a/crates/paimon/src/table/global_index_drop_builder.rs b/crates/paimon/src/table/global_index_drop_builder.rs index 505139fe..170e7097 100644 --- a/crates/paimon/src/table/global_index_drop_builder.rs +++ b/crates/paimon/src/table/global_index_drop_builder.rs @@ -261,6 +261,7 @@ mod tests { file_size: 128, row_count: (row_range_end - row_range_start + 1), deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start, row_range_end, @@ -279,6 +280,7 @@ mod tests { file_size: 64, row_count: 1, deletion_vectors_ranges: None, + external_path: None, global_index_meta: None, } } @@ -297,6 +299,7 @@ mod tests { cardinality: Some(1), }, )])), + external_path: None, global_index_meta: None, } } diff --git a/crates/paimon/src/table/global_index_scanner.rs b/crates/paimon/src/table/global_index_scanner.rs index b4360426..05eb460a 100644 --- a/crates/paimon/src/table/global_index_scanner.rs +++ b/crates/paimon/src/table/global_index_scanner.rs @@ -36,6 +36,7 @@ use crate::spec::{ DataField, DataType, Datum, FileKind, GlobalIndexSearchMode, IndexFileMeta, IndexManifestEntry, Predicate, PredicateOperator, }; +use crate::table::index_file_path::IndexFileLocation; use crate::table::{DeletionFile, RowRange, Table}; use crate::{Error, Result}; use futures::{StreamExt, TryStreamExt}; @@ -58,7 +59,6 @@ type EvaluateFuture<'a> = std::pin::Pin< type PredicateTuple<'a> = (PredicateOperator, &'a [Datum], &'a DataType); const DELETION_VECTORS_INDEX_TYPE: &str = "DELETION_VECTORS"; -const INDEX_DIR: &str = "index"; async fn try_fold_bounded( futures: impl IntoIterator, @@ -138,7 +138,8 @@ pub(crate) struct GlobalIndexScanner { coverage_by_field: HashMap>, /// Schema fields for field_id lookup. schema_fields: Vec, - /// Cache of opened BTree readers, keyed by file name. + /// Cache of opened BTree readers, keyed by resolved path: two entries can + /// share a file name yet resolve to different locations. reader_cache: Mutex>>, #[cfg(test)] query_io_probe: Option>, @@ -147,12 +148,22 @@ pub(crate) struct GlobalIndexScanner { /// A resolved global index entry with parsed metadata. struct GlobalIndexEntry { file_name: String, + external_path: Option, index_type: GlobalIndexFileKind, file_size: i64, row_range_start: i64, meta: BTreeIndexMeta, } +impl GlobalIndexEntry { + /// The entry's on-disk path: its external path if set, else the table's + /// global index directory. Also the BTree reader-cache key. + fn resolved_path(&self, table_path: &str) -> String { + IndexFileLocation::Global { table_path } + .resolve(&self.file_name, self.external_path.as_deref()) + } +} + #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum GlobalIndexFileKind { BTree, @@ -362,6 +373,7 @@ impl GlobalIndexScanner { let resolved = GlobalIndexEntry { file_name: entry.index_file.file_name.clone(), + external_path: entry.index_file.external_path.clone(), index_type: match index_type { BTREE_GLOBAL_INDEX_TYPE => GlobalIndexFileKind::BTree, BITMAP_GLOBAL_INDEX_TYPE => GlobalIndexFileKind::Bitmap, @@ -864,9 +876,10 @@ impl GlobalIndexScanner { } // Each concurrent task owns its reader. Only return it to the shared - // cache after all predicates for this shard have completed. + // cache after all predicates for this shard have completed. Keyed by the + // resolved path so it matches the take in `get_or_open_reader`. if let Some(OpenedGlobalIndexReader::BTree(reader)) = reader.take() { - self.return_reader(entry.file_name.clone(), reader); + self.return_reader(entry.resolved_path(&self.table_path), reader); } Ok(file_result) } @@ -882,24 +895,28 @@ impl GlobalIndexScanner { } } - /// Get a cached reader or open a new one for the given file. + /// Get a cached reader or open a new one for the given resolved path. The + /// cache is keyed by the resolved path (not the bare file name) so two + /// entries that share a file name but resolve to different locations — e.g. + /// distinct external paths — never reuse each other's reader. async fn get_or_open_reader( &self, entry: &GlobalIndexEntry, meta: &BTreeIndexMeta, data_type: &DataType, ) -> Result { + let resolved_path = entry.resolved_path(&self.table_path); + // Try to take from cache { let mut cache = self.reader_cache.lock().unwrap(); - if let Some(reader) = cache.remove(&entry.file_name) { + if let Some(reader) = cache.remove(&resolved_path) { return Ok(OpenedGlobalIndexReader::BTree(reader)); } } // Open new reader - let path = format!("{}/{INDEX_DIR}/{}", self.table_path, entry.file_name); - let input = self.file_io.new_input(&path)?; + let input = self.file_io.new_input(&resolved_path)?; let file_size = if entry.file_size > 0 { entry.file_size as u64 } else { @@ -912,7 +929,7 @@ impl GlobalIndexScanner { .await .map(OpenedGlobalIndexReader::BTree) .map_err(|e| crate::Error::DataInvalid { - message: format!("Failed to open BTree index file: {}", entry.file_name), + message: format!("Failed to open BTree index file: {resolved_path}"), source: Some(Box::new(e)), }) } @@ -954,7 +971,7 @@ impl GlobalIndexScanner { &self, entry: &GlobalIndexEntry, ) -> std::io::Result { - let path = format!("{}/{INDEX_DIR}/{}", self.table_path, entry.file_name); + let path = entry.resolved_path(&self.table_path); let input = self .file_io .new_input(&path) @@ -1018,9 +1035,9 @@ impl GlobalIndexScanner { } /// Return a reader to the cache for future reuse. - fn return_reader(&self, file_name: String, reader: BTreeIndexReader) { + fn return_reader(&self, resolved_path: String, reader: BTreeIndexReader) { let mut cache = self.reader_cache.lock().unwrap(); - cache.insert(file_name, reader); + cache.insert(resolved_path, reader); } fn find_field_id_by_name(&self, column: &str) -> Result> { @@ -1354,9 +1371,16 @@ pub(crate) async fn deleted_row_ranges_for_data_evolution_dvs( .await?; let mut first_row_ids: HashMap<(Vec, i32, String), i64> = HashMap::new(); + // A deletion vector is an index file, so it may live beside its bucket's data + // files. Capture each bucket's directory from the plan rather than rebuilding + // it, so custom data directories are honored. + let mut bucket_paths: HashMap<(Vec, i32), String> = HashMap::new(); for split in plan.splits() { let partition = split.partition().to_serialized_bytes(); let bucket = split.bucket(); + bucket_paths + .entry((partition.clone(), bucket)) + .or_insert_with(|| split.bucket_path().to_string()); for file in split.data_files() { if let Some(first_row_id) = file.first_row_id { first_row_ids.insert( @@ -1369,6 +1393,7 @@ pub(crate) async fn deleted_row_ranges_for_data_evolution_dvs( let mut ranges = Vec::new(); let table_path = table.location().trim_end_matches('/'); + let index_file_in_data_file_dir = table.schema().core_options().index_file_in_data_file_dir(); for entry in index_entries { if entry.kind != FileKind::Add || entry.index_file.index_type != DELETION_VECTORS_INDEX_TYPE { @@ -1377,7 +1402,10 @@ pub(crate) async fn deleted_row_ranges_for_data_evolution_dvs( let Some(dv_ranges) = entry.index_file.deletion_vectors_ranges.as_ref() else { continue; }; - let index_path = format!("{table_path}/{INDEX_DIR}/{}", entry.index_file.file_name); + // A deletion vector is resolved against the bucket that owns it; a bucket with + // no captured directory has no live split, and the row-id join below rejects + // every data file in the entry before any path is needed. + let bucket_path = bucket_paths.get(&(entry.partition.clone(), entry.bucket)); for (data_file_name, meta) in dv_ranges { let key = ( entry.partition.clone(), @@ -1393,8 +1421,25 @@ pub(crate) async fn deleted_row_ranges_for_data_evolution_dvs( source: None, } })?; + // The join above found a live row-tracked file in this bucket, so the + // loop over the plan captured its directory. + let bucket_path = bucket_path.ok_or_else(|| crate::Error::DataInvalid { + message: format!( + "no bucket directory captured for deletion vector '{}'", + entry.index_file.file_name + ), + source: None, + })?; let deletion_file = DeletionFile::new( - index_path.clone(), + IndexFileLocation::BucketLocal { + table_path, + bucket_path, + index_file_in_data_file_dir, + } + .resolve( + &entry.index_file.file_name, + entry.index_file.external_path.as_deref(), + ), meta.offset as i64, meta.length as i64, meta.cardinality, @@ -1975,6 +2020,7 @@ mod tests { file_size: 0, row_count: 0, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start, row_range_end, diff --git a/crates/paimon/src/table/hybrid_search_builder.rs b/crates/paimon/src/table/hybrid_search_builder.rs index 5068dc80..af6dbb8c 100644 --- a/crates/paimon/src/table/hybrid_search_builder.rs +++ b/crates/paimon/src/table/hybrid_search_builder.rs @@ -647,6 +647,7 @@ impl<'a> HybridSearchBuilder<'a> { table.file_io().clone(), materialize_reader, table.location().trim_end_matches('/').to_string(), + table.schema().core_options().index_file_in_data_file_dir(), ); let result = read.search_route(&plan, query, route.limit).await?; let positions = result @@ -1387,6 +1388,7 @@ mod pk_hybrid_tests { file_size: i64::try_from(vector_index_size).unwrap(), row_count, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: 0, row_range_end: row_count - 1, @@ -1420,6 +1422,7 @@ mod pk_hybrid_tests { file_size: 1, row_count, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: 0, row_range_end: row_count - 1, diff --git a/crates/paimon/src/table/index_file_path.rs b/crates/paimon/src/table/index_file_path.rs new file mode 100644 index 00000000..26d71f6a --- /dev/null +++ b/crates/paimon/src/table/index_file_path.rs @@ -0,0 +1,177 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Resolves the on-disk path of an index file recorded in an index manifest. +//! +//! An index file's location depends on how it was written: +//! * an externally-stored file records its absolute path in the manifest and +//! is read from exactly that path, wherever it lives; +//! * a global index (data-evolution, row-id space) always lives under the +//! table's `index/` directory; +//! * a source-backed primary-key index lives beside its bucket's data files +//! when the table stores index files in the data-file directory, and under +//! the table `index/` directory otherwise. +//! +//! The bucket-local fallback resolves against the bucket directory captured on +//! the data split, never a path rebuilt from the table root, so custom data +//! directories and postpone-bucket layouts are honored. + +const INDEX_DIR: &str = "index"; + +/// How to resolve an index file that carries no explicit external path. +pub(crate) enum IndexFileLocation<'a> { + /// Global index files (data-evolution row-id space) always live under the + /// table's `index/` directory. + Global { table_path: &'a str }, + /// Source-backed primary-key index files live beside their bucket's data + /// files when the table keeps index files in the data-file directory, and + /// under the table `index/` directory otherwise. + BucketLocal { + table_path: &'a str, + /// The bucket directory captured from the data split (e.g. + /// `warehouse/db/tbl/bucket-3`). Used directly, not rebuilt. + bucket_path: &'a str, + /// Whether the table stores index files in the data-file (bucket) + /// directory (`index-file-in-data-file-dir`). + index_file_in_data_file_dir: bool, + }, +} + +impl IndexFileLocation<'_> { + /// The directory a file with no explicit external path resolves into. A + /// writer needs it to create the directory it is about to write into, so it + /// must come from here rather than be re-derived from the resolved path. + pub(crate) fn directory(&self) -> String { + match self { + IndexFileLocation::Global { table_path } => format!("{table_path}/{INDEX_DIR}"), + IndexFileLocation::BucketLocal { + table_path, + bucket_path, + index_file_in_data_file_dir, + } => { + if *index_file_in_data_file_dir { + (*bucket_path).to_string() + } else { + format!("{table_path}/{INDEX_DIR}") + } + } + } + } + + /// Resolve the full path of `file_name`, honoring an explicit + /// `external_path` when present. + pub(crate) fn resolve(&self, file_name: &str, external_path: Option<&str>) -> String { + match external_path { + Some(external) => external.to_string(), + None => format!("{}/{file_name}", self.directory()), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn external_path_wins_over_every_mode() { + let external = "s3://other-bucket/abs/idx-0"; + let global = IndexFileLocation::Global { + table_path: "warehouse/db/tbl", + }; + assert_eq!(global.resolve("idx-0", Some(external)), external); + + let bucket_local = IndexFileLocation::BucketLocal { + table_path: "warehouse/db/tbl", + bucket_path: "warehouse/db/tbl/bucket-3", + index_file_in_data_file_dir: true, + }; + assert_eq!(bucket_local.resolve("idx-0", Some(external)), external); + } + + #[test] + fn global_uses_table_index_directory() { + let loc = IndexFileLocation::Global { + table_path: "warehouse/db/tbl", + }; + assert_eq!(loc.resolve("idx-0", None), "warehouse/db/tbl/index/idx-0"); + } + + #[test] + fn bucket_local_uses_bucket_directory_when_enabled() { + let loc = IndexFileLocation::BucketLocal { + table_path: "warehouse/db/tbl", + bucket_path: "warehouse/db/tbl/bucket-3", + index_file_in_data_file_dir: true, + }; + assert_eq!( + loc.resolve("idx-0", None), + "warehouse/db/tbl/bucket-3/idx-0" + ); + } + + #[test] + fn bucket_local_falls_back_to_table_index_when_disabled() { + let loc = IndexFileLocation::BucketLocal { + table_path: "warehouse/db/tbl", + bucket_path: "warehouse/db/tbl/bucket-3", + index_file_in_data_file_dir: false, + }; + assert_eq!(loc.resolve("idx-0", None), "warehouse/db/tbl/index/idx-0"); + } + + #[test] + fn bucket_local_uses_captured_custom_bucket_path() { + // A custom data directory / postpone-bucket layout must be honored via + // the captured bucket path, not a path rebuilt from the table root. + let loc = IndexFileLocation::BucketLocal { + table_path: "warehouse/db/tbl", + bucket_path: "s3://data-warehouse/custom/tbl/bucket-postpone", + index_file_in_data_file_dir: true, + }; + assert_eq!( + loc.resolve("idx-0", None), + "s3://data-warehouse/custom/tbl/bucket-postpone/idx-0" + ); + } + + #[test] + fn directory_is_the_parent_resolve_writes_into() { + // A writer creates `directory()` and then writes `resolve()`; the two must + // agree, or it creates one directory and writes into another. + let locations = [ + IndexFileLocation::Global { + table_path: "warehouse/db/tbl", + }, + IndexFileLocation::BucketLocal { + table_path: "warehouse/db/tbl", + bucket_path: "warehouse/db/tbl/pt=1/bucket-3", + index_file_in_data_file_dir: false, + }, + IndexFileLocation::BucketLocal { + table_path: "warehouse/db/tbl", + bucket_path: "warehouse/db/tbl/pt=1/bucket-3", + index_file_in_data_file_dir: true, + }, + ]; + for loc in &locations { + assert_eq!( + loc.resolve("idx-0", None), + format!("{}/idx-0", loc.directory()) + ); + } + } +} diff --git a/crates/paimon/src/table/lumina_index_build_builder.rs b/crates/paimon/src/table/lumina_index_build_builder.rs index d340c97e..531ba807 100644 --- a/crates/paimon/src/table/lumina_index_build_builder.rs +++ b/crates/paimon/src/table/lumina_index_build_builder.rs @@ -247,6 +247,7 @@ impl<'a> LuminaIndexBuildBuilder<'a> { )?, row_count, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: shard.row_range_start, row_range_end: shard.row_range_end, @@ -1711,6 +1712,7 @@ mod tests { file_size: 1, row_count: (end - start + 1), deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: start, row_range_end: end, diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs index 35f1e45a..720f157f 100644 --- a/crates/paimon/src/table/mod.rs +++ b/crates/paimon/src/table/mod.rs @@ -53,6 +53,7 @@ pub(crate) mod global_index_scanner; mod global_index_types; mod hybrid_search_builder; mod incremental_scan; +pub(crate) mod index_file_path; mod kv_file_reader; mod kv_file_writer; mod lumina_index_build_builder; diff --git a/crates/paimon/src/table/pk_full_text_bucket_search.rs b/crates/paimon/src/table/pk_full_text_bucket_search.rs index ecb2ba23..a92395cf 100644 --- a/crates/paimon/src/table/pk_full_text_bucket_search.rs +++ b/crates/paimon/src/table/pk_full_text_bucket_search.rs @@ -37,6 +37,7 @@ use crate::deletion_vector::DeletionVector; use crate::ftindex::reader::FullTextArchiveReader; use crate::io::FileIO; use crate::spec::PrimaryKeyIndexSourceMeta; +use crate::table::index_file_path::IndexFileLocation; use crate::table::pk_full_text_read::PrimaryKeyFullTextCandidate; use crate::table::pk_full_text_scan::PrimaryKeyFullTextSearchSplit; @@ -187,10 +188,11 @@ fn owning_active_source( /// Search one bucket's full-text payloads and return its scored candidates /// (unsorted; the read path fuses them cross-bucket via `top_k_by_score`). /// -/// `dvs` is keyed by data file name; `table_path` roots the index directory -/// (`{table_path}/index/{payload}`). Mirrors Java +/// `dvs` is keyed by data file name. Each payload is opened at its resolved +/// bucket-local (or external) path. Mirrors Java /// `PrimaryKeyFullTextBucketSearch.searchRankings`, flattened to one candidate /// list per bucket. +#[allow(clippy::too_many_arguments)] pub(crate) async fn search_bucket( split: &PrimaryKeyFullTextSearchSplit, query: &str, @@ -198,6 +200,7 @@ pub(crate) async fn search_bucket( dvs: &HashMap, file_io: &FileIO, table_path: &str, + index_file_in_data_file_dir: bool, split_index: usize, ) -> crate::Result> { if limit == 0 { @@ -243,7 +246,12 @@ pub(crate) async fn search_bucket( } } - let path = format!("{table_path}/index/{}", payload.file_name); + let path = IndexFileLocation::BucketLocal { + table_path, + bucket_path: data_split.bucket_path(), + index_file_in_data_file_dir, + } + .resolve(&payload.file_name, payload.external_path.as_deref()); let input = file_io.new_input(&path)?; let reader = FullTextArchiveReader::from_input_file(&input).await?; let hits = match &prepared.include { @@ -371,6 +379,7 @@ mod tests { file_size: 1, row_count: total, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(gim(7, frame(level, files))), } } @@ -458,6 +467,7 @@ mod tests { &dvs, &file_io, table_path, + false, 0, ) .await @@ -479,6 +489,46 @@ mod tests { ); } + /// A full-text archive is an index file, so with `index-file-in-data-file-dir` + /// it lives beside the bucket's data files. The directory must come from the + /// split, not be rebuilt from the table root. + #[tokio::test] + async fn archive_resolves_into_the_split_bucket_directory() { + let file_io = FileIOBuilder::new("memory").build().unwrap(); + let table_path = "memory:/ftbs_bucket_local"; + let bytes = build_archive(&[(0, "alpha"), (1, "beta"), (2, "alpha")]); + // `data_split` roots this bucket outside the table path on purpose. + write_archive(&file_io, "memory:/t/bucket-0/ft-0", bytes).await; + + let split = PrimaryKeyFullTextSearchSplit::new( + data_split(vec![dfm("d0", 3)]), + vec![ft_payload("ft-0", 1, &[("d0", 3)])], + Vec::new(), + ) + .unwrap(); + + let dvs: HashMap = HashMap::new(); + let out = search_bucket( + &split, + r#"{"match":{"query":"alpha"}}"#, + 10, + &dvs, + &file_io, + table_path, + true, + 0, + ) + .await + .unwrap(); + + let mut got: Vec<(String, i64)> = out + .iter() + .map(|c| (c.data_file_name.clone(), c.row_position)) + .collect(); + got.sort(); + assert_eq!(got, vec![("d0".to_string(), 0), ("d0".to_string(), 2)]); + } + // ---- (b) a DV-deleted archive position is excluded from results ---- #[tokio::test] async fn deletion_vector_excludes_matched_position() { @@ -515,6 +565,7 @@ mod tests { &dvs, &file_io, table_path, + false, 0, ) .await @@ -635,6 +686,7 @@ mod tests { &dvs, &file_io, "memory:/x", + false, 0 ) .await diff --git a/crates/paimon/src/table/pk_full_text_bucket_state.rs b/crates/paimon/src/table/pk_full_text_bucket_state.rs index c32cf6e4..18c41155 100644 --- a/crates/paimon/src/table/pk_full_text_bucket_state.rs +++ b/crates/paimon/src/table/pk_full_text_bucket_state.rs @@ -306,6 +306,7 @@ mod tests { file_size: 1, row_count, deletion_vectors_ranges: None, + external_path: None, global_index_meta, } } diff --git a/crates/paimon/src/table/pk_full_text_read.rs b/crates/paimon/src/table/pk_full_text_read.rs index d5eed2fc..2eadc2e8 100644 --- a/crates/paimon/src/table/pk_full_text_read.rs +++ b/crates/paimon/src/table/pk_full_text_read.rs @@ -277,6 +277,7 @@ pub(crate) struct PrimaryKeyFullTextRead { file_io: FileIO, materialize_reader: DataFileReader, table_path: String, + index_file_in_data_file_dir: bool, } impl PrimaryKeyFullTextRead { @@ -284,11 +285,13 @@ impl PrimaryKeyFullTextRead { file_io: FileIO, materialize_reader: DataFileReader, table_path: String, + index_file_in_data_file_dir: bool, ) -> Self { Self { file_io, materialize_reader, table_path, + index_file_in_data_file_dir, } } @@ -344,6 +347,7 @@ impl PrimaryKeyFullTextRead { &dvs, &self.file_io, &self.table_path, + self.index_file_in_data_file_dir, split_index, ) .await?; @@ -686,6 +690,7 @@ mod read_tests { file_size: 1, row_count: total, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: 0, row_range_end: 0, @@ -904,7 +909,8 @@ mod read_tests { .unwrap()], }; - let read = PrimaryKeyFullTextRead::new(file_io.clone(), reader, table_path.to_string()); + let read = + PrimaryKeyFullTextRead::new(file_io.clone(), reader, table_path.to_string(), false); let batches = collect( read.read(&plan, r#"{"match":{"query":"alpha"}}"#, 10) .await @@ -965,7 +971,8 @@ mod read_tests { .unwrap()], }; - let read = PrimaryKeyFullTextRead::new(file_io.clone(), reader, table_path.to_string()); + let read = + PrimaryKeyFullTextRead::new(file_io.clone(), reader, table_path.to_string(), false); let batches = collect( read.read(&plan, r#"{"match":{"query":"alpha"}}"#, 10) .await @@ -994,7 +1001,8 @@ mod read_tests { ) .unwrap()], }; - let read = PrimaryKeyFullTextRead::new(file_io.clone(), reader, table_path.to_string()); + let read = + PrimaryKeyFullTextRead::new(file_io.clone(), reader, table_path.to_string(), false); let batches = collect( read.read(&plan, r#"{"match":{"query":"zeta"}}"#, 10) .await @@ -1029,7 +1037,8 @@ mod read_tests { .unwrap()], }; - let read = PrimaryKeyFullTextRead::new(file_io.clone(), reader, table_path.to_string()); + let read = + PrimaryKeyFullTextRead::new(file_io.clone(), reader, table_path.to_string(), false); let route = read .search_route(&plan, r#"{"match":{"query":"alpha"}}"#, 10) .await @@ -1079,7 +1088,8 @@ mod read_tests { .unwrap()], }; - let read = PrimaryKeyFullTextRead::new(file_io.clone(), reader, table_path.to_string()); + let read = + PrimaryKeyFullTextRead::new(file_io.clone(), reader, table_path.to_string(), false); let route = read .search_route(&plan, r#"{"match":{"query":"zeta"}}"#, 10) .await diff --git a/crates/paimon/src/table/pk_full_text_scan.rs b/crates/paimon/src/table/pk_full_text_scan.rs index 2ffe0b49..350b17dc 100644 --- a/crates/paimon/src/table/pk_full_text_scan.rs +++ b/crates/paimon/src/table/pk_full_text_scan.rs @@ -517,6 +517,7 @@ mod tests { file_size: 1, row_count, deletion_vectors_ranges: None, + external_path: None, global_index_meta, } } diff --git a/crates/paimon/src/table/pk_vector_scan.rs b/crates/paimon/src/table/pk_vector_scan.rs index cfc528b2..8521aad6 100644 --- a/crates/paimon/src/table/pk_vector_scan.rs +++ b/crates/paimon/src/table/pk_vector_scan.rs @@ -23,15 +23,24 @@ use std::collections::{BTreeMap, HashSet}; use crate::spec::{ - should_read_pk_index_source, BinaryRow, DataFileMeta, FileKind, GlobalIndexMeta, IndexManifest, - Predicate, PrimaryKeyIndexSourceFile, PrimaryKeyIndexSourceMeta, + should_read_pk_index_source, BinaryRow, DataFileMeta, FileKind, IndexManifest, Predicate, + PrimaryKeyIndexSourceFile, PrimaryKeyIndexSourceMeta, }; +use crate::table::index_file_path::IndexFileLocation; use crate::table::pk_vector_orchestrator::PkVectorSearchSplit; use crate::table::source::{DataSplit, DataSplitBuilder, DeletionFile}; use crate::table::Table; use crate::vindex::pkvector::bucket::{BucketActiveFile, BucketAnnSegment}; -const INDEX_DIR: &str = "index"; +/// A payload whose bucket-local path is resolved in planning Phase C, once the +/// owning bucket's data split (and directory) is known. +struct UnresolvedAnnSegment { + source_meta: PrimaryKeyIndexSourceMeta, + file_name: String, + external_path: Option, + file_size: u64, + index_meta: Vec, +} fn data_invalid(message: impl Into) -> crate::Error { crate::Error::DataInvalid { @@ -294,22 +303,44 @@ impl<'a> PkVectorScan<'a> { continue; } let partition = BinaryRow::from_serialized_bytes(&entry.partition)?; - let resolved_path = - format!("{table_path}/{INDEX_DIR}/{}", entry.index_file.file_name); let file_size = u64::try_from(entry.index_file.file_size) .map_err(|_| data_invalid("index file size must not be negative"))?; + let source_meta = + PrimaryKeyIndexSourceMeta::from_global_index_meta(&gim).map_err(|_| { + data_invalid(format!( + "index file {} is not active", + entry.index_file.file_name + )) + })?; entries.push(( partition, entry.bucket, - gim, - resolved_path, - file_size, - entry.index_file.file_name.clone(), + UnresolvedAnnSegment { + source_meta, + file_name: entry.index_file.file_name.clone(), + external_path: entry.index_file.external_path.clone(), + file_size, + // The Lumina reader consumes this as its serialized index + // metadata; the vindex reader ignores it and loads metadata + // from the segment file bytes. Absent value defaults to empty. + index_meta: gim.index_meta.clone().unwrap_or_default(), + }, )); } } - let splits = plan_from_inputs(snapshot_id, data_splits, entries)?; + let index_file_in_data_file_dir = self + .table + .schema() + .core_options() + .index_file_in_data_file_dir(); + let splits = plan_from_inputs( + snapshot_id, + data_splits, + entries, + table_path, + index_file_in_data_file_dir, + )?; Ok(PkVectorScanPlan { snapshot_id, splits, @@ -320,32 +351,22 @@ impl<'a> PkVectorScan<'a> { /// Pure planning core, drivable without a live snapshot: group ANN payloads and /// data splits by `(partition, bucket)`, then assemble one search split per /// bucket that has data. Index-only buckets are dropped, not errored. -#[allow(clippy::type_complexity)] fn plan_from_inputs( snapshot_id: i64, data_splits: Vec, - index_entries: Vec<(BinaryRow, i32, GlobalIndexMeta, String, u64, String)>, + index_entries: Vec<(BinaryRow, i32, UnresolvedAnnSegment)>, + table_path: &str, + index_file_in_data_file_dir: bool, ) -> crate::Result> { type Key = (Vec, i32); - // Phase A: group ANN payloads by (partition, bucket). - let mut segments_by_bucket: BTreeMap> = BTreeMap::new(); - for (partition, bucket, gim, path, file_size, file_name) in index_entries { - let source_meta = PrimaryKeyIndexSourceMeta::from_global_index_meta(&gim) - .map_err(|_| data_invalid(format!("index file {file_name} is not active")))?; + // Phase A: group unresolved ANN payloads by (partition, bucket). The on-disk + // path is resolved in Phase C, once the bucket's data split (and thus its + // bucket directory) is known. + let mut payloads_by_bucket: BTreeMap> = BTreeMap::new(); + for (partition, bucket, payload) in index_entries { let key = (partition.to_serialized_bytes(), bucket); - segments_by_bucket - .entry(key) - .or_default() - .push(BucketAnnSegment { - source_meta, - path, - file_size, - // The Lumina reader consumes this as its serialized index - // metadata; the vindex reader ignores it and loads metadata from - // the segment file bytes. Absent value defaults to an empty vec. - index_meta: gim.index_meta.clone().unwrap_or_default(), - }); + payloads_by_bucket.entry(key).or_default().push(payload); } // Phase B: group data splits by (partition, bucket). @@ -358,14 +379,28 @@ fn plan_from_inputs( acc.add(split)?; } - // Phase C: assemble one split per bucket that has data. + // Phase C: assemble one split per bucket that has data, resolving each + // payload's path against the bucket directory now that it is known. let mut out = Vec::new(); for (key, acc) in accum_by_bucket { let data_split = acc.build()?; - let ann_segments = current_ann_segments( - data_split.data_files(), - segments_by_bucket.remove(&key).unwrap_or_default(), - )?; + let location = IndexFileLocation::BucketLocal { + table_path, + bucket_path: data_split.bucket_path(), + index_file_in_data_file_dir, + }; + let resolved_segments: Vec = payloads_by_bucket + .remove(&key) + .unwrap_or_default() + .into_iter() + .map(|p| BucketAnnSegment { + source_meta: p.source_meta, + path: location.resolve(&p.file_name, p.external_path.as_deref()), + file_size: p.file_size, + index_meta: p.index_meta, + }) + .collect(); + let ann_segments = current_ann_segments(data_split.data_files(), resolved_segments)?; let active_files: Vec = data_split .data_files() .iter() @@ -381,7 +416,7 @@ fn plan_from_inputs( active_files, }); } - // Index-only buckets left in segments_by_bucket are intentionally dropped. + // Index-only buckets left in payloads_by_bucket are intentionally dropped. Ok(out) } @@ -491,51 +526,106 @@ mod tests { } } + /// An unresolved ANN payload as the manifest loop builds one. + fn payload( + file_name: &str, + gim: GlobalIndexMeta, + external_path: Option<&str>, + ) -> UnresolvedAnnSegment { + UnresolvedAnnSegment { + source_meta: PrimaryKeyIndexSourceMeta::from_global_index_meta(&gim).unwrap(), + file_name: file_name.to_string(), + external_path: external_path.map(str::to_string), + file_size: 10, + index_meta: gim.index_meta.clone().unwrap_or_default(), + } + } + #[test] fn drops_index_only_bucket_without_error() { // Payload for (part=[], bucket 0) but NO data split -> no split, no error. let entries = vec![( BinaryRow::new(0), 0, - gim(2, 5, &[("d0", 3)]), - "idx/seg0".to_string(), - 10u64, - "seg0".to_string(), + payload("seg0", gim(2, 5, &[("d0", 3)]), None), )]; - let splits = plan_from_inputs(1, Vec::new(), entries).unwrap(); + let splits = plan_from_inputs(1, Vec::new(), entries, "memory:/t", false).unwrap(); assert!(splits.is_empty()); } - #[test] - fn builds_one_split_per_bucket_with_data() { - let entries = vec![( - BinaryRow::new(0), - 0, - gim(2, 5, &[("d0", 3)]), - "idx/seg0".to_string(), - 10u64, - "seg0".to_string(), - )]; - let data = DataSplitBuilder::new() + /// One split whose bucket directory is not derivable from the table root, so a + /// resolved segment path proves the split's own bucket path was used. + fn bucket_split(bucket_path: &str) -> DataSplit { + DataSplitBuilder::new() .with_snapshot(1) .with_partition(BinaryRow::new(0)) .with_bucket(0) - .with_bucket_path("memory:/t/bucket-0".to_string()) + .with_bucket_path(bucket_path.to_string()) .with_total_buckets(1) .with_data_files(vec![dfm("d0", 3, 5, Some(1))]) .build() - .unwrap(); - let splits = plan_from_inputs(1, vec![data], entries).unwrap(); + .unwrap() + } + + #[test] + fn builds_one_split_per_bucket_with_data() { + let entries = vec![( + BinaryRow::new(0), + 0, + payload("seg0", gim(2, 5, &[("d0", 3)]), None), + )]; + let data = bucket_split("memory:/t/bucket-0"); + let splits = plan_from_inputs(1, vec![data], entries, "memory:/t", false).unwrap(); assert_eq!(splits.len(), 1); assert_eq!(splits[0].ann_segments.len(), 1); let seg = &splits[0].ann_segments[0]; - assert_eq!(seg.path, "idx/seg0"); + assert_eq!(seg.path, "memory:/t/index/seg0"); assert_eq!(seg.file_size, 10); assert_eq!(seg.source_meta.resolve(0).unwrap(), ("d0".to_string(), 0)); assert_eq!(splits[0].active_files.len(), 1); // d0 is COMPACT + level>0 assert_eq!(splits[0].active_files[0].file_name, "d0"); } + #[test] + fn ann_segment_resolves_into_the_split_bucket_directory() { + // The reported failure: a table with `index-file-in-data-file-dir` keeps its + // ANN segments beside the bucket's data files, and the search opened + // `
/index/` instead. The directory must come from the split, so + // a bucket path that is not derivable from the table root still resolves. + let entries = vec![( + BinaryRow::new(0), + 0, + payload("seg0", gim(2, 5, &[("d0", 3)]), None), + )]; + let data = bucket_split("s3://elsewhere/t/pt=1/bucket-0"); + let splits = plan_from_inputs(1, vec![data], entries, "memory:/t", true).unwrap(); + assert_eq!( + splits[0].ann_segments[0].path, + "s3://elsewhere/t/pt=1/bucket-0/seg0" + ); + } + + #[test] + fn ann_segment_external_path_wins_over_both_layouts() { + for index_file_in_data_file_dir in [false, true] { + let entries = vec![( + BinaryRow::new(0), + 0, + payload("seg0", gim(2, 5, &[("d0", 3)]), Some("s3://other/ann/seg0")), + )]; + let data = bucket_split("memory:/t/bucket-0"); + let splits = plan_from_inputs( + 1, + vec![data], + entries, + "memory:/t", + index_file_in_data_file_dir, + ) + .unwrap(); + assert_eq!(splits[0].ann_segments[0].path, "s3://other/ann/seg0"); + } + } + #[test] fn current_segments_require_exact_level_source_set() { let active = vec![ @@ -584,7 +674,7 @@ mod tests { .with_data_files(vec![dfm("d0", 3, 5, Some(1))]) .build() .unwrap(); - assert!(plan_from_inputs(1, vec![data], Vec::new()).is_err()); + assert!(plan_from_inputs(1, vec![data], Vec::new(), "memory:/t", false).is_err()); } #[test] @@ -609,7 +699,9 @@ mod tests { .with_data_files(vec![dfm("dup", 3, 5, Some(1))]) .build() .unwrap(); - assert!(plan_from_inputs(1, vec![split_a, split_b], Vec::new()).is_err()); + assert!( + plan_from_inputs(1, vec![split_a, split_b], Vec::new(), "memory:/t", false).is_err() + ); } #[test] @@ -628,7 +720,7 @@ mod tests { .with_data_deletion_files(vec![None, Some(dv)]) .build() .unwrap(); - let splits = plan_from_inputs(1, vec![data], Vec::new()).unwrap(); + let splits = plan_from_inputs(1, vec![data], Vec::new(), "memory:/t", false).unwrap(); assert_eq!(splits.len(), 1); let dvs = splits[0] .data_split diff --git a/crates/paimon/src/table/referenced_files.rs b/crates/paimon/src/table/referenced_files.rs index 851e8719..d4bb617b 100644 --- a/crates/paimon/src/table/referenced_files.rs +++ b/crates/paimon/src/table/referenced_files.rs @@ -36,6 +36,9 @@ use crate::table::{BranchManager, SnapshotManager, TagManager}; use futures::future::try_join_all; use futures::stream::{self, StreamExt, TryStreamExt}; +/// Name prefix of an index file, Java `FileStorePathFactory.INDEX_PREFIX`. +const INDEX_FILE_PREFIX: &str = "index-"; + /// Per-scope aggregated summary of referenced files (deduplicated). /// /// Each row represents the unique referenced files for a scope: @@ -665,7 +668,9 @@ fn is_partition_segment(segment: &str) -> bool { !key.is_empty() } -fn is_data_file_in_bucket(segments: &[&str], partition_depth: usize) -> bool { +/// Whether `segments` names a file directly inside a bucket directory, +/// `[/]bucket-N/`, whatever kind of file it is. +fn is_file_in_bucket(segments: &[&str], partition_depth: usize) -> bool { if segments.len() != partition_depth + 2 { return false; } @@ -674,7 +679,25 @@ fn is_data_file_in_bucket(segments: &[&str], partition_depth: usize) -> bool { .iter() .all(|segment| is_partition_segment(segment)) && is_bucket_dir_name(segments[partition_depth]) - && !segments[partition_depth + 1].starts_with("index-") +} + +/// An `index-` prefixed file in a bucket directory is an index file, not a data +/// file: that is where `index-file-in-data-file-dir` puts them. Classification +/// follows the physical form, not the current table option, so a file written +/// under one setting is still recognized after the setting changes — same as Java +/// `FileType.classify`, which maps any `index-*` basename to `BUCKET_INDEX`. +fn is_bucket_index_file_name(file_name: &str) -> bool { + file_name.starts_with(INDEX_FILE_PREFIX) +} + +fn is_data_file_in_bucket(segments: &[&str], partition_depth: usize) -> bool { + is_file_in_bucket(segments, partition_depth) + && !is_bucket_index_file_name(segments[partition_depth + 1]) +} + +fn is_index_file_in_bucket(segments: &[&str], partition_depth: usize) -> bool { + is_file_in_bucket(segments, partition_depth) + && is_bucket_index_file_name(segments[partition_depth + 1]) } fn is_data_file_in_data_dir( @@ -718,6 +741,7 @@ fn classify_physical_path( ["manifest", name] if is_manifest_file_name(name) => PhysicalFileKind::Manifest, ["statistics", _] => PhysicalFileKind::Statistics, ["index", _] => PhysicalFileKind::Index, + _ if is_index_file_in_bucket(&segments, partition_depth) => PhysicalFileKind::Index, _ => { if let Some(data_dir) = data_file_path_directory { let data_dir = table_relative_path(table_location, data_dir).unwrap_or(data_dir); @@ -1108,7 +1132,7 @@ mod tests { .await; write_test_file( &file_io, - &format!("{table_path}/bucket-0/index-should-not-be-data"), + &format!("{table_path}/bucket-0/index-in-bucket-dir"), "bucket index", ) .await; @@ -1158,7 +1182,13 @@ mod tests { .unwrap(); assert_eq!(result.manifest_file_count, 4); - assert_eq!(result.index_file_count, 1); + // `
/index/index-0` plus the `index-` prefixed file in a bucket + // directory, which `index-file-in-data-file-dir` puts there. + assert_eq!(result.index_file_count, 2); + assert_eq!( + result.index_file_size, + ("index".len() + "bucket index".len()) as i64 + ); assert_eq!(result.data_file_count, 3); } @@ -1203,7 +1233,12 @@ mod tests { .unwrap(); assert_eq!(result.data_file_count, 1); - assert_eq!(result.index_file_count, 0); + // A bucket-local index file counts as an index file at any partition depth. + assert_eq!(result.index_file_count, 1); + assert_eq!( + result.index_file_size, + "partition bucket index".len() as i64 + ); } #[tokio::test] diff --git a/crates/paimon/src/table/sorted_global_index_build_builder.rs b/crates/paimon/src/table/sorted_global_index_build_builder.rs index de72364b..5f44e39e 100644 --- a/crates/paimon/src/table/sorted_global_index_build_builder.rs +++ b/crates/paimon/src/table/sorted_global_index_build_builder.rs @@ -363,6 +363,7 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> { )?, row_count, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: shard.row_range_start, row_range_end: shard.row_range_end, @@ -3121,6 +3122,7 @@ mod tests { file_size: 1, row_count: (hole_end - hole_start + 1), deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: hole_start, row_range_end: hole_end, diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index 6e5d78c3..82e5b8f5 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -24,14 +24,15 @@ use crate::io::FileIO; use crate::spec::stats::BinaryTableStats; use crate::spec::FileKind; use crate::spec::{ - bucket_dir_name, extract_datum, merge_active_entries, BinaryRow, BinaryRowBuilder, CommitKind, - CoreOptions, DataFileMeta, DataType, Datum, GlobalIndexColumnUpdateAction, IndexManifest, - IndexManifestEntry, Manifest, ManifestEntry, ManifestFileMeta, ManifestList, PartitionComputer, - PartitionStatistics, Predicate, Snapshot, EMPTY_SERIALIZED_ROW, MANIFEST_ENTRY_SCHEMA, - POSTPONE_BUCKET, + bucket_path, bucket_path_under, extract_datum, merge_active_entries, BinaryRow, + BinaryRowBuilder, CommitKind, CoreOptions, DataFileMeta, DataType, Datum, + GlobalIndexColumnUpdateAction, IndexManifest, IndexManifestEntry, Manifest, ManifestEntry, + ManifestFileMeta, ManifestList, PartitionComputer, PartitionStatistics, Predicate, Snapshot, + EMPTY_SERIALIZED_ROW, MANIFEST_ENTRY_SCHEMA, POSTPONE_BUCKET, }; use crate::table::commit_message::CommitMessage; use crate::table::global_index_build_common::same_extra_field_ids; +use crate::table::index_file_path::IndexFileLocation; use crate::table::partition_filter::PartitionFilter; use crate::table::snapshot_commit::SnapshotCommit; use crate::table::{SnapshotManager, Table, TableScan}; @@ -704,6 +705,10 @@ impl TableCommit { .ensure_type_paimon_served(&self.table.identifier().full_name())?; self.table.ensure_not_branch_reference_for_write()?; + let table_path = self.table.location().trim_end_matches('/'); + let index_file_in_data_file_dir = + CoreOptions::new(self.table.schema().options()).index_file_in_data_file_dir(); + for message in commit_messages { let bucket_path = self.bucket_path(&message.partition, message.bucket)?; for file in message @@ -715,9 +720,18 @@ impl TableCommit { let _ = self.table.file_io().delete_file(&path).await; } } - let index_dir = format!("{}/index", self.table.location().trim_end_matches('/')); + // An index file must be deleted where it was written: beside this + // bucket's data files when the table keeps index files there, at its + // external path when it has one, and under the table `index/` + // directory otherwise. Mirrors Java `FileStoreCommitImpl.abort`, + // which deletes through `indexFileFactory(partition, bucket)`. + let index_location = IndexFileLocation::BucketLocal { + table_path, + bucket_path: &bucket_path, + index_file_in_data_file_dir, + }; for file in &message.new_index_files { - let path = format!("{index_dir}/{}", file.file_name); + let path = index_location.resolve(&file.file_name, file.external_path.as_deref()); let _ = self.table.file_io().delete_file(&path).await; } } @@ -725,13 +739,13 @@ impl TableCommit { } fn bucket_path(&self, partition: &[u8], bucket: i32) -> Result { - let base = self.table.location().trim_end_matches('/'); let partition_keys = self.table.schema().partition_keys(); if partition_keys.is_empty() { - return Ok(format!("{base}/{}", bucket_dir_name(bucket))); + // An unpartitioned table's buckets sit directly under the table path, + // so the partition blob is never decoded — callers are free to pass an + // empty one. + return Ok(bucket_path_under(self.table.location(), "", bucket)); } - - let partition_row = BinaryRow::from_serialized_bytes(partition)?; let core_options = CoreOptions::new(self.table.schema().options()); let computer = PartitionComputer::new( partition_keys, @@ -739,11 +753,12 @@ impl TableCommit { core_options.partition_default_name(), core_options.legacy_partition_name(), )?; - Ok(format!( - "{base}/{}{}", - computer.generate_partition_path(&partition_row)?, - bucket_dir_name(bucket) - )) + bucket_path( + self.table.location(), + Some(&computer), + &BinaryRow::from_serialized_bytes(partition)?, + bucket, + ) } /// Try to commit with retries. @@ -3290,6 +3305,7 @@ mod tests { file_size: 128, row_count: (row_range_end - row_range_start + 1), deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start, row_range_end, @@ -3330,6 +3346,7 @@ mod tests { cardinality: Some(1), }, )])), + external_path: None, global_index_meta: None, } } @@ -5482,6 +5499,7 @@ mod tests { file_size: 5, row_count: 1, deletion_vectors_ranges: None, + external_path: None, global_index_meta: None, }]; commit.abort(&[message]).await.unwrap(); @@ -5492,6 +5510,89 @@ mod tests { ); } + #[tokio::test] + async fn test_abort_deletes_index_files_from_the_data_file_directory() { + // With `index-file-in-data-file-dir`, a new index file is written beside the + // bucket's data files, so abort must delete it there. Deleting is best-effort + // (`let _ =`), so a wrong path leaks the file silently. + let file_io = test_file_io(); + let table_path = "memory:/test_abort_index_in_bucket_dir"; + setup_dirs(&file_io, table_path).await; + + let table = test_table_with_options( + &file_io, + table_path, + HashMap::from([( + "index-file-in-data-file-dir".to_string(), + "true".to_string(), + )]), + ); + let commit = TableCommit::new(table, "test-user".to_string()); + + let bucket_dir = format!("{table_path}/bucket-0"); + let index_path = format!("{bucket_dir}/index-in-bucket"); + file_io.mkdirs(&format!("{bucket_dir}/")).await.unwrap(); + file_io + .new_output(&index_path) + .unwrap() + .write(bytes::Bytes::from_static(b"index")) + .await + .unwrap(); + + let mut message = CommitMessage::new(vec![], 0, vec![]); + message.new_index_files = vec![IndexFileMeta { + index_type: "HASH".to_string(), + file_name: "index-in-bucket".to_string(), + file_size: 5, + row_count: 1, + deletion_vectors_ranges: None, + external_path: None, + global_index_meta: None, + }]; + commit.abort(&[message]).await.unwrap(); + + assert!( + !file_io.exists(&index_path).await.unwrap(), + "abort must remove an index file written into the bucket data-file directory" + ); + } + + #[tokio::test] + async fn test_abort_deletes_index_files_at_their_external_path() { + let file_io = test_file_io(); + let table_path = "memory:/test_abort_index_external"; + setup_dirs(&file_io, table_path).await; + + let commit = setup_commit(&file_io, table_path); + + let external_dir = "memory:/elsewhere/index"; + let external_path = format!("{external_dir}/index-external"); + file_io.mkdirs(&format!("{external_dir}/")).await.unwrap(); + file_io + .new_output(&external_path) + .unwrap() + .write(bytes::Bytes::from_static(b"index")) + .await + .unwrap(); + + let mut message = CommitMessage::new(vec![], 0, vec![]); + message.new_index_files = vec![IndexFileMeta { + index_type: "HASH".to_string(), + file_name: "index-external".to_string(), + file_size: 5, + row_count: 1, + deletion_vectors_ranges: None, + external_path: Some(external_path.clone()), + global_index_meta: None, + }]; + commit.abort(&[message]).await.unwrap(); + + assert!( + !file_io.exists(&external_path).await.unwrap(), + "abort must remove an index file recorded at an external path" + ); + } + #[tokio::test] async fn test_delete_conflict_rejects_missing_file() { let file_io = test_file_io(); diff --git a/crates/paimon/src/table/table_scan.rs b/crates/paimon/src/table/table_scan.rs index 601eeae9..929a9a75 100644 --- a/crates/paimon/src/table/table_scan.rs +++ b/crates/paimon/src/table/table_scan.rs @@ -34,13 +34,14 @@ use super::stats_filter::{ use super::{find_field_id_by_name, Table}; use crate::io::FileIO; use crate::spec::{ - avro::SharedSchemaCache, bucket_dir_name, BinaryRow, BucketFunctionType, CoreOptions, - DataField, DataFileMeta, FileKind, GlobalIndexSearchMode, IndexManifest, IndexManifestEntry, + avro::SharedSchemaCache, bucket_path, BinaryRow, BucketFunctionType, CoreOptions, DataField, + DataFileMeta, FileKind, GlobalIndexSearchMode, IndexManifest, IndexManifestEntry, ManifestEntry, PartitionComputer, Predicate, Snapshot, ROW_ID_FIELD_ID, ROW_ID_FIELD_NAME, SEQUENCE_NUMBER_FIELD_ID, SEQUENCE_NUMBER_FIELD_NAME, VALUE_KIND_FIELD_ID, VALUE_KIND_FIELD_NAME, }; use crate::table::bin_pack::split_for_batch; +use crate::table::index_file_path::IndexFileLocation; use crate::table::merge_tree_split_generator::{ merge_tree_split_for_batch, KeyComparator, SplitGroup, }; @@ -58,7 +59,6 @@ use std::sync::Arc; /// Path segment for manifest directory under table. const MANIFEST_DIR: &str = "manifest"; /// Path segment for index directory under table. -const INDEX_DIR: &str = "index"; const DELETION_VECTORS_INDEX_TYPE: &str = "DELETION_VECTORS"; #[derive(Debug, Default)] @@ -445,16 +445,42 @@ fn retain_index_manifest_entry_for_scan( })) } -/// Builds a map from (partition, bucket) to (data_file_name -> DeletionFile) from index manifest entries. -/// Only considers ADD entries with index_type "DELETION_VECTORS" and their deletion_vectors_ranges. +/// A deletion-vector entry whose on-disk path is not resolved yet. +/// +/// A deletion vector is an index file, so it follows the same layout rules as +/// every other index file: an explicit external path wins, otherwise it lives +/// beside its bucket's data files when the table keeps index files in the +/// data-file directory, and under the table `index/` directory otherwise. The +/// bucket directory is only known once a split's bucket path is computed, so +/// resolution is deferred until then. +#[derive(Debug, Clone, PartialEq, Eq)] +struct UnresolvedDeletionFile { + file_name: String, + external_path: Option, + offset: i64, + length: i64, + cardinality: Option, +} + +impl UnresolvedDeletionFile { + fn resolve(&self, location: &IndexFileLocation<'_>) -> DeletionFile { + DeletionFile::new( + location.resolve(&self.file_name, self.external_path.as_deref()), + self.offset, + self.length, + self.cardinality, + ) + } +} + +/// Builds a map from (partition, bucket) to (data_file_name -> deletion vector) from index manifest +/// entries. Only considers ADD entries with index_type "DELETION_VECTORS" and their +/// deletion_vectors_ranges. Paths stay unresolved; see [`UnresolvedDeletionFile`]. fn build_deletion_files_map( index_entries: &[crate::spec::IndexManifestEntry], - table_path: &str, -) -> HashMap> { +) -> HashMap> { use crate::spec::FileKind; - let table_path = table_path.trim_end_matches('/'); - let index_path_prefix = format!("{table_path}/{INDEX_DIR}"); - let mut map: HashMap> = + let mut map: HashMap> = HashMap::with_capacity(index_entries.len()); for entry in index_entries { if entry.kind != FileKind::Add { @@ -468,17 +494,17 @@ fn build_deletion_files_map( _ => continue, }; let key = PartitionBucket::new(entry.partition.clone(), entry.bucket); - let dv_path = format!("{}/{}", index_path_prefix, entry.index_file.file_name); let per_bucket = map.entry(key).or_default(); for (data_file_name, meta) in ranges { per_bucket.insert( data_file_name.clone(), - DeletionFile::new( - dv_path.clone(), - meta.offset as i64, - meta.length as i64, - meta.cardinality, - ), + UnresolvedDeletionFile { + file_name: entry.index_file.file_name.clone(), + external_path: entry.index_file.external_path.clone(), + offset: meta.offset as i64, + length: meta.length as i64, + cardinality: meta.cardinality, + }, ); } } @@ -1900,9 +1926,12 @@ impl<'a> PaimonTableScan<'a> { // The index manifest was read before data manifests so global-index row // ranges can prune manifest I/O. Reuse it here for deletion vectors. - let deletion_files_map = index_entries - .as_deref() - .map(|entries| build_deletion_files_map(entries, base_path)); + let deletion_files_map = index_entries.as_deref().map(build_deletion_files_map); + let index_file_in_data_file_dir = self + .table + .schema() + .core_options() + .index_file_in_data_file_dir(); let mut data_file_field_ids_cache = DataFileFieldIdsCache::new(); let can_push_down_limit = self.can_push_down_limit_hint(effective_row_ranges.as_deref()); @@ -1915,11 +1944,18 @@ impl<'a> PaimonTableScan<'a> { 'groups: for ((partition, bucket), (total_buckets, data_files)) in groups { let partition_row = BinaryRow::from_serialized_bytes(&partition)?; - let bucket_path = if let Some(ref computer) = partition_computer { - let partition_path = computer.generate_partition_path(&partition_row)?; - format!("{base_path}/{partition_path}{}", bucket_dir_name(bucket)) - } else { - format!("{base_path}/{}", bucket_dir_name(bucket)) + let bucket_path = bucket_path( + base_path, + partition_computer.as_ref(), + &partition_row, + bucket, + )?; + // Deletion vectors are index files, so they resolve against this bucket's + // directory, now that it is known. + let dv_location = IndexFileLocation::BucketLocal { + table_path: base_path, + bucket_path: &bucket_path, + index_file_in_data_file_dir, }; // Original `partition` Vec consumed by PartitionBucket for DV map lookup. @@ -2045,7 +2081,11 @@ impl<'a> PaimonTableScan<'a> { let data_deletion_files = per_bucket_deletion_map.map(|per_bucket| { file_group .iter() - .map(|f| per_bucket.get(&f.file_name).cloned()) + .map(|f| { + per_bucket + .get(&f.file_name) + .map(|unresolved| unresolved.resolve(&dv_location)) + }) .collect::>>() }); @@ -4234,22 +4274,122 @@ mod tests { cardinality: Some(33), }, )])), + external_path: None, global_index_meta: None, }, }]; - let map = super::build_deletion_files_map(&entries, "file:/tmp/table"); + let map = super::build_deletion_files_map(&entries); let by_bucket = map .get(&super::PartitionBucket::new(vec![1, 2, 3], 7)) .expect("partition bucket should exist"); - let deletion_file = by_bucket + let unresolved = by_bucket + .get("data-file.parquet") + .expect("deletion file should exist"); + + // Default layout: no external path, index files not in the data-file dir. + assert_eq!( + unresolved.resolve(&super::IndexFileLocation::BucketLocal { + table_path: "file:/tmp/table", + bucket_path: "file:/tmp/table/bucket-7", + index_file_in_data_file_dir: false, + }), + DeletionFile::new("file:/tmp/table/index/index-file".into(), 11, 22, Some(33)) + ); + } + + #[test] + fn test_deletion_vector_paths_follow_index_file_layout() { + let dv = super::UnresolvedDeletionFile { + file_name: "index-file".into(), + external_path: None, + offset: 11, + length: 22, + cardinality: Some(33), + }; + + // Index files under the table index directory (the default). + assert_eq!( + dv.resolve(&super::IndexFileLocation::BucketLocal { + table_path: "file:/tmp/table", + bucket_path: "file:/tmp/table/pt=1/bucket-7", + index_file_in_data_file_dir: false, + }) + .path(), + "file:/tmp/table/index/index-file" + ); + + // Index files kept beside the bucket's data files. + assert_eq!( + dv.resolve(&super::IndexFileLocation::BucketLocal { + table_path: "file:/tmp/table", + bucket_path: "file:/tmp/table/pt=1/bucket-7", + index_file_in_data_file_dir: true, + }) + .path(), + "file:/tmp/table/pt=1/bucket-7/index-file" + ); + } + + #[test] + fn test_deletion_vector_external_path_wins_over_both_layouts() { + let dv = super::UnresolvedDeletionFile { + file_name: "index-file".into(), + external_path: Some("s3://other/dv/index-file".into()), + offset: 0, + length: 1, + cardinality: None, + }; + + for index_file_in_data_file_dir in [false, true] { + assert_eq!( + dv.resolve(&super::IndexFileLocation::BucketLocal { + table_path: "file:/tmp/table", + bucket_path: "file:/tmp/table/bucket-0", + index_file_in_data_file_dir, + }) + .path(), + "s3://other/dv/index-file" + ); + } + } + + #[test] + fn test_build_deletion_files_map_carries_external_path() { + let entries = vec![IndexManifestEntry { + version: 1, + kind: FileKind::Add, + partition: vec![9], + bucket: 2, + index_file: IndexFileMeta { + index_type: "DELETION_VECTORS".into(), + file_name: "index-file".into(), + file_size: 128, + row_count: 1, + deletion_vectors_ranges: Some(indexmap::IndexMap::from([( + "data-file.parquet".into(), + DeletionVectorMeta { + offset: 1, + length: 2, + cardinality: None, + }, + )])), + external_path: Some("s3://other/dv/index-file".into()), + global_index_meta: None, + }, + }]; + + let map = super::build_deletion_files_map(&entries); + let dv = map + .get(&super::PartitionBucket::new(vec![9], 2)) + .expect("partition bucket should exist") .get("data-file.parquet") .expect("deletion file should exist"); assert_eq!( - deletion_file, - &DeletionFile::new("file:/tmp/table/index/index-file".into(), 11, 22, Some(33)) + dv.external_path.as_deref(), + Some("s3://other/dv/index-file") ); } @@ -4266,6 +4406,7 @@ mod tests { file_size: 1, row_count: 1, deletion_vectors_ranges: None, + external_path: None, global_index_meta: None, }, }; @@ -4327,6 +4468,7 @@ mod tests { file_size: 1, row_count: 1, deletion_vectors_ranges: None, + external_path: None, global_index_meta: (index_type == "btree").then_some(GlobalIndexMeta { row_range_start: 0, row_range_end: 0, @@ -4385,6 +4527,7 @@ mod tests { file_size: 1, row_count: 1, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: 0, row_range_end: 0, diff --git a/crates/paimon/src/table/table_write.rs b/crates/paimon/src/table/table_write.rs index 1d13f3b3..c37df7b6 100644 --- a/crates/paimon/src/table/table_write.rs +++ b/crates/paimon/src/table/table_write.rs @@ -329,7 +329,7 @@ impl TableWrite { merge_engine, ))) } else if is_dynamic_bucket { - BucketAssignerEnum::Dynamic(DynamicBucketAssigner::new( + BucketAssignerEnum::Dynamic(Box::new(DynamicBucketAssigner::new( partition_field_indices, primary_key_indices.clone(), schema.fields().to_vec(), @@ -337,7 +337,12 @@ impl TableWrite { table.file_io().clone(), table.location().to_string(), is_overwrite, - )) + // The same computer this writer already built: a hash index kept in + // the data-file directory must land in the directory the writer and + // the reader both derive, so both must agree on partition naming. + partition_computer.clone(), + core_options.index_file_in_data_file_dir(), + ))) } else if total_buckets == POSTPONE_BUCKET { BucketAssignerEnum::Constant(ConstantBucketAssigner::new( partition_field_indices, @@ -830,11 +835,7 @@ impl TableWrite { // Collect index files from bucket assigner let file_io = self.table.file_io(); - let index_dir = format!("{}/index", self.table.location()); - let mut index_files_by_key = self - .bucket_assigner - .prepare_commit_index(file_io, &index_dir) - .await?; + let mut index_files_by_key = self.bucket_assigner.prepare_commit_index(file_io).await?; let mut messages = Vec::new(); for (partition_bytes, bucket, files) in results { diff --git a/crates/paimon/src/table/vector_search_builder.rs b/crates/paimon/src/table/vector_search_builder.rs index aa1fdc6d..eb4e0a42 100644 --- a/crates/paimon/src/table/vector_search_builder.rs +++ b/crates/paimon/src/table/vector_search_builder.rs @@ -32,6 +32,7 @@ use crate::table::global_index_scanner::{ deleted_row_ranges_for_data_evolution_dvs, search_limit_with_deleted_rows, unindexed_ranges_for_global_index_entries, RowRangeIndex, }; +use crate::table::index_file_path::IndexFileLocation; use crate::table::pk_vector_data_file_reader::{ append_batch_vectors, DataFilePkVectorReaderFactory, }; @@ -76,7 +77,6 @@ use std::io::Cursor; use std::sync::Arc; use std::time::{Duration, Instant}; -const INDEX_DIR: &str = "index"; const RAW_SCORE_MATRIX_MIN_QUERY_COUNT: usize = 4; const RAW_SCORE_MATRIX_TARGET_ELEMENTS: usize = 1 << 20; const RAW_TOP_K_MIN_PARTITION_SIZE: usize = 1 << 12; @@ -1675,7 +1675,8 @@ async fn evaluate_batch_vector_search( let global_meta = entry.index_file.global_index_meta.as_ref().unwrap(); let backend = VectorIndexBackend::from_index_type(&entry.index_file.index_type) .expect("filtered vector index type"); - let path = format!("{table_path}/{INDEX_DIR}/{}", entry.index_file.file_name); + let path = IndexFileLocation::Global { table_path } + .resolve(&entry.index_file.file_name, entry.index_file.external_path.as_deref()); let file_name = entry.index_file.file_name.clone(); let file_size = entry.index_file.file_size as u64; let index_meta_bytes = global_meta.index_meta.clone().unwrap_or_default(); @@ -2835,7 +2836,10 @@ async fn resolve_raw_vector_metric( } } } - let path = format!("{table_path}/{INDEX_DIR}/{}", entry.index_file.file_name); + let path = IndexFileLocation::Global { table_path }.resolve( + &entry.index_file.file_name, + entry.index_file.external_path.as_deref(), + ); let input = file_io.new_input(&path)?; let read_error = |e| crate::Error::DataInvalid { message: format!( @@ -4320,6 +4324,7 @@ mod tests { file_size: 100, row_count: 10, deletion_vectors_ranges: None, + external_path: None, global_index_meta: None, }, version: 1, @@ -5703,6 +5708,7 @@ mod tests { file_size: i64::try_from(index_file_size).unwrap(), row_count, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: 0, row_range_end: row_count - 1, @@ -6233,6 +6239,7 @@ mod tests { file_size: 100, row_count: 10, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: 0, row_range_end: 9, diff --git a/crates/paimon/src/table/vindex_index_build_builder.rs b/crates/paimon/src/table/vindex_index_build_builder.rs index b0680fe7..54455e95 100644 --- a/crates/paimon/src/table/vindex_index_build_builder.rs +++ b/crates/paimon/src/table/vindex_index_build_builder.rs @@ -646,6 +646,7 @@ impl<'a> VindexIndexBuildBuilder<'a> { )?, row_count, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: shard.row_range_start, row_range_end: shard.row_range_end, @@ -1625,6 +1626,7 @@ mod tests { file_size: 1, row_count: end - start + 1, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: start, row_range_end: end, @@ -1896,6 +1898,7 @@ mod tests { file_size: 1, row_count: (coverage[0].to() - coverage[0].from() + 1) as i64, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: coverage[0].from(), row_range_end: coverage[0].to(), diff --git a/crates/paimon/tests/pk_vector_baseline_test.rs b/crates/paimon/tests/pk_vector_baseline_test.rs index a5c4e90b..8b24b2b5 100644 --- a/crates/paimon/tests/pk_vector_baseline_test.rs +++ b/crates/paimon/tests/pk_vector_baseline_test.rs @@ -426,6 +426,7 @@ async fn build_table_with_first_row_id( file_size: i64::try_from(index_file_size).unwrap(), row_count, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: 0, row_range_end: row_count - 1, @@ -1273,6 +1274,7 @@ async fn pk_vector_refine_factor_matches_exact_ground_truth() { file_size: i64::try_from(index_file_size).unwrap(), row_count, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: 0, row_range_end: row_count - 1, diff --git a/crates/paimon/tests/pk_vector_batch_test.rs b/crates/paimon/tests/pk_vector_batch_test.rs index 87e4ca59..33c8c7a9 100644 --- a/crates/paimon/tests/pk_vector_batch_test.rs +++ b/crates/paimon/tests/pk_vector_batch_test.rs @@ -287,6 +287,7 @@ async fn build_table(vectors: &[[f32; DIM]]) -> (tempfile::TempDir, Table) { file_size: i64::try_from(index_file_size).unwrap(), row_count, deletion_vectors_ranges: None, + external_path: None, global_index_meta: Some(GlobalIndexMeta { row_range_start: 0, row_range_end: row_count - 1, diff --git a/docs/src/sql.md b/docs/src/sql.md index 34a033e9..a00951ef 100644 --- a/docs/src/sql.md +++ b/docs/src/sql.md @@ -1986,7 +1986,7 @@ Columns: | `partition` | STRING | Partition spec for the indexed data, formatted as a Java row cast string; `{}` for unpartitioned tables | | `bucket` | INT | Bucket id covered by the index file | | `index_type` | STRING | Index type, such as `btree`, `bitmap`, `multivalue`, `ivf-flat`, `lumina`, or `DELETION_VECTORS` | -| `file_name` | STRING | Index file name under the table index directory | +| `file_name` | STRING | Index file name. It resolves to the table `index/` directory, or to the bucket's data-file directory when `index-file-in-data-file-dir` is set; an index file with an external path is read from that path instead | | `file_size` | BIGINT | Index file size in bytes | | `row_count` | BIGINT | Number of rows covered by the index file | | `dv_ranges` | ARRAY | Deletion-vector ranges, only populated for deletion-vector metadata | @@ -2003,7 +2003,8 @@ Files are classified by their table-relative path: - `manifest/manifest-*`, `manifest/manifest-list-*`, and `manifest/index-manifest-*` → manifest - `statistics/*` → manifest file counters for the current compatible output schema - `index/*` → index -- `/bucket-*/*` and `/bucket-postpone/*` → data, using the table's partition depth, except names starting with `index-` +- `/bucket-*/index-*` and `/bucket-postpone/index-*` → index, where `index-file-in-data-file-dir` puts them; classification follows the file's physical form, not the current option value +- `/bucket-*/*` and `/bucket-postpone/*` → data, using the table's partition depth - unknown files are ignored by this summary ```sql