From af2167ba00f821956e4ece000a57832028462b3c Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Tue, 28 Jul 2026 11:31:38 +0800 Subject: [PATCH 1/3] fix(table): resolve index files by external path and bucket layout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Index files were always read from `/index/`, ignoring both the `_EXTERNAL_PATH` recorded in the index manifest and the `index-file-in-data-file-dir` table option. A table that keeps index files beside its bucket's data files fails to read them, e.g. a primary-key vector search reports failed to open ANN index file '
/index/index--0' for range reads while the file actually lives in the bucket directory. Decode `_EXTERNAL_PATH` from the index manifest (Java `IndexFileMeta` SCHEMA field 5) and add it to the write schema so a rewritten manifest keeps it — it is currently dropped silently — add the `index-file-in-data-file-dir` option, and resolve every index file through one place (`table/index_file_path.rs`) with two modes: * global, always `
/index`: the data-evolution global index, and vector and full-text search over it; * bucket-local, the data-file directory when the option is set: primary-key vector ANN segments, primary-key full-text archives, deletion vectors, and the dynamic-bucket hash index. Each mode mirrors the factory Java uses for that consumer: `DataEvolutionGlobalIndexScanner` resolves through `globalIndexFileFactory`, while `IndexFileHandler` resolves hash, deletion-vector and primary-key vector files through `pathFactories.get(partition, bucket)`, which selects `IndexInDataFileDirPathFactory` when the option is set. An explicit external path wins over both layouts, as in `toPath(IndexFileMeta)`. For the two index kinds this crate writes itself, deletion vectors and the hash index, reads and writes move together, so a file written here is found again: * the data-evolution writer resolves an existing deletion vector through the same path when merging, and writes a new one where the reader will look; * the dynamic-bucket assigner resolves per partition and bucket for both restore and commit. `BucketAssigner::prepare_commit_index` no longer takes an index directory — three of its four implementations ignored it, and the fourth now derives the layout itself; * `TableCommit::abort` deletes a newly written index file where it was written, mirroring Java `FileStoreCommitImpl.abort`, which deletes through `indexFileFactory(partition, bucket)`. Deleting is best-effort, so the old fixed path leaked the file silently instead of failing. A bucket directory comes from the split that references the file when a split is at hand, and otherwise from the partition and bucket being committed. Both go through one `spec::bucket_path`, mirroring Java `FileStorePathFactory.bucketPath`: the layout is only correct while every producer and consumer of a bucket directory agrees byte for byte, and nothing else enforces that. The option is immutable, as in Java, where it is annotated `@Immutable` and `SchemaManager.checkAlterTableOption` rejects altering it: it selects the directory index files are written to, so flipping it on a populated table would hide every index file already written. `$physical_files_size` now counts an `index-` prefixed file in a bucket directory as an index file rather than dropping it, matching Java `FileType.classify`, which maps any `index-*` basename to `BUCKET_INDEX` regardless of directory. Classification follows the file's physical form, not the current option value, so a file stays recognizable after the setting it was written under changes. The BTree reader cache is keyed by the resolved path so two entries sharing a file name cannot reuse each other's reader. `data-file.path-directory` remains unsupported, as it is throughout this crate: bucket paths are rooted directly at the table for data files as much as for index files, so honoring it belongs with data-file path handling rather than here. --- bindings/c/src/tests.rs | 1 + .../src/system_tables/table_indexes.rs | 1 + .../datafusion/tests/read_tables.rs | 1 + crates/paimon/src/catalog/filesystem.rs | 81 ++++++- crates/paimon/src/spec/avro/decode_helpers.rs | 14 +- .../spec/avro/index_manifest_entry_decode.rs | 7 +- crates/paimon/src/spec/core_options.rs | 11 + crates/paimon/src/spec/index_file_meta.rs | 9 + crates/paimon/src/spec/index_manifest.rs | 126 ++++++++++- crates/paimon/src/spec/mod.rs | 2 +- crates/paimon/src/spec/partition_utils.rs | 36 ++- crates/paimon/src/table/bucket_assigner.rs | 12 +- .../src/table/bucket_assigner_constant.rs | 1 - .../paimon/src/table/bucket_assigner_cross.rs | 1 - .../src/table/bucket_assigner_dynamic.rs | 180 ++++++++++++++- .../paimon/src/table/bucket_assigner_fixed.rs | 1 - .../paimon/src/table/data_evolution_writer.rs | 89 +++++++- .../src/table/full_text_search_builder.rs | 14 +- .../src/table/global_index_drop_builder.rs | 3 + .../paimon/src/table/global_index_scanner.rs | 74 +++++-- .../paimon/src/table/hybrid_search_builder.rs | 3 + crates/paimon/src/table/index_file_path.rs | 177 +++++++++++++++ .../src/table/lumina_index_build_builder.rs | 2 + crates/paimon/src/table/mod.rs | 1 + .../src/table/pk_full_text_bucket_search.rs | 58 ++++- .../src/table/pk_full_text_bucket_state.rs | 1 + crates/paimon/src/table/pk_full_text_read.rs | 20 +- crates/paimon/src/table/pk_full_text_scan.rs | 1 + crates/paimon/src/table/pk_vector_scan.rs | 208 +++++++++++++----- crates/paimon/src/table/referenced_files.rs | 45 +++- .../sorted_global_index_build_builder.rs | 2 + crates/paimon/src/table/table_commit.rs | 133 +++++++++-- crates/paimon/src/table/table_scan.rs | 203 ++++++++++++++--- crates/paimon/src/table/table_write.rs | 15 +- .../paimon/src/table/vector_search_builder.rs | 13 +- .../src/table/vindex_index_build_builder.rs | 3 + .../paimon/tests/pk_vector_baseline_test.rs | 2 + crates/paimon/tests/pk_vector_batch_test.rs | 1 + docs/src/sql.md | 5 +- 39 files changed, 1364 insertions(+), 193 deletions(-) create mode 100644 crates/paimon/src/table/index_file_path.rs diff --git a/bindings/c/src/tests.rs b/bindings/c/src/tests.rs index 37fa0c919..db8d52d5c 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 cff4796ad..2533b8280 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 e4fc47211..8d770d35f 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 45628ad66..681bc3a69 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 b36aec155..38165329b 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 38bf0c717..50cfece1c 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 b17d31685..7afff6f67 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 3b1b2f560..7af984127 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 24cac0f2c..3cb3ead4b 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 a2613fecf..a165ffe07 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 3db1884c0..e4245c312 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 a6291e4d2..64abf810f 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 5d0f2a4ee..843d2464a 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 5a1d33988..4ae181574 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 030385d94..57fbc48e4 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 428b07dda..684a98a84 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 1736626c1..d28553485 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 81beacc20..fac4db4a6 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 505139fe6..170e7097b 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 b4360426e..05eb460a5 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 5068dc802..af6dbb8c6 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 000000000..26d71f6a4 --- /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 d340c97eb..531ba8079 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 35f1e45a0..720f157f4 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 ecb2ba232..a92395cf6 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 c32cf6e4f..18c411557 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 d5eed2fca..2eadc2e85 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 2ffe0b49a..350b17dc2 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 cfc528b25..8521aad6c 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 851e87198..d4bb617bb 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 de72364b7..5f44e39e3 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 6e5d78c39..82e5b8f5c 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 601eeae90..929a9a752 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 1d13f3b35..c37df7b66 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 aa1fdc6dc..eb4e0a421 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 b0680fe73..54455e95b 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 a5c4e90b5..8b24b2b56 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 87e4ca59d..33c8c7a9f 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 34a033e96..a00951efa 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 From 88f1df15e3fbecc02c07aae69a2072d2b11ffb00 Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Mon, 31 Aug 2026 02:24:44 +0800 Subject: [PATCH 2/3] fix(table): resolve a committed index file by its own layout on abort MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `abort` resolved every index file in a commit message as bucket-local, so with `index-file-in-data-file-dir` set it looked for a data-evolution global index file beside the bucket's data files while the file sits under `
/index`. Deleting is best-effort, so a failed commit leaked it silently. Java `FileStoreCommitImpl.abort` has the same gap: it resolves every `newIndexFiles()` entry through `indexFileFactory(partition, bucket)`, while `SortedGlobalIndexWriter.flushIndex` writes through `globalIndexFileFactory()` and returns those metas in `DataIncrement.indexIncrement(...)`. Which layout a file was written under is a property of the file, not of the message carrying it. A deletion vector or the dynamic-bucket hash index carries no `_GLOBAL_INDEX` at all. A global index file carries one whose `_SOURCE_META` is either absent — what this crate's index builders write — or marked with `DataEvolutionIndexSourceMeta`'s magic, which Java added for exactly this question: "the marker distinguishes this metadata from primary-key index source metadata", and `PrimaryKeyIndexSourceMeta` starts with its own version instead. Anything else stays bucket-local, which is what Java assumes for every index file. Both abort tests now plant a same-named file in the other layout and assert it survives, so neither direction can regress into deleting a path the commit does not own. The frames the classification tests use are the ones Java serializes, and the primary-key frame is fed through `PrimaryKeyIndexSourceMeta::deserialize`, so a frame the classifier sends bucket-local is one a reader accepts. --- crates/paimon/src/table/index_file_path.rs | 194 +++++++++++++++++++++ crates/paimon/src/table/table_commit.rs | 102 +++++++++-- 2 files changed, 284 insertions(+), 12 deletions(-) diff --git a/crates/paimon/src/table/index_file_path.rs b/crates/paimon/src/table/index_file_path.rs index 26d71f6a4..b2766b478 100644 --- a/crates/paimon/src/table/index_file_path.rs +++ b/crates/paimon/src/table/index_file_path.rs @@ -29,6 +29,12 @@ //! 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. +//! +//! A reader knows which of these it is asking for, because the mode follows the +//! index kind it reads. Cleanup after a failed commit does not — see +//! [`committed_index_file_path`]. + +use crate::spec::IndexFileMeta; const INDEX_DIR: &str = "index"; @@ -82,9 +88,62 @@ impl IndexFileLocation<'_> { } } +/// `"DEIX"` as a big-endian int, Java `DataEvolutionIndexSourceMeta`'s marker. +/// `PrimaryKeyIndexSourceMeta` starts with its own version instead, so the marker +/// tells the two apart — which is what Java added it for. +const DATA_EVOLUTION_SOURCE_META_MAGIC: &[u8; 4] = b"DEIX"; + +/// Whether an index file carried by a commit message is a global index file. +/// +/// A `_GLOBAL_INDEX` on its own does not say — a source-backed primary-key index +/// carries one too. The source metadata does: Java marks its own with +/// `DataEvolutionIndexSourceMeta`'s magic, added for exactly this question, while +/// `PrimaryKeyIndexSourceMeta` starts with its own version. A deletion vector or +/// the dynamic-bucket hash index carries no `_GLOBAL_INDEX` at all. +/// +/// Absent source metadata is global because the index builders in this crate +/// write it that way, and they are the only producers of the global index files +/// it commits. Java records that predate `_SOURCE_META` cannot reach here: a +/// commit message is built by this crate's writers, never decoded from Java. +fn is_global_index_file(file: &IndexFileMeta) -> bool { + file.global_index_meta + .as_ref() + .is_some_and(|meta| match meta.source_meta.as_deref() { + None => true, + Some(source_meta) => source_meta.starts_with(DATA_EVOLUTION_SOURCE_META_MAGIC), + }) +} + +/// Where an index file carried by a commit message was written. +/// +/// Cleanup after a failed commit resolves index files this crate just wrote, and +/// they are not all one layout: a data-evolution index build is global, while a +/// deletion vector or the dynamic-bucket hash index is bucket-local. Each file +/// is therefore classified on its own rather than the layout being taken from +/// the message. Anything not recognizable as a global index file is bucket-local, +/// which is what Java `FileStoreCommitImpl.abort` assumes for every index file. +pub(crate) fn committed_index_file_path( + table_path: &str, + bucket_path: &str, + index_file_in_data_file_dir: bool, + file: &IndexFileMeta, +) -> String { + let location = if is_global_index_file(file) { + IndexFileLocation::Global { table_path } + } else { + IndexFileLocation::BucketLocal { + table_path, + bucket_path, + index_file_in_data_file_dir, + } + }; + location.resolve(&file.file_name, file.external_path.as_deref()) +} + #[cfg(test)] mod tests { use super::*; + use crate::spec::{GlobalIndexMeta, PrimaryKeyIndexSourceMeta}; #[test] fn external_path_wins_over_every_mode() { @@ -174,4 +233,139 @@ mod tests { ); } } + + /// A committed index file with the given `_GLOBAL_INDEX` and `_SOURCE_META`. + fn committed_file(global_index_meta: Option>>) -> IndexFileMeta { + IndexFileMeta { + index_type: "btree".to_string(), + file_name: "idx-0".to_string(), + file_size: 128, + row_count: 1, + deletion_vectors_ranges: None, + external_path: None, + global_index_meta: global_index_meta.map(|source_meta| GlobalIndexMeta { + row_range_start: 0, + row_range_end: 0, + index_field_id: 0, + extra_field_ids: None, + index_meta: None, + source_meta, + }), + } + } + + fn committed_path(file: &IndexFileMeta) -> String { + committed_index_file_path( + "warehouse/db/tbl", + "warehouse/db/tbl/pt=1/bucket-3", + true, + file, + ) + } + + /// Java `DataEvolutionIndexSourceMeta.serialize`: magic, version, scan + /// snapshot id. + fn data_evolution_source_meta(scan_snapshot_id: i64) -> Vec { + let mut bytes = b"DEIX".to_vec(); + bytes.extend_from_slice(&1i32.to_be_bytes()); + bytes.extend_from_slice(&scan_snapshot_id.to_be_bytes()); + bytes + } + + /// Java `PrimaryKeyIndexSourceMeta.serialize`: version, data level, source + /// count, then each source's `writeUTF` name and row count. + fn primary_key_source_meta(data_level: i32, source_name: &str, row_count: i64) -> Vec { + let mut bytes = 1i32.to_be_bytes().to_vec(); + bytes.extend_from_slice(&data_level.to_be_bytes()); + bytes.extend_from_slice(&1i32.to_be_bytes()); + bytes.extend_from_slice(&(source_name.len() as u16).to_be_bytes()); + bytes.extend_from_slice(source_name.as_bytes()); + bytes.extend_from_slice(&row_count.to_be_bytes()); + // A frame the classifier rejects has to be one a reader would accept, + // otherwise the test only proves that garbage is not global. + PrimaryKeyIndexSourceMeta::deserialize(&bytes).expect("a valid primary-key source frame"); + bytes + } + + #[test] + fn a_global_index_file_is_committed_under_the_table_index_directory() { + // This crate's index builders leave `_SOURCE_META` empty, and Java marks + // its own with `DataEvolutionIndexSourceMeta`'s `DEIX`. Both are global + // even when the table keeps index files in the data-file directory. + for source_meta in [None, Some(data_evolution_source_meta(7))] { + let file = committed_file(Some(source_meta)); + assert_eq!(committed_path(&file), "warehouse/db/tbl/index/idx-0"); + } + } + + #[test] + fn a_source_backed_primary_key_index_file_is_committed_bucket_local() { + // Primary-key source metadata starts with its version, never `DEIX`. + let file = committed_file(Some(Some(primary_key_source_meta(1, "data-0.parquet", 3)))); + assert_eq!( + committed_path(&file), + "warehouse/db/tbl/pt=1/bucket-3/idx-0" + ); + } + + #[test] + fn only_a_whole_data_evolution_marker_makes_source_metadata_global() { + // The marker is the whole four bytes, as in Java's length-checked + // big-endian comparison: a shorter or partial prefix is not it. Java's own + // marker test likewise looks no further than those four bytes, so a + // truncated frame that still carries them stays global. + for not_global in [vec![], b"D".to_vec(), b"DEI".to_vec(), b"XIED".to_vec()] { + let file = committed_file(Some(Some(not_global.clone()))); + assert_eq!( + committed_path(&file), + "warehouse/db/tbl/pt=1/bucket-3/idx-0", + "{not_global:?} does not carry the marker" + ); + } + let file = committed_file(Some(Some(b"DEIX".to_vec()))); + assert_eq!(committed_path(&file), "warehouse/db/tbl/index/idx-0"); + } + + #[test] + fn an_index_file_without_global_index_meta_is_committed_bucket_local() { + // Deletion vectors and the dynamic-bucket hash index carry no + // `_GLOBAL_INDEX` at all. + let file = committed_file(None); + assert_eq!( + committed_path(&file), + "warehouse/db/tbl/pt=1/bucket-3/idx-0" + ); + } + + #[test] + fn a_committed_index_file_keeps_its_external_path_in_either_layout() { + let external = "s3://other-bucket/abs/idx-0"; + for global_index_meta in [None, Some(None)] { + let mut file = committed_file(global_index_meta); + file.external_path = Some(external.to_string()); + assert_eq!(committed_path(&file), external); + } + } + + #[test] + fn committed_index_files_share_one_directory_without_the_bucket_dir_option() { + // Without the option every layout resolves under the table `index/` + // directory, so classification cannot change the outcome. + for global_index_meta in [ + None, + Some(None), + Some(Some(primary_key_source_meta(1, "data-0.parquet", 3))), + ] { + let file = committed_file(global_index_meta); + assert_eq!( + committed_index_file_path( + "warehouse/db/tbl", + "warehouse/db/tbl/pt=1/bucket-3", + false, + &file, + ), + "warehouse/db/tbl/index/idx-0" + ); + } + } } diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index 82e5b8f5c..8ce158c08 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -32,7 +32,7 @@ use crate::spec::{ }; 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::index_file_path::committed_index_file_path; use crate::table::partition_filter::PartitionFilter; use crate::table::snapshot_commit::SnapshotCommit; use crate::table::{SnapshotManager, Table, TableScan}; @@ -720,18 +720,24 @@ impl TableCommit { let _ = self.table.file_io().delete_file(&path).await; } } - // 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, - }; + // An index file must be deleted where it was written: at its external + // path when it has one, under the table `index/` directory when it is a + // global index file, and beside this bucket's data files when the table + // keeps bucket-local index files there. Which of the last two applies + // is a property of the file, not of the message — a data-evolution + // index build is global while a deletion vector from the same table is + // bucket-local — so each file is classified rather than all of them + // assumed bucket-local, as Java `FileStoreCommitImpl.abort` does + // through `indexFileFactory(partition, bucket)`. Deleting is + // best-effort, so a wrong path leaks the file silently instead of + // failing. for file in &message.new_index_files { - let path = index_location.resolve(&file.file_name, file.external_path.as_deref()); + let path = committed_index_file_path( + table_path, + &bucket_path, + index_file_in_data_file_dir, + file, + ); let _ = self.table.file_io().delete_file(&path).await; } } @@ -5539,6 +5545,18 @@ mod tests { .await .unwrap(); + // A same-named file in the other layout belongs to someone else. Abort + // resolves one path, so this one must survive. + let index_dir = format!("{table_path}/index"); + let sentinel = format!("{index_dir}/index-in-bucket"); + file_io.mkdirs(&format!("{index_dir}/")).await.unwrap(); + file_io + .new_output(&sentinel) + .unwrap() + .write(bytes::Bytes::from_static(b"other")) + .await + .unwrap(); + let mut message = CommitMessage::new(vec![], 0, vec![]); message.new_index_files = vec![IndexFileMeta { index_type: "HASH".to_string(), @@ -5555,6 +5573,66 @@ mod tests { !file_io.exists(&index_path).await.unwrap(), "abort must remove an index file written into the bucket data-file directory" ); + assert!( + file_io.exists(&sentinel).await.unwrap(), + "abort must not touch a same-named file in the table index directory" + ); + } + + #[tokio::test] + async fn test_abort_deletes_a_global_index_file_when_index_files_live_in_the_bucket_dir() { + // A data-evolution index build writes a global index file under + // `
/index` even when `index-file-in-data-file-dir` is set, so + // resolving it as bucket-local leaves it behind — deleting is best-effort + // (`let _ =`), so the wrong path fails silently. + let file_io = test_file_io(); + let table_path = "memory:/test_abort_global_index_with_bucket_dir_option"; + 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 index_dir = format!("{table_path}/index"); + let index_path = format!("{index_dir}/index-global-0"); + file_io.mkdirs(&format!("{index_dir}/")).await.unwrap(); + file_io + .new_output(&index_path) + .unwrap() + .write(bytes::Bytes::from_static(b"index")) + .await + .unwrap(); + + // A same-named file in the other layout belongs to someone else. Abort + // resolves one path, so this one must survive. + let bucket_dir = format!("{table_path}/bucket-0"); + let sentinel = format!("{bucket_dir}/index-global-0"); + file_io.mkdirs(&format!("{bucket_dir}/")).await.unwrap(); + file_io + .new_output(&sentinel) + .unwrap() + .write(bytes::Bytes::from_static(b"other")) + .await + .unwrap(); + + let mut message = CommitMessage::new(vec![], 0, vec![]); + message.new_index_files = vec![test_global_index_file("index-global-0", 0, 0, 4)]; + commit.abort(&[message]).await.unwrap(); + + assert!( + !file_io.exists(&index_path).await.unwrap(), + "abort must remove a global index file from the table index directory" + ); + assert!( + file_io.exists(&sentinel).await.unwrap(), + "abort must not touch a same-named file in the bucket data-file directory" + ); } #[tokio::test] From 3df9e15b31954088a0cf16b67ac9ada16ff6c865 Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Mon, 31 Aug 2026 20:02:31 +0800 Subject: [PATCH 3/3] fix(spec): pin the index layout against dynamic overrides, and gate its alter on snapshots MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two holes in making `index-file-in-data-file-dir` live. `Table::copy_with_options` still merged it as a dynamic override, and every resolver reads the copy's schema — the writers included. A table persisted with one value and copied with the other wrote hash and deletion-vector index files according to the copy, while committing manifest entries that record only a file name, so a normally loaded table could not resolve them. `TableSchema::copy_with_options` now pins the option to the stored value, as it already does for `type`, and drops the override entirely when nothing is stored so the default stands. Java rejects such an override outright (`AbstractFileStoreTable.checkImmutability`), but this copy cannot fail. The alter guard rejected every set and remove, including before the first write and including a `SetOption` that repeats the stored value. Java reaches `checkAlterTableOption` only under `hasSnapshots && !unchanged`, so a caller can choose the layout through ALTER before anything is written and schema reconciliation stays idempotent. The guard now takes that flag, and `alter_table` resolves it only when a change touches the option, mirroring Java's `LazyField` rather than paying a snapshot lookup on every alter. `type` stays unconditional, as it is in Java: a format table holds data without ever writing a snapshot. Setting the option to `false` where it was never stored is a change, not a no-op — Java compares the stored string and normalizes only `type`, `primary-key` and `partition` — so it is rejected once snapshots exist even though it names the layout already in use. Gating on snapshot existence leaves one window open, in Java as much as here, and the guard's documentation now says so: a writer that started before the alter still holds the old layout, and a commit records the schema id it was built with without checking whether the latest schema moved. Closing it needs schema publication and snapshot commit ordered against each other, not a stricter alter. `test_table_with_options` built its schema through `copy_with_options`, which the pin now filters, so it persists its options instead — the way a catalog-loaded table carries them. --- crates/paimon/src/catalog/filesystem.rs | 223 +++++++++++++++++++++--- crates/paimon/src/spec/schema.rs | 80 ++++++++- crates/paimon/src/table/mod.rs | 57 ++++++ crates/paimon/src/table/table_commit.rs | 14 +- crates/paimon/src/table/table_write.rs | 62 +++++++ 5 files changed, 413 insertions(+), 23 deletions(-) diff --git a/crates/paimon/src/catalog/filesystem.rs b/crates/paimon/src/catalog/filesystem.rs index 681bc3a69..b8073c363 100644 --- a/crates/paimon/src/catalog/filesystem.rs +++ b/crates/paimon/src/catalog/filesystem.rs @@ -30,7 +30,7 @@ use crate::spec::{ CoreOptions, Schema, TableSchema, TableType, INDEX_FILE_IN_DATA_FILE_DIR_OPTION, TABLE_TYPE_OPTION, }; -use crate::table::{ObjectTable, SchemaManager, Table}; +use crate::table::{ObjectTable, SchemaManager, SnapshotManager, Table}; use async_trait::async_trait; use bytes::Bytes; use opendal::raw::get_basename; @@ -542,7 +542,26 @@ impl Catalog for FileSystemCatalog { full_name: identifier.full_name(), })?; - reject_immutable_option_changes(current.options(), &changes)?; + // Only the index-layout option is snapshot-gated, so the listing a + // snapshot lookup costs is paid only when a change touches it. Java defers + // the same lookup through a `LazyField`. + let touches_snapshot_gated_option = changes.iter().any(|change| { + matches!( + change, + crate::spec::SchemaChange::SetOption { key, .. } + | crate::spec::SchemaChange::RemoveOption { key } + if key == INDEX_FILE_IN_DATA_FILE_DIR_OPTION + ) + }); + let has_snapshots = if touches_snapshot_gated_option { + SnapshotManager::new(self.file_io.clone(), table_path.clone()) + .get_latest_snapshot_id() + .await? + .is_some() + } else { + false + }; + reject_immutable_option_changes(current.options(), &changes, has_snapshots)?; let new_schema = current .apply_changes(changes) @@ -559,13 +578,31 @@ impl Catalog for FileSystemCatalog { /// 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. +/// reader that cannot see its data. Only case-insensitive no-ops pass, and a +/// snapshot is not required: a format table holds data without ever writing +/// one, so a snapshot check would not catch it. Java special-cases `type` the +/// same way (`SchemaManager.generateTableSchema`). /// * `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. +/// table already has. Only files already written are at stake, so this follows +/// Java in rejecting an actual change only once the table has snapshots, and in +/// letting a `SetOption` that repeats the stored value through: Java compares +/// the stored string literally, and reaches `checkAlterTableOption` only under +/// `hasSnapshots && !unchanged`. +/// +/// Gating on snapshot existence leaves one window open, in Java as much as here: a +/// writer that started before the alter still holds the old layout, and a commit +/// records the schema id it was built with without checking whether the latest +/// schema moved, so it can publish the first snapshot after the layout was flipped +/// under it. Closing it needs schema publication and snapshot commit to be ordered +/// against each other — reading the latest schema once before committing is not +/// enough, since the alter can land right after that read. A stricter alter is not +/// the answer either: rejecting the change outright would take away the only +/// post-creation point at which the layout can be chosen. fn reject_immutable_option_changes( current_options: &HashMap, changes: &[crate::spec::SchemaChange], + has_snapshots: bool, ) -> Result<()> { let current_type = current_options .get(TABLE_TYPE_OPTION) @@ -587,18 +624,34 @@ fn reject_immutable_option_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 => + crate::spec::SchemaChange::SetOption { key, value } + if key == INDEX_FILE_IN_DATA_FILE_DIR_OPTION + && has_snapshots + && current_options.get(key.as_str()) != Some(value) => { return Err(Error::Unsupported { message: format!( - "changing '{INDEX_FILE_IN_DATA_FILE_DIR_OPTION}' is not supported: \ + "changing '{INDEX_FILE_IN_DATA_FILE_DIR_OPTION}' on a table with \ + snapshots is not supported: \ it selects the directory index files are written to, so the files \ already written would no longer be found" ), }); } + crate::spec::SchemaChange::RemoveOption { key } + if key == INDEX_FILE_IN_DATA_FILE_DIR_OPTION && has_snapshots => + { + // Java rejects the reset whenever the table has snapshots, whether + // or not the option is currently set + // (`SchemaManager.checkResetTableOption`). + return Err(Error::Unsupported { + message: format!( + "removing '{INDEX_FILE_IN_DATA_FILE_DIR_OPTION}' from a table with \ + snapshots is not supported: it selects the directory index files \ + are written to, so the files already written would no longer be found" + ), + }); + } _ => {} } } @@ -852,17 +905,36 @@ 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; + /// A table whose only snapshot exists so the immutable-option guard sees one. + async fn give_the_table_a_snapshot(catalog: &FileSystemCatalog, identifier: &Identifier) { + use crate::spec::{CommitKind, Snapshot}; + let table_path = catalog.table_path(identifier); + let snapshot = Snapshot::builder() + .version(3) + .id(1) + .schema_id(0) + .base_manifest_list("base-list".to_string()) + .delta_manifest_list("delta-list".to_string()) + .commit_user("test-user".to_string()) + .commit_identifier(0) + .commit_kind(CommitKind::APPEND) + .time_millis(1000) + .build(); + assert!( + SnapshotManager::new(catalog.file_io.clone(), table_path) + .commit_snapshot(&snapshot) + .await + .unwrap(), + "the fixture snapshot must be the table's first" + ); + } - // 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(); + async fn create_table_for_alter( + catalog: &FileSystemCatalog, + options: HashMap, + ) -> Identifier { catalog - .create_database("db1", false, HashMap::new()) + .create_database("db1", true, HashMap::new()) .await .unwrap(); let schema = Schema::builder() @@ -870,6 +942,7 @@ mod tests { "id", crate::spec::DataType::Int(crate::spec::IntType::new()), ) + .options(options) .build() .unwrap(); let identifier = Identifier::new("db1", "t"); @@ -877,12 +950,29 @@ mod tests { .create_table(&identifier, schema, false) .await .unwrap(); + identifier + } - for change in [ - SchemaChange::set_option( + #[tokio::test] + async fn test_alter_table_cannot_change_where_index_files_live_once_written() { + use crate::spec::SchemaChange; + + // The option selects the directory every bucket-local index file is written + // to and read from, so flipping it once files exist would hide every one of + // them. Only files already written are at stake, so the guard follows Java + // in gating on snapshot existence rather than rejecting outright. + let (_temp_dir, catalog) = create_test_catalog(); + let identifier = create_table_for_alter( + &catalog, + HashMap::from([( "index-file-in-data-file-dir".to_string(), "true".to_string(), - ), + )]), + ) + .await; + give_the_table_a_snapshot(&catalog, &identifier).await; + + for change in [ SchemaChange::set_option( "index-file-in-data-file-dir".to_string(), "false".to_string(), @@ -895,6 +985,99 @@ mod tests { .unwrap_err(); assert!(matches!(err, Error::Unsupported { .. }), "{err:?}"); } + + // Repeating the stored value changes nothing, so Java lets it through and + // schema reconciliation stays idempotent. + catalog + .alter_table( + &identifier, + vec![SchemaChange::set_option( + "index-file-in-data-file-dir".to_string(), + "true".to_string(), + )], + false, + ) + .await + .unwrap(); + } + + #[tokio::test] + async fn test_alter_table_rejects_spelling_out_the_default_index_layout_once_written() { + use crate::spec::SchemaChange; + + // Java compares the stored string and does not treat "the default, written + // out" as unchanged for this option: `isUnchangedNormalizedKey` special-cases + // only `type`, `primary-key` and `partition`. Setting `false` on a table that + // never stored the option is therefore a change, and is rejected once + // snapshots exist, even though it names the layout the table already uses. + let (_temp_dir, catalog) = create_test_catalog(); + let identifier = create_table_for_alter(&catalog, HashMap::new()).await; + give_the_table_a_snapshot(&catalog, &identifier).await; + + let err = catalog + .alter_table( + &identifier, + vec![SchemaChange::set_option( + "index-file-in-data-file-dir".to_string(), + "false".to_string(), + )], + false, + ) + .await + .unwrap_err(); + assert!(matches!(err, Error::Unsupported { .. }), "{err:?}"); + } + + #[tokio::test] + async fn test_alter_table_can_choose_where_index_files_live_before_the_first_write() { + use crate::spec::SchemaChange; + + // Nothing is written yet, so there is no file to strand: Java reaches + // `checkAlterTableOption` only once the table has snapshots, which lets a + // caller pick the layout through ALTER before the first write. + let (_temp_dir, catalog) = create_test_catalog(); + let identifier = create_table_for_alter(&catalog, HashMap::new()).await; + + catalog + .alter_table( + &identifier, + vec![SchemaChange::set_option( + "index-file-in-data-file-dir".to_string(), + "true".to_string(), + )], + false, + ) + .await + .unwrap(); + assert_eq!( + catalog + .get_table(&identifier) + .await + .unwrap() + .schema() + .options() + .get("index-file-in-data-file-dir") + .map(String::as_str), + Some("true") + ); + + catalog + .alter_table( + &identifier, + vec![SchemaChange::remove_option( + "index-file-in-data-file-dir".to_string(), + )], + false, + ) + .await + .unwrap(); + assert!(!catalog + .get_table(&identifier) + .await + .unwrap() + .schema() + .options() + .contains_key("index-file-in-data-file-dir")); } #[tokio::test] diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index 635e188ee..ece7992ca 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -18,8 +18,9 @@ use crate::spec::core_options::{ first_row_supports_changelog_producer, ChangelogProducer, CoreOptions, MergeEngine, BLOB_DESCRIPTOR_FIELD_OPTION, BLOB_FIELD_OPTION, BLOB_VIEW_FIELD_OPTION, BUCKET_KEY_OPTION, - CHANGELOG_PRODUCER_OPTION, POSTPONE_BUCKET, QUERY_AUTH_ENABLED_OPTION, SEQUENCE_FIELD_OPTION, - TABLE_READ_SEQUENCE_NUMBER_ENABLED_OPTION, TABLE_TYPE_OPTION, + CHANGELOG_PRODUCER_OPTION, INDEX_FILE_IN_DATA_FILE_DIR_OPTION, POSTPONE_BUCKET, + QUERY_AUTH_ENABLED_OPTION, SEQUENCE_FIELD_OPTION, TABLE_READ_SEQUENCE_NUMBER_ENABLED_OPTION, + TABLE_TYPE_OPTION, }; use crate::spec::types::{ArrayType, DataType, MapType, MultisetType, RowType, VarCharType}; use crate::spec::{ @@ -147,6 +148,15 @@ impl TableSchema { /// A stored `query-auth.enabled = true` can't be turned off by a dynamic /// override, and the declared `type` can't be changed by one: an override /// could re-route foreign data through the Paimon reader. + /// + /// `index-file-in-data-file-dir` can't be changed by one either. It selects + /// the directory every bucket-local index file is written to and read from, + /// while an index manifest records only the file name, so a copy carrying an + /// overridden value would write hash and deletion-vector index files where a + /// normally loaded table cannot find them. Java rejects such an override + /// outright in `AbstractFileStoreTable.checkImmutability`; this copy cannot + /// fail, so the stored value wins instead, and an absent one leaves the + /// default standing. pub fn copy_with_options(&self, mut extra: HashMap) -> Self { if self.core_options().query_auth_enabled() { extra.insert(QUERY_AUTH_ENABLED_OPTION.to_string(), "true".to_string()); @@ -155,6 +165,13 @@ impl TableSchema { Some(declared) => extra.insert(TABLE_TYPE_OPTION.to_string(), declared.clone()), None => extra.remove(TABLE_TYPE_OPTION), }; + match self.options.get(INDEX_FILE_IN_DATA_FILE_DIR_OPTION) { + Some(stored) => extra.insert( + INDEX_FILE_IN_DATA_FILE_DIR_OPTION.to_string(), + stored.clone(), + ), + None => extra.remove(INDEX_FILE_IN_DATA_FILE_DIR_OPTION), + }; let mut new_schema = self.clone(); new_schema.options.extend(extra); new_schema @@ -2437,6 +2454,65 @@ mod tests { ); } + #[test] + fn a_dynamic_copy_cannot_move_where_index_files_live() { + // Every reader and writer of a bucket-local index file asks the schema for + // this option — the deletion-vector and hash-index writers included — and an + // index manifest records only a file name, so a copy that changed it would + // write files where a normally loaded table cannot find them. + let stored = TableSchema::new( + 0, + &Schema::builder() + .column("id", DataType::Int(IntType::new())) + .option(INDEX_FILE_IN_DATA_FILE_DIR_OPTION, "true") + .build() + .unwrap(), + ); + let copied = stored.copy_with_options(HashMap::from([( + INDEX_FILE_IN_DATA_FILE_DIR_OPTION.to_string(), + "false".to_string(), + )])); + assert!( + copied.core_options().index_file_in_data_file_dir(), + "an override must not turn a stored bucket-local layout off" + ); + + // The default is equally load-bearing: turning it on for a table written + // without it would look for existing index files in the bucket directory. + let unset = TableSchema::new( + 0, + &Schema::builder() + .column("id", DataType::Int(IntType::new())) + .build() + .unwrap(), + ); + let copied = unset.copy_with_options(HashMap::from([( + INDEX_FILE_IN_DATA_FILE_DIR_OPTION.to_string(), + "true".to_string(), + )])); + assert!( + !copied.core_options().index_file_in_data_file_dir(), + "an override must not turn a defaulted layout on" + ); + assert!( + !copied + .options() + .contains_key(INDEX_FILE_IN_DATA_FILE_DIR_OPTION), + "an absent option must stay absent rather than be pinned to a literal" + ); + + // Unrelated overrides still merge. + let copied = stored.copy_with_options(HashMap::from([( + "scan.snapshot-id".to_string(), + "7".to_string(), + )])); + assert_eq!( + copied.options().get("scan.snapshot-id"), + Some(&"7".to_string()) + ); + assert!(copied.core_options().index_file_in_data_file_dir()); + } + #[test] fn test_copy_with_replaced_options() { let schema = Schema::builder() diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs index 720f157f4..e24d0a74d 100644 --- a/crates/paimon/src/table/mod.rs +++ b/crates/paimon/src/table/mod.rs @@ -583,3 +583,60 @@ pub(crate) fn query_auth_table() -> Table { None, ) } + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use super::Table; + use crate::catalog::Identifier; + use crate::io::FileIOBuilder; + use crate::spec::{DataType, IntType, Schema, TableSchema}; + + /// The value every reader and writer of a bucket-local index file consults. + /// + /// Each consumer resolves its own path from it — deletion vectors in + /// `table_scan` and `data_evolution_writer`, primary-key ANN segments in + /// `pk_vector_scan`, full-text archives, the dynamic-bucket hash index, and + /// `TableCommit::abort` — and each of those resolutions is covered where it + /// lives. What a copy must not do is hand them a different value than the one + /// the files on disk were written under. + #[test] + fn a_dynamic_copy_reads_the_stored_index_layout() { + let file_io = FileIOBuilder::new("memory").build().unwrap(); + let stored = TableSchema::new( + 0, + &Schema::builder() + .column("id", DataType::Int(IntType::new())) + .option("index-file-in-data-file-dir", "true") + .build() + .unwrap(), + ); + let table = Table::new( + file_io, + Identifier::new("default", "t"), + "memory:/t".to_string(), + stored, + None, + ); + + let copied = table.copy_with_options(HashMap::from([( + "index-file-in-data-file-dir".to_string(), + "false".to_string(), + )])); + assert!( + copied.schema().core_options().index_file_in_data_file_dir(), + "a read through a copied table must resolve index files under the stored layout" + ); + + // The copy is otherwise a normal copy. + let copied = table.copy_with_options(HashMap::from([( + "scan.snapshot-id".to_string(), + "3".to_string(), + )])); + assert_eq!( + copied.schema().options().get("scan.snapshot-id"), + Some(&"3".to_string()) + ); + } +} diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index 8ce158c08..9f3f707d1 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -3255,16 +3255,28 @@ mod tests { ) } + /// A table whose schema *stores* `options`, as a catalog-loaded one would. + /// + /// Not `copy_with_options`: that applies dynamic overrides, and options fixed + /// at creation — `index-file-in-data-file-dir` among them — are pinned to the + /// stored value there, so a test configuring one has to persist it. fn test_table_with_options( file_io: &FileIO, table_path: &str, options: HashMap, ) -> Table { + use crate::spec::{DataType, IntType, Schema, VarCharType}; + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("name", DataType::VarChar(VarCharType::string_type())) + .options(options) + .build() + .unwrap(); Table::new( file_io.clone(), Identifier::new("default", "test_table"), table_path.to_string(), - test_schema().copy_with_options(options), + TableSchema::new(0, &schema), None, ) } diff --git a/crates/paimon/src/table/table_write.rs b/crates/paimon/src/table/table_write.rs index c37df7b66..1a0c39938 100644 --- a/crates/paimon/src/table/table_write.rs +++ b/crates/paimon/src/table/table_write.rs @@ -3083,6 +3083,68 @@ pub(in crate::table) mod tests { TableSchema::new(0, &schema) } + #[tokio::test] + async fn a_dynamic_copy_cannot_move_where_a_written_hash_index_lands() { + // An index manifest records only a file name, so a write must place the + // hash index where a normally loaded table will look for it. A dynamic + // override of `index-file-in-data-file-dir` would break that pairing, so + // `copy_with_options` pins the option to the stored value. + let file_io = test_file_io(); + let table_path = "memory:/test_hash_index_layout_survives_a_copy"; + setup_dirs(&file_io, table_path).await; + + let schema = Schema::builder() + .column("pt", DataType::VarChar(VarCharType::string_type())) + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .partition_keys(["pt"]) + .primary_key(["pt", "id"]) + .option("changelog-producer", "input") + .option("index-file-in-data-file-dir", "true") + .build() + .unwrap(); + let table = Table::new( + file_io.clone(), + Identifier::new("default", "test_hash_index_layout_survives_a_copy"), + table_path.to_string(), + TableSchema::new(0, &schema), + None, + ); + let copied = table.copy_with_options(HashMap::from([( + "index-file-in-data-file-dir".to_string(), + "false".to_string(), + )])); + + let mut table_write = TableWrite::new(&copied, "test-user".to_string()).unwrap(); + table_write + .write_arrow_batch(&make_partitioned_batch_with_value_kind( + vec!["a", "a"], + vec![1, 2], + vec![10, 20], + vec![0, 0], + )) + .await + .unwrap(); + let messages = table_write.prepare_commit().await.unwrap(); + assert_eq!(messages[0].new_index_files[0].index_type, "HASH"); + let name = messages[0].new_index_files[0].file_name.clone(); + + assert!( + file_io + .exists(&format!("{table_path}/pt=a/bucket-0/{name}")) + .await + .unwrap(), + "the hash index must stay in the bucket data-file directory the stored option selects" + ); + assert!( + !file_io + .exists(&format!("{table_path}/index/{name}")) + .await + .unwrap(), + "the override must not have moved it to the table index directory" + ); + } + #[tokio::test] async fn test_default_changelog_producer_accepts_value_kind() { let file_io = test_file_io();