This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 0a4470e8 Align dynamic bucket writes and HASH index recovery with Java
(#1016)
0a4470e8 is described below
commit 0a4470e85102665aa4f166197648cd25c6cb7ba8
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Oct 3 21:52:26 2026 +0800
Align dynamic bucket writes and HASH index recovery with Java (#1016)
---
crates/paimon/src/spec/core_options.rs | 20 +-
crates/paimon/src/table/bucket_assigner_dynamic.rs | 622 +++++++++++++++++----
crates/paimon/src/table/data_evolution_writer.rs | 8 +-
crates/paimon/src/table/index_file_path.rs | 250 +--------
crates/paimon/src/table/table_scan.rs | 7 -
crates/paimon/src/table/table_write.rs | 42 +-
crates/paimon/tests/dynamic_bucket_write_test.rs | 202 +++++++
crates/paimon/tests/scan_planning_parity_test.rs | 33 +-
crates/paimon/tests/table_update_paths_test.rs | 52 +-
9 files changed, 844 insertions(+), 392 deletions(-)
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index 8760c454..f842ed2e 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -156,7 +156,9 @@ const DEFAULT_PARQUET_ROW_GROUP_MAX_INFLIGHT_BYTES: i64 =
256 * 1024 * 1024;
const DEFAULT_MOSAIC_READ_PREFETCH_ROW_GROUPS: usize = 8;
const DEFAULT_MOSAIC_READ_PREFETCH_MAX_BYTES: i64 = 64 * 1024 * 1024;
const DYNAMIC_BUCKET_TARGET_ROW_NUM_OPTION: &str =
"dynamic-bucket.target-row-num";
-const DEFAULT_DYNAMIC_BUCKET_TARGET_ROW_NUM: i64 = 200_000;
+const DEFAULT_DYNAMIC_BUCKET_TARGET_ROW_NUM: i64 = 2_000_000;
+pub(crate) const MAX_DYNAMIC_BUCKETS: i32 = 32768;
+const DYNAMIC_BUCKET_MAX_BUCKETS_OPTION: &str = "dynamic-bucket.max-buckets";
const DEFAULT_GLOBAL_INDEX_ROW_COUNT_PER_SHARD: i64 = 100_000;
const DEFAULT_GLOBAL_INDEX_THREAD_NUM: i64 = 32;
pub(crate) const DEFAULT_GLOBAL_INDEX_VINDEX_READ_THREAD_NUM: usize = 64;
@@ -1714,7 +1716,7 @@ impl<'a> CoreOptions<'a> {
/// Target row number per bucket for dynamic bucket mode (bucket=-1).
/// When a bucket reaches this number, a new bucket is created.
- /// Default is 200,000. Java Paimon defaults this to 2,000,000.
+ /// Default is 2,000,000, matching Java Paimon.
pub fn dynamic_bucket_target_row_num(&self) -> i64 {
self.options
.get(DYNAMIC_BUCKET_TARGET_ROW_NUM_OPTION)
@@ -1722,6 +1724,20 @@ impl<'a> CoreOptions<'a> {
.unwrap_or(DEFAULT_DYNAMIC_BUCKET_TARGET_ROW_NUM)
}
+ /// Maximum number of dynamic buckets in each partition. `-1` grows up to
+ /// Java's signed-short bucket id limit; an explicit limit reuses full
buckets.
+ pub fn dynamic_bucket_max_buckets(&self) -> crate::Result<i32> {
+ let value = self
+ .parse_i64_option(DYNAMIC_BUCKET_MAX_BUCKETS_OPTION)?
+ .unwrap_or(-1);
+ if value != -1 &&
!(1..=i64::from(MAX_DYNAMIC_BUCKETS)).contains(&value) {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!("'{DYNAMIC_BUCKET_MAX_BUCKETS_OPTION}' must
be -1 or between 1 and {MAX_DYNAMIC_BUCKETS}, but was {value}"),
+ });
+ }
+ Ok(value as i32)
+ }
+
/// When true, blob field reads return serialized BlobDescriptor bytes
/// instead of actual blob bytes. Default is false.
pub fn blob_as_descriptor(&self) -> bool {
diff --git a/crates/paimon/src/table/bucket_assigner_dynamic.rs
b/crates/paimon/src/table/bucket_assigner_dynamic.rs
index 2919d599..6dbc50aa 100644
--- a/crates/paimon/src/table/bucket_assigner_dynamic.rs
+++ b/crates/paimon/src/table/bucket_assigner_dynamic.rs
@@ -21,15 +21,20 @@
//! and per-partition index (`PartitionIndex`) used by both dynamic and
cross-partition modes.
use crate::io::FileIO;
+use crate::spec::MAX_DYNAMIC_BUCKETS;
use crate::spec::{
- batch_hash_codes, batch_to_serialized_bytes, bucket_path_under, BinaryRow,
DataField,
- IndexFileMeta, IndexManifest, IndexManifestEntry, PartitionComputer,
EMPTY_SERIALIZED_ROW,
+ batch_hash_codes, batch_to_serialized_bytes, bucket_path_under, BinaryRow,
CoreOptions,
+ DataField, IndexFileMeta, IndexManifest, IndexManifestEntry,
PartitionComputer,
+ EMPTY_SERIALIZED_ROW,
};
use crate::table::bucket_assigner::{BatchAssignOutput, BucketAssigner,
PartitionBucketKey};
+use crate::table::data_file_path_factory::{DataFilePath, DataFilePathFactory};
use crate::table::index_file_path::IndexFileLocation;
-use crate::table::SnapshotManager;
+use crate::table::partition_filter::PartitionFilter;
+use crate::table::{Snapshot, SnapshotManager, Table, TableScan};
use crate::Result;
use arrow_array::RecordBatch;
+use rand::seq::SliceRandom;
use std::collections::{HashMap, HashSet};
use uuid::Uuid;
@@ -51,11 +56,15 @@ impl HashIndexFile {
async fn read(file_io: &FileIO, path: &str) -> Result<Vec<i32>> {
let input = file_io.new_input(path)?;
let content = input.read().await?;
- debug_assert!(
- content.len() % 4 == 0,
- "hash index file size {} is not aligned to 4 bytes",
- content.len()
- );
+ if content.len() % 4 != 0 {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Corrupt HASH index {path}: expected a multiple of 4
bytes, got {}",
+ content.len()
+ ),
+ source: None,
+ });
+ }
let count = content.len() / 4;
let mut hashes = Vec::with_capacity(count);
for i in 0..count {
@@ -72,10 +81,13 @@ impl HashIndexFile {
}
/// Write key hashes to a new hash index file, returning its metadata.
- async fn write(file_io: &FileIO, dir: &str, hashes: &[i32]) ->
Result<IndexFileMeta> {
- let file_name = format!("index-{}-0", Uuid::new_v4());
- let path = format!("{dir}/{file_name}");
-
+ async fn write_at(
+ file_io: &FileIO,
+ file_name: String,
+ location: DataFilePath,
+ hashes: &[i32],
+ ) -> Result<IndexFileMeta> {
+ file_io.mkdirs(location.parent()).await?;
let mut buf = Vec::with_capacity(hashes.len() * 4);
for &h in hashes {
buf.extend_from_slice(&h.to_be_bytes());
@@ -85,8 +97,11 @@ impl HashIndexFile {
.len()
.try_into()
.expect("hash index file size exceeds i64::MAX");
- let output = file_io.new_output(&path)?;
- output.write(bytes::Bytes::from(buf)).await?;
+ let output = file_io.new_output(&location.path)?;
+ if let Err(error) = output.write(bytes::Bytes::from(buf)).await {
+ let _ = file_io.delete_file(&location.path).await;
+ return Err(error);
+ }
Ok(IndexFileMeta {
index_type: HASH_INDEX.to_string(),
@@ -97,7 +112,7 @@ impl HashIndexFile {
.try_into()
.expect("hash index row count exceeds i32::MAX"),
deletion_vectors_ranges: None,
- external_path: None,
+ external_path: location.external_path,
global_index_meta: None,
})
}
@@ -113,11 +128,13 @@ impl HashIndexFile {
/// `prepare_commit` writes the full hash set to a new hash index file.
///
/// Reference:
[org.apache.paimon.index.DynamicBucketIndexMaintainer](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/index/DynamicBucketIndexMaintainer.java)
-pub(crate) struct DynamicBucketIndexMaintainer {
+struct DynamicBucketIndexMaintainer {
/// All key hashes in this bucket (restored + new).
hashes: HashSet<i32>,
/// Whether any new hashes were added since last commit.
modified: bool,
+ /// Java caches a per-bucket index path factory across checkpoints.
+ paths: Option<DataFilePathFactory>,
}
impl DynamicBucketIndexMaintainer {
@@ -127,6 +144,7 @@ impl DynamicBucketIndexMaintainer {
Self {
hashes,
modified: false,
+ paths: None,
}
}
@@ -141,13 +159,34 @@ impl DynamicBucketIndexMaintainer {
pub async fn prepare_commit(
&mut self,
file_io: &FileIO,
- index_dir: &str,
+ layout: &HashIndexLayout<'_>,
+ bucket: i32,
+ options: &HashMap<String, String>,
) -> Result<Vec<IndexFileMeta>> {
if !self.modified {
return Ok(Vec::new());
}
let hashes: Vec<i32> = self.hashes.iter().copied().collect();
- let meta = HashIndexFile::write(file_io, index_dir, &hashes).await?;
+ let file_name = format!("index-{}-0", Uuid::new_v4());
+ let location = if layout.index_file_in_data_file_dir {
+ if self.paths.is_none() {
+ self.paths = Some(DataFilePathFactory::new(
+ layout.table_path,
+ layout.partition_path,
+ bucket,
+ options,
+ )?);
+ }
+ self.paths.as_ref().unwrap().new_path(&file_name)?
+ } else {
+ let external_path =
+ super::external_path::new_index_external_path(options, false,
"", &file_name)?;
+ DataFilePath {
+ path: layout.resolve(bucket, &file_name,
external_path.as_deref()),
+ external_path,
+ }
+ };
+ let meta = HashIndexFile::write_at(file_io, file_name, location,
&hashes).await?;
self.modified = false;
Ok(vec![meta])
}
@@ -187,7 +226,8 @@ impl HashIndexLayout<'_> {
bucket_path_under(&data_root, self.partition_path, bucket)
}
- /// The directory a new hash index file for `bucket` is written into.
+ /// The default directory for a hash index without an external location.
+ #[cfg(test)]
fn directory(&self, bucket: i32) -> String {
let bucket_path = self.bucket_path(bucket);
self.location(&bucket_path).directory()
@@ -214,8 +254,9 @@ struct PartitionIndex {
non_full_buckets: HashMap<i32, i64>,
/// All known bucket ids
all_buckets: HashSet<i32>,
- /// Next bucket id to allocate (avoids linear scan).
+ /// Next unused bucket id to try, including holes in a restored index.
next_bucket_id: i32,
+ bucket_ids: Vec<i32>,
target_bucket_row_number: i64,
/// Per-bucket index maintainers for writing hash index files at commit
time.
bucket_maintainers: HashMap<i32, DynamicBucketIndexMaintainer>,
@@ -229,6 +270,7 @@ impl PartitionIndex {
non_full_buckets: HashMap::new(),
all_buckets: HashSet::new(),
next_bucket_id: 0,
+ bucket_ids: Vec::new(),
target_bucket_row_number,
bucket_maintainers: HashMap::new(),
}
@@ -253,6 +295,21 @@ impl PartitionIndex {
continue;
}
let bucket = entry.bucket;
+ if !(0..MAX_DYNAMIC_BUCKETS).contains(&bucket) {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Dynamic bucket id must be between 0 and {}, but was
{bucket}",
+ MAX_DYNAMIC_BUCKETS - 1
+ ),
+ source: None,
+ });
+ }
+ if bucket_hashes.contains_key(&bucket) {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Multiple HASH indexes for dynamic bucket
{bucket}"),
+ source: None,
+ });
+ }
let path = layout.resolve(
bucket,
&entry.index_file.file_name,
@@ -260,8 +317,23 @@ impl PartitionIndex {
);
let hashes = HashIndexFile::read(file_io, &path).await?;
let count = hashes.len() as i64;
+ if count != entry.index_file.row_count {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Corrupt HASH index {path}: expected {} hashes, got
{count}",
+ entry.index_file.row_count
+ ),
+ source: None,
+ });
+ }
for &h in &hashes {
- hash_to_bucket.insert(h, bucket);
+ if let Some(previous) = hash_to_bucket.insert(h, bucket) {
+ if previous != bucket {
+ return Err(crate::Error::DataInvalid {
+ message: format!("HASH index assigns hash {h} to
both buckets {previous} and {bucket}"), source: None,
+ });
+ }
+ }
}
*bucket_row_counts.entry(bucket).or_insert(0) += count;
bucket_hashes.entry(bucket).or_default().extend(hashes);
@@ -278,13 +350,14 @@ impl PartitionIndex {
.map(|(bucket, hashes)| (bucket,
DynamicBucketIndexMaintainer::new(hashes)))
.collect();
- let next_bucket_id = all_buckets.iter().copied().max().map_or(0, |m| m
+ 1);
+ let bucket_ids = all_buckets.iter().copied().collect();
Ok(Self {
hash_to_bucket,
non_full_buckets,
all_buckets,
- next_bucket_id,
+ next_bucket_id: 0,
+ bucket_ids,
target_bucket_row_number,
bucket_maintainers,
})
@@ -295,10 +368,10 @@ impl PartitionIndex {
/// 1. If the hash was seen before, return its existing bucket.
/// 2. Otherwise, find a non-full bucket and assign the hash there.
/// 3. If all buckets are full, create a new bucket.
- fn assign(&mut self, hash: i32) -> i32 {
+ fn assign(&mut self, hash: i32, max_buckets: i32) -> Result<i32> {
// 1. Already assigned
if let Some(&bucket) = self.hash_to_bucket.get(&hash) {
- return bucket;
+ return Ok(bucket);
}
// 2. Find a non-full bucket
@@ -322,20 +395,45 @@ impl PartitionIndex {
.entry(bucket)
.or_insert_with(|| DynamicBucketIndexMaintainer::new(vec![]))
.notify_new_record(hash);
- return bucket;
+ return Ok(bucket);
}
- // 3. Create a new bucket
- let new_bucket = self.next_bucket_id;
- self.next_bucket_id += 1;
- self.all_buckets.insert(new_bucket);
- self.non_full_buckets.insert(new_bucket, 1);
- self.hash_to_bucket.insert(hash, new_bucket);
+ // Java PartitionIndex first allocates an unused id in range. A
configured
+ // maximum is a soft row-count limit: once reached, reuse an existing
bucket.
+ let limit = if max_buckets == -1 {
+ MAX_DYNAMIC_BUCKETS
+ } else {
+ max_buckets
+ };
+ while self.next_bucket_id < limit &&
self.all_buckets.contains(&self.next_bucket_id) {
+ self.next_bucket_id += 1;
+ }
+ let bucket = if self.next_bucket_id < limit {
+ let bucket = self.next_bucket_id;
+ self.next_bucket_id += 1;
+ self.all_buckets.insert(bucket);
+ self.bucket_ids.push(bucket);
+ self.non_full_buckets.insert(bucket, 1);
+ bucket
+ } else if max_buckets == -1 {
+ return Err(crate::Error::DataInvalid {
+ message: "No dynamic bucket id remains below Java
Short.MAX_VALUE. Increase dynamic-bucket.target-row-num.".to_string(), source:
None,
+ });
+ } else {
+ *self
+ .bucket_ids
+ .choose(&mut rand::thread_rng())
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: "No dynamic bucket is available".to_string(),
+ source: None,
+ })?
+ };
+ self.hash_to_bucket.insert(hash, bucket);
self.bucket_maintainers
- .entry(new_bucket)
+ .entry(bucket)
.or_insert_with(|| DynamicBucketIndexMaintainer::new(vec![]))
.notify_new_record(hash);
- new_bucket
+ Ok(bucket)
}
/// Write hash index files for all modified buckets, returning (bucket,
index_files) pairs.
@@ -343,13 +441,23 @@ impl PartitionIndex {
&mut self,
file_io: &FileIO,
layout: &HashIndexLayout<'_>,
+ options: &HashMap<String, String>,
+ created_paths: &mut Vec<String>,
) -> Result<Vec<(i32, Vec<IndexFileMeta>)>> {
let mut result = Vec::new();
let buckets: Vec<i32> =
self.bucket_maintainers.keys().copied().collect();
for bucket in buckets {
if let Some(maintainer) = self.bucket_maintainers.get_mut(&bucket)
{
- let index_dir = layout.directory(bucket);
- let files = maintainer.prepare_commit(file_io,
&index_dir).await?;
+ let files = maintainer
+ .prepare_commit(file_io, layout, bucket, options)
+ .await?;
+ for file in &files {
+ created_paths.push(layout.resolve(
+ bucket,
+ &file.file_name,
+ file.external_path.as_deref(),
+ ));
+ }
if !files.is_empty() {
result.push((bucket, files));
}
@@ -373,9 +481,9 @@ pub(crate) struct DynamicBucketAssigner {
fields: Vec<DataField>,
partition_indexes: HashMap<Vec<u8>, PartitionIndex>,
target_bucket_row_number: i64,
- file_io: FileIO,
- table_location: String,
- data_file_path_directory: Option<String>,
+ table: Table,
+ max_buckets: i32,
+ snapshot: Option<Snapshot>,
/// Cached index manifest entries from the latest snapshot (loaded once).
cached_index_entries: Option<Vec<IndexManifestEntry>>,
/// Overwrite mode: skip loading existing index entries.
@@ -389,37 +497,33 @@ pub(crate) struct DynamicBucketAssigner {
}
impl DynamicBucketAssigner {
- #[allow(clippy::too_many_arguments)]
pub fn new(
+ table: Table,
partition_field_indices: Vec<usize>,
primary_key_indices: Vec<usize>,
- fields: Vec<DataField>,
- target_bucket_row_number: i64,
- file_io: FileIO,
- table_location: String,
is_overwrite: bool,
partition_computer: PartitionComputer,
- index_file_in_data_file_dir: bool,
- ) -> Self {
- Self {
+ ) -> Result<Self> {
+ let options = CoreOptions::new(table.schema().options());
+ if table.schema().options().contains_key("bucket-key") {
+ return Err(crate::Error::ConfigInvalid {
+ message: "Cannot define 'bucket-key' in dynamic bucket
mode".to_string(),
+ });
+ }
+ Ok(Self {
partition_field_indices,
primary_key_indices,
- fields,
+ fields: table.schema().fields().to_vec(),
partition_indexes: HashMap::new(),
- target_bucket_row_number,
- file_io,
- table_location,
- data_file_path_directory: None,
+ target_bucket_row_number: options.dynamic_bucket_target_row_num(),
+ max_buckets: options.dynamic_bucket_max_buckets()?,
+ snapshot: None,
cached_index_entries: None,
is_overwrite,
partition_computer,
- index_file_in_data_file_dir,
- }
- }
-
- pub(super) fn with_data_file_path_directory(mut self, directory:
Option<&str>) -> Self {
- self.data_file_path_directory = directory.map(str::to_string);
- self
+ index_file_in_data_file_dir: options.index_file_in_data_file_dir(),
+ table,
+ })
}
pub fn set_overwrite(&mut self, is_overwrite: bool) {
@@ -436,21 +540,24 @@ impl DynamicBucketAssigner {
self.cached_index_entries = Some(Vec::new());
return Ok(());
}
- let snapshot_manager =
- SnapshotManager::new(self.file_io.clone(),
self.table_location.clone());
+ let snapshot_manager = SnapshotManager::new(
+ self.table.file_io().clone(),
+ self.table.location().to_string(),
+ );
let latest_snapshot = snapshot_manager.get_latest_snapshot().await?;
- let entries = if let Some(snapshot) = latest_snapshot {
+ let entries = if let Some(snapshot) = &latest_snapshot {
if let Some(index_manifest_name) = snapshot.index_manifest() {
let manifest_dir = snapshot_manager.manifest_dir();
let index_manifest_path =
format!("{manifest_dir}/{index_manifest_name}");
- IndexManifest::read(&self.file_io, &index_manifest_path).await?
+ IndexManifest::read(self.table.file_io(),
&index_manifest_path).await?
} else {
Vec::new()
}
} else {
Vec::new()
};
+ self.snapshot = latest_snapshot;
self.cached_index_entries = Some(entries);
Ok(())
}
@@ -472,16 +579,19 @@ impl DynamicBucketAssigner {
.cloned()
.collect();
+ self.validate_data_buckets(partition_bytes, &partition_entries)
+ .await?;
if !partition_entries.is_empty() {
let partition_path = self.partition_path(partition_bytes)?;
+ let options = CoreOptions::new(self.table.schema().options());
let layout = HashIndexLayout {
- data_file_path_directory:
self.data_file_path_directory.as_deref(),
- table_path: self.table_location.trim_end_matches('/'),
+ data_file_path_directory: options.data_file_path_directory(),
+ 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,
+ self.table.file_io(),
&layout,
&partition_entries,
self.target_bucket_row_number,
@@ -491,6 +601,39 @@ impl DynamicBucketAssigner {
Ok(PartitionIndex::empty(self.target_bucket_row_number))
}
+
+ /// Use the same snapshot as index restoration. Treating existing data
without
+ /// its HASH index as a new partition can assign an old key to a different
bucket.
+ async fn validate_data_buckets(
+ &self,
+ partition: &[u8],
+ indexes: &[IndexManifestEntry],
+ ) -> Result<()> {
+ let Some(snapshot) = &self.snapshot else {
+ return Ok(());
+ };
+ let fields = self.table.schema().partition_fields();
+ let filter = if fields.is_empty() {
+ None
+ } else {
+ Some(PartitionFilter::from_partition_set(
+ HashSet::from([partition.to_vec()]),
+ &fields,
+ )?)
+ };
+ let scan =
+ TableScan::new(&self.table, filter, vec![], None, None,
None).with_scan_all_files();
+ let indexed: HashSet<i32> = indexes.iter().map(|entry|
entry.bucket).collect();
+ for entry in scan.plan_manifest_entries(snapshot).await? {
+ if !indexed.contains(&entry.bucket()) {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Dynamic-bucket partition has data files
but no complete HASH index for bucket {}. Rewrite the partition before
incremental writes.", entry.bucket()),
+ source: None,
+ });
+ }
+ }
+ Ok(())
+ }
}
impl BucketAssigner for DynamicBucketAssigner {
@@ -527,7 +670,7 @@ impl BucketAssigner for DynamicBucketAssigner {
let mut buckets = Vec::with_capacity(batch.num_rows());
for (row_idx, pb) in partition_bytes_vec.iter().enumerate() {
let partition_index = self.partition_indexes.get_mut(pb).unwrap();
- buckets.push(partition_index.assign(hash_codes[row_idx]));
+ buckets.push(partition_index.assign(hash_codes[row_idx],
self.max_buckets)?);
}
Ok(BatchAssignOutput {
@@ -542,29 +685,48 @@ impl BucketAssigner for DynamicBucketAssigner {
&mut self,
file_io: &FileIO,
) -> Result<HashMap<PartitionBucketKey, Vec<IndexFileMeta>>> {
- 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<Vec<u8>> =
self.partition_indexes.keys().cloned().collect();
- 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 {
- data_file_path_directory:
self.data_file_path_directory.as_deref(),
- 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,
&layout).await?;
- for (bucket, idx_files) in bucket_files {
- result.insert((partition_bytes.clone(), bucket),
idx_files);
+ let mut created_paths = Vec::new();
+ let result = async {
+ 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<Vec<u8>> =
self.partition_indexes.keys().cloned().collect();
+ 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 options = CoreOptions::new(self.table.schema().options());
+ let layout = HashIndexLayout {
+ data_file_path_directory:
options.data_file_path_directory(),
+ 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,
+ &layout,
+ self.table.schema().options(),
+ &mut created_paths,
+ )
+ .await?;
+ for (bucket, idx_files) in bucket_files {
+ result.insert((partition_bytes.clone(), bucket),
idx_files);
+ }
}
}
+ Ok(result)
}
- Ok(result)
+ .await;
+ if result.is_err() {
+ for path in created_paths {
+ let _ = file_io.delete_file(&path).await;
+ }
+ }
+ result
}
}
@@ -572,6 +734,28 @@ impl BucketAssigner for DynamicBucketAssigner {
mod tests {
use super::*;
+ fn table_layout(table_path: &str) -> HashIndexLayout<'_> {
+ HashIndexLayout {
+ table_path,
+ partition_path: "",
+ data_file_path_directory: None,
+ index_file_in_data_file_dir: false,
+ }
+ }
+
+ async fn write_hash_index(
+ file_io: &FileIO,
+ dir: &str,
+ hashes: &[i32],
+ ) -> Result<IndexFileMeta> {
+ let file_name = format!("index-{}-0", Uuid::new_v4());
+ let location = DataFilePath {
+ path: format!("{dir}/{file_name}"),
+ external_path: None,
+ };
+ HashIndexFile::write_at(file_io, file_name, location, hashes).await
+ }
+
// -- DynamicBucketIndexMaintainer tests --
#[tokio::test]
@@ -582,20 +766,29 @@ mod tests {
let mut m = DynamicBucketIndexMaintainer::new(vec![]);
// No modification → empty
- let files = m.prepare_commit(&file_io, &dir).await.unwrap();
+ let files = m
+ .prepare_commit(&file_io, &table_layout(&dir), 0, &HashMap::new())
+ .await
+ .unwrap();
assert!(files.is_empty());
// Add hashes
m.notify_new_record(1);
m.notify_new_record(2);
m.notify_new_record(1); // duplicate, no effect
- let files = m.prepare_commit(&file_io, &dir).await.unwrap();
+ let files = m
+ .prepare_commit(&file_io, &table_layout(&dir), 0, &HashMap::new())
+ .await
+ .unwrap();
assert_eq!(files.len(), 1);
assert_eq!(files[0].index_type, HASH_INDEX);
assert_eq!(files[0].row_count, 2);
// No new modification → empty again
- let files = m.prepare_commit(&file_io, &dir).await.unwrap();
+ let files = m
+ .prepare_commit(&file_io, &table_layout(&dir), 0, &HashMap::new())
+ .await
+ .unwrap();
assert!(files.is_empty());
}
@@ -607,46 +800,261 @@ mod tests {
let mut m = DynamicBucketIndexMaintainer::new(vec![10, 20]);
// Restored hashes don't count as modified
- let files = m.prepare_commit(&file_io, &dir).await.unwrap();
+ let files = m
+ .prepare_commit(&file_io, &table_layout(&dir), 0, &HashMap::new())
+ .await
+ .unwrap();
assert!(files.is_empty());
// Adding a new hash triggers write (includes restored + new)
m.notify_new_record(30);
- let files = m.prepare_commit(&file_io, &dir).await.unwrap();
+ let files = m
+ .prepare_commit(&file_io, &table_layout(&dir), 0, &HashMap::new())
+ .await
+ .unwrap();
assert_eq!(files.len(), 1);
assert_eq!(files[0].row_count, 3);
}
+ #[tokio::test]
+ async fn test_hash_external_round_robin_survives_checkpoints() {
+ let tmp = tempfile::tempdir().unwrap();
+ let root = format!("file://{}", tmp.path().display());
+ let io = FileIO::from_url(&root).unwrap().build().unwrap();
+ let options = HashMap::from([
+ (
+ "data-file.external-paths".to_string(),
+ format!("{root}/one,{root}/two"),
+ ),
+ (
+ "data-file.external-paths.strategy".to_string(),
+ "round-robin".to_string(),
+ ),
+ ("data-file.path-directory".to_string(), "data".to_string()),
+ ]);
+ let layout = HashIndexLayout {
+ table_path: &root,
+ partition_path: "p=a/",
+ data_file_path_directory: Some("data"),
+ index_file_in_data_file_dir: true,
+ };
+ let mut maintainer = DynamicBucketIndexMaintainer::new(vec![]);
+ let mut locations = Vec::new();
+ for hash in [1, 2, 3] {
+ maintainer.notify_new_record(hash);
+ let files = maintainer
+ .prepare_commit(&io, &layout, 0, &options)
+ .await
+ .unwrap();
+ let path = files[0].external_path.as_ref().unwrap();
+ assert!(path.contains("/data/p=a/bucket-0/index-"));
+ assert_eq!(
+ HashIndexFile::read(&io, path).await.unwrap().len(),
+ hash as usize
+ );
+ locations.push(path.rsplit_once("/data/").unwrap().0.to_string());
+ }
+ assert_ne!(locations[0], locations[1]);
+ assert_eq!(locations[0], locations[2]);
+ }
+
// -- PartitionIndex tests --
#[test]
fn test_assign_new_keys() {
let mut index = PartitionIndex::empty(3);
- assert_eq!(index.assign(100), 0);
- assert_eq!(index.assign(200), 0);
- assert_eq!(index.assign(300), 0);
+ assert_eq!(index.assign(100, -1).unwrap(), 0);
+ assert_eq!(index.assign(200, -1).unwrap(), 0);
+ assert_eq!(index.assign(300, -1).unwrap(), 0);
// Bucket 0 is full, next key goes to bucket 1
- assert_eq!(index.assign(400), 1);
+ assert_eq!(index.assign(400, -1).unwrap(), 1);
}
#[test]
fn test_assign_existing_key() {
let mut index = PartitionIndex::empty(10);
- assert_eq!(index.assign(42), 0);
+ assert_eq!(index.assign(42, -1).unwrap(), 0);
// Same hash returns same bucket
- assert_eq!(index.assign(42), 0);
+ assert_eq!(index.assign(42, -1).unwrap(), 0);
}
#[test]
fn test_multiple_buckets() {
let mut index = PartitionIndex::empty(2);
- assert_eq!(index.assign(1), 0);
- assert_eq!(index.assign(2), 0);
+ assert_eq!(index.assign(1, -1).unwrap(), 0);
+ assert_eq!(index.assign(2, -1).unwrap(), 0);
// Bucket 0 full
- assert_eq!(index.assign(3), 1);
- assert_eq!(index.assign(4), 1);
+ assert_eq!(index.assign(3, -1).unwrap(), 1);
+ assert_eq!(index.assign(4, -1).unwrap(), 1);
// Bucket 1 full
- assert_eq!(index.assign(5), 2);
+ assert_eq!(index.assign(5, -1).unwrap(), 2);
+ }
+
+ #[test]
+ fn test_dynamic_bucket_short_limit_and_explicit_reuse() {
+ let mut index = PartitionIndex::empty(1);
+ for bucket in 0..MAX_DYNAMIC_BUCKETS {
+ assert_eq!(index.assign(bucket, -1).unwrap(), bucket);
+ }
+ assert!(index
+ .assign(MAX_DYNAMIC_BUCKETS, -1)
+ .unwrap_err()
+ .to_string()
+ .contains("Short.MAX_VALUE"));
+ // Existing keys remain usable at the upper bound.
+ assert_eq!(
+ index.assign(MAX_DYNAMIC_BUCKETS - 1, -1).unwrap(),
+ MAX_DYNAMIC_BUCKETS - 1
+ );
+ let bucket = index
+ .assign(MAX_DYNAMIC_BUCKETS, MAX_DYNAMIC_BUCKETS)
+ .unwrap();
+ assert!((0..MAX_DYNAMIC_BUCKETS).contains(&bucket));
+ assert_eq!(
+ index
+ .assign(MAX_DYNAMIC_BUCKETS, MAX_DYNAMIC_BUCKETS)
+ .unwrap(),
+ bucket
+ );
+ }
+
+ #[tokio::test]
+ async fn test_dynamic_bucket_restore_reuses_unused_ids() {
+ let tmp = tempfile::tempdir().unwrap();
+ let root = format!("file://{}", tmp.path().display());
+ let io = FileIO::from_url(&root).unwrap().build().unwrap();
+ let meta = write_hash_index(&io, &format!("{root}/index"), &[42])
+ .await
+ .unwrap();
+ let entries = vec![IndexManifestEntry {
+ version: 1,
+ kind: crate::spec::FileKind::Add,
+ partition: EMPTY_SERIALIZED_ROW.to_vec(),
+ bucket: 2,
+ index_file: meta,
+ }];
+ let layout = HashIndexLayout {
+ table_path: &root,
+ partition_path: "",
+ data_file_path_directory: None,
+ index_file_in_data_file_dir: false,
+ };
+ let mut index = PartitionIndex::load(&io, &layout, &entries, 1)
+ .await
+ .unwrap();
+ assert_eq!(index.assign(42, 3).unwrap(), 2);
+ assert_eq!(index.assign(43, 3).unwrap(), 0);
+ assert_eq!(index.assign(44, 3).unwrap(), 1);
+ assert!((0..3).contains(&index.assign(45, 3).unwrap()));
+ }
+
+ #[tokio::test]
+ async fn test_dynamic_bucket_rejects_invalid_restored_indexes() {
+ let tmp = tempfile::tempdir().unwrap();
+ let root = format!("file://{}", tmp.path().display());
+ let io = FileIO::from_url(&root).unwrap().build().unwrap();
+ let meta = write_hash_index(&io, &format!("{root}/index"), &[42])
+ .await
+ .unwrap();
+ let entry = IndexManifestEntry {
+ version: 1,
+ kind: crate::spec::FileKind::Add,
+ partition: EMPTY_SERIALIZED_ROW.to_vec(),
+ bucket: 0,
+ index_file: meta,
+ };
+ let layout = HashIndexLayout {
+ table_path: &root,
+ partition_path: "",
+ data_file_path_directory: None,
+ index_file_in_data_file_dir: false,
+ };
+ for bucket in [-1, MAX_DYNAMIC_BUCKETS] {
+ let mut invalid = entry.clone();
+ invalid.bucket = bucket;
+ let result = PartitionIndex::load(&io, &layout, &[invalid],
1).await;
+ assert!(result
+ .err()
+ .unwrap()
+ .to_string()
+ .contains("Dynamic bucket id"));
+ }
+ let duplicate = vec![entry.clone(), entry.clone()];
+ assert!(PartitionIndex::load(&io, &layout, &duplicate, 1)
+ .await
+ .err()
+ .unwrap()
+ .to_string()
+ .contains("Multiple HASH"));
+ let mut conflicting = entry.clone();
+ conflicting.bucket = 1;
+ assert!(
+ PartitionIndex::load(&io, &layout, &[entry.clone(), conflicting],
1)
+ .await
+ .err()
+ .unwrap()
+ .to_string()
+ .contains("both buckets")
+ );
+ let mut truncated = entry.clone();
+ truncated.index_file.row_count = 2;
+ assert!(PartitionIndex::load(&io, &layout, &[truncated], 1)
+ .await
+ .err()
+ .unwrap()
+ .to_string()
+ .contains("expected 2 hashes"));
+ let path = layout.resolve(0, &entry.index_file.file_name, None);
+ io.new_output(&path)
+ .unwrap()
+ .write(bytes::Bytes::from_static(&[0, 0, 0]))
+ .await
+ .unwrap();
+ assert!(PartitionIndex::load(&io, &layout, &[entry], 1)
+ .await
+ .err()
+ .unwrap()
+ .to_string()
+ .contains("multiple of 4"));
+ }
+
+ #[tokio::test]
+ async fn test_dynamic_bucket_requires_configured_index_location() {
+ let tmp = tempfile::tempdir().unwrap();
+ let root = format!("file://{}", tmp.path().display());
+ let io = FileIO::from_url(&root).unwrap().build().unwrap();
+ let meta = write_hash_index(&io, &format!("{root}/index"), &[42])
+ .await
+ .unwrap();
+ let layout = HashIndexLayout {
+ table_path: &root,
+ partition_path: "p=a/",
+ data_file_path_directory: Some("data"),
+ index_file_in_data_file_dir: true,
+ };
+ let canonical = layout.resolve(0, &meta.file_name, None);
+ let entry = IndexManifestEntry {
+ version: 1,
+ kind: crate::spec::FileKind::Add,
+ partition: EMPTY_SERIALIZED_ROW.to_vec(),
+ bucket: 0,
+ index_file: meta,
+ };
+ let error = PartitionIndex::load(&io, &layout,
std::slice::from_ref(&entry), 1)
+ .await
+ .err()
+ .unwrap();
+ assert!(error.to_string().contains(&entry.index_file.file_name));
+ io.new_output(&canonical)
+ .unwrap()
+ .write(bytes::Bytes::from_static(&[0, 0, 0, 43]))
+ .await
+ .unwrap();
+ let index = PartitionIndex::load(&io, &layout, &[entry], 1)
+ .await
+ .unwrap();
+ assert_eq!(index.hash_to_bucket.get(&43), Some(&0));
+ assert!(!index.hash_to_bucket.contains_key(&42));
}
// -- HashIndexFile tests --
@@ -670,7 +1078,7 @@ mod tests {
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 meta = write_hash_index(&file_io, &dir,
&hashes).await.unwrap();
let entries = vec![IndexManifestEntry {
version: 1,
kind: crate::spec::FileKind::Add,
@@ -702,7 +1110,7 @@ mod tests {
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])
+ let mut index_file = write_hash_index(&file_io, &external_dir,
&[42i32])
.await
.unwrap();
index_file.external_path = Some(format!("{external_dir}/{}",
index_file.file_name));
@@ -735,7 +1143,7 @@ mod tests {
let file_io = FileIO::from_url(&dir).unwrap().build().unwrap();
let hashes = vec![42, -1, 0, i32::MAX, i32::MIN];
- let meta = HashIndexFile::write(&file_io, &dir,
&hashes).await.unwrap();
+ let meta = write_hash_index(&file_io, &dir, &hashes).await.unwrap();
assert_eq!(meta.index_type, HASH_INDEX);
assert_eq!(meta.row_count, 5);
@@ -752,7 +1160,7 @@ mod tests {
let dir = format!("file://{}", tmp.path().display());
let file_io = FileIO::from_url(&dir).unwrap().build().unwrap();
- let meta = HashIndexFile::write(&file_io, &dir, &[]).await.unwrap();
+ let meta = write_hash_index(&file_io, &dir, &[]).await.unwrap();
assert_eq!(meta.row_count, 0);
assert_eq!(meta.file_size, 0);
diff --git a/crates/paimon/src/table/data_evolution_writer.rs
b/crates/paimon/src/table/data_evolution_writer.rs
index 7976bbf3..760f07ba 100644
--- a/crates/paimon/src/table/data_evolution_writer.rs
+++ b/crates/paimon/src/table/data_evolution_writer.rs
@@ -770,7 +770,7 @@ impl DataEvolutionDeleteWriter {
let mut bitmaps = IndexMap::new();
let mut deleted_index_files = Vec::new();
- for mut entry in index_entries {
+ for entry in index_entries {
if entry.kind != FileKind::Add
|| entry.bucket != bucket
|| entry.partition != partition
@@ -779,12 +779,6 @@ impl DataEvolutionDeleteWriter {
continue;
}
deleted_index_files.push(entry.index_file.clone());
- // Preserve the original manifest identity in deleted_index_files.
- // Only the copy used for reading acquires the legacy physical
path.
- layout
- .location()
- .resolve_legacy_deletion_vector(self.table.file_io(), &mut
entry.index_file)
- .await?;
let Some(ranges) =
entry.index_file.deletion_vectors_ranges.as_ref() else {
continue;
};
diff --git a/crates/paimon/src/table/index_file_path.rs
b/crates/paimon/src/table/index_file_path.rs
index 8523e03e..0b740385 100644
--- a/crates/paimon/src/table/index_file_path.rs
+++ b/crates/paimon/src/table/index_file_path.rs
@@ -34,9 +34,7 @@
//! index kind it reads. Cleanup after a failed commit does not — see
//! [`committed_index_file_path`].
-use crate::io::FileIO;
-use crate::spec::{bucket_path, BinaryRow, IndexFileMeta, IndexManifestEntry,
PartitionComputer};
-use crate::table::Table;
+use crate::spec::IndexFileMeta;
const INDEX_DIR: &str = "index";
@@ -80,35 +78,6 @@ impl IndexFileLocation<'_> {
}
}
- /// Older Python DV writers ignored the bucket-directory option. Resolve
- /// their existing files without changing the manifest or masking missing
- /// canonical files with a path that does not exist either.
- pub(super) async fn resolve_legacy_deletion_vector(
- &self,
- file_io: &FileIO,
- file: &mut IndexFileMeta,
- ) -> crate::Result<()> {
- let Self::BucketLocal {
- table_path,
- index_file_in_data_file_dir: true,
- ..
- } = self
- else {
- return Ok(());
- };
- if file.index_type != "DELETION_VECTORS" ||
file.external_path.is_some() {
- return Ok(());
- }
- let canonical_path = self.resolve(&file.file_name, None);
- if !file_io.exists(&canonical_path).await? {
- let legacy_path = format!("{table_path}/{INDEX_DIR}/{}",
file.file_name);
- if file_io.exists(&legacy_path).await? {
- file.external_path = Some(legacy_path);
- }
- }
- Ok(())
- }
-
/// 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 {
@@ -119,53 +88,6 @@ impl IndexFileLocation<'_> {
}
}
-/// Resolve each retained DV index once, before either split planning or a
-/// global-index evaluator consumes it. Explicit paths and ordinary table/index
-/// layouts require no additional filesystem requests.
-pub(crate) async fn resolve_legacy_deletion_vector_entries(
- table: &Table,
- entries: &mut [IndexManifestEntry],
-) -> crate::Result<()> {
- let schema = table.schema();
- let options = schema.core_options();
- if !options.index_file_in_data_file_dir() {
- return Ok(());
- }
- let partition_computer = if schema.partition_keys().is_empty() {
- None
- } else {
- Some(PartitionComputer::new(
- schema.partition_keys(),
- schema.fields(),
- options.partition_default_name(),
- options.legacy_partition_name(),
- )?)
- };
- let table_path = table.location().trim_end_matches('/');
- for entry in entries {
- if entry.index_file.index_type != "DELETION_VECTORS"
- || entry.index_file.external_path.is_some()
- {
- continue;
- }
- let partition = BinaryRow::from_serialized_bytes(&entry.partition)?;
- let bucket_path = bucket_path(
- &table.data_file_location(),
- partition_computer.as_ref(),
- &partition,
- entry.bucket,
- )?;
- IndexFileLocation::BucketLocal {
- table_path,
- bucket_path: &bucket_path,
- index_file_in_data_file_dir: true,
- }
- .resolve_legacy_deletion_vector(table.file_io(), &mut entry.index_file)
- .await?;
- }
- Ok(())
-}
-
/// `"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.
@@ -312,94 +234,6 @@ mod tests {
}
}
- #[tokio::test]
- async fn timestamp_dvs_use_canonical_paths_and_explicit_precedence() {
- use crate::catalog::Identifier;
- use crate::io::FileIOBuilder;
- use crate::spec::{
- bucket_path_under, BinaryRowBuilder, DataType, FileKind, Schema,
TableSchema,
- TimestampType,
- };
- for (legacy, millis, java_partition) in [
- (false, 0, "ts=1970-01-01 00%3A00%3A00.000/"),
- (true, 100, "ts=1970-01-01T00%3A00%3A00.100/"),
- ] {
- let dir = tempfile::tempdir().unwrap();
- let table_path = dir.path().to_str().unwrap();
- let schema = Schema::builder()
- .column("ts",
DataType::Timestamp(TimestampType::new(3).unwrap()))
- .partition_keys(vec!["ts".to_string()])
- .option("partition.legacy-name", legacy.to_string())
- .option("index-file-in-data-file-dir", "true")
- .build()
- .unwrap();
- let schema = TableSchema::new(0, &schema);
- let file_io = FileIOBuilder::new("file").build().unwrap();
- let table = Table::new(
- file_io,
- Identifier::new("db", "t"),
- table_path.to_string(),
- schema,
- None,
- );
- let java_bucket = bucket_path_under(table_path, java_partition, 7);
- let table_index = format!("{table_path}/index");
- for path in [&java_bucket, &table_index] {
- std::fs::create_dir_all(path).unwrap();
- }
- for (parent, name, bytes) in [
- (&java_bucket, "idx-java", b"Java".as_slice()),
- (&table_index, "idx-java", b"wrong table copy".as_slice()),
- (&table_index, "idx-table", b"table".as_slice()),
- (&java_bucket, "idx-explicit", b"wrong Java copy".as_slice()),
- (&table_index, "idx-explicit", b"wrong table copy".as_slice()),
- ] {
- std::fs::write(format!("{parent}/{name}"), bytes).unwrap();
- }
- let mut row = BinaryRowBuilder::new(1);
- row.write_timestamp_compact(0, millis);
- let partition = row.build_serialized();
- let mut entries: Vec<_> = ["idx-java", "idx-table",
"idx-explicit", "idx-missing"]
- .into_iter()
- .map(|name| {
- let mut file = committed_file(None);
- file.file_name = name.to_string();
- file.index_type = "DELETION_VECTORS".to_string();
- IndexManifestEntry {
- version: 1,
- kind: FileKind::Add,
- partition: partition.clone(),
- bucket: 7,
- index_file: file,
- }
- })
- .collect();
- let explicit =
format!("{table_path}/missing-explicit/idx-explicit");
- entries[2].index_file.external_path = Some(explicit.clone());
- resolve_legacy_deletion_vector_entries(&table, &mut entries)
- .await
- .unwrap();
- assert_eq!(entries[0].index_file.external_path, None);
- assert_eq!(
- entries[1].index_file.external_path,
- Some(format!("{table_index}/idx-table"))
- );
- assert_eq!(entries[2].index_file.external_path,
Some(explicit.clone()));
- assert_eq!(entries[3].index_file.external_path, None);
- let location = IndexFileLocation::BucketLocal {
- table_path,
- bucket_path: &java_bucket,
- index_file_in_data_file_dir: true,
- };
- for (entry, expected) in
entries.iter().take(2).zip([b"Java".as_slice(), b"table"]) {
- let file = &entry.index_file;
- let path = location.resolve(&file.file_name,
file.external_path.as_deref());
- assert_eq!(std::fs::read(path).unwrap(), expected);
- }
- assert!(!table.file_io().exists(&explicit).await.unwrap());
- }
- }
-
/// A committed index file with the given `_GLOBAL_INDEX` and
`_SOURCE_META`.
fn committed_file(global_index_meta: Option<Option<Vec<u8>>>) ->
IndexFileMeta {
IndexFileMeta {
@@ -420,75 +254,27 @@ mod tests {
}
}
- #[tokio::test]
- async fn
legacy_dv_directory_fallback_preserves_canonical_and_external_precedence() {
- use crate::io::FileIOBuilder;
-
+ #[test]
+ fn missing_bucket_index_does_not_use_table_directory() {
let dir = tempfile::tempdir().unwrap();
let table_path = dir.path().to_str().unwrap();
let bucket_path = format!("{table_path}/pt=1/bucket-7");
- let legacy_path = format!("{table_path}/index/idx-0");
- let canonical_path = format!("{bucket_path}/idx-0");
- let external_path = format!("{table_path}/external/idx-0");
- for parent in [
- format!("{table_path}/index"),
- bucket_path.clone(),
- format!("{table_path}/external"),
- ] {
- std::fs::create_dir_all(parent).unwrap();
+ let table_index = format!("{table_path}/index/idx-0");
+ std::fs::create_dir_all(format!("{table_path}/index")).unwrap();
+ std::fs::write(&table_index, b"wrong directory").unwrap();
+ for index_type in ["DELETION_VECTORS", "HASH"] {
+ let mut file = committed_file(None);
+ file.index_type = index_type.to_string();
+ let path = committed_index_file_path(table_path, &bucket_path,
true, &file);
+ assert_eq!(path, format!("{bucket_path}/idx-0"));
+ assert_eq!(
+ std::fs::read(path).unwrap_err().kind(),
+ std::io::ErrorKind::NotFound
+ );
+ file.external_path = Some(table_index.clone());
+ let path = committed_index_file_path(table_path, &bucket_path,
true, &file);
+ assert_eq!(std::fs::read(path).unwrap(), b"wrong directory");
}
- let file_io = FileIOBuilder::new("file").build().unwrap();
- let location = IndexFileLocation::BucketLocal {
- table_path,
- bucket_path: &bucket_path,
- index_file_in_data_file_dir: true,
- };
- let mut original = committed_file(None);
- original.index_type = "DELETION_VECTORS".to_string();
-
- // Missing files retain the canonical path and therefore fail on read.
- let mut missing = original.clone();
- location
- .resolve_legacy_deletion_vector(&file_io, &mut missing)
- .await
- .unwrap();
- assert_eq!(missing.external_path, None);
-
- std::fs::write(&legacy_path, b"legacy DV").unwrap();
- let mut legacy = original.clone();
- location
- .resolve_legacy_deletion_vector(&file_io, &mut legacy)
- .await
- .unwrap();
- assert_eq!(legacy.external_path.as_deref(),
Some(legacy_path.as_str()));
-
- std::fs::write(&canonical_path, b"canonical DV").unwrap();
- let mut canonical = original.clone();
- location
- .resolve_legacy_deletion_vector(&file_io, &mut canonical)
- .await
- .unwrap();
- assert_eq!(canonical.external_path, None);
- assert_eq!(
- std::fs::read(location.resolve("idx-0", None)).unwrap(),
- b"canonical DV"
- );
-
- // An explicit path is never substituted, even when the target is
absent.
- original.external_path = Some(external_path.clone());
- location
- .resolve_legacy_deletion_vector(&file_io, &mut original)
- .await
- .unwrap();
- assert_eq!(original.external_path, Some(external_path));
-
- std::fs::remove_file(&canonical_path).unwrap();
- let mut other_index = committed_file(None);
- location
- .resolve_legacy_deletion_vector(&file_io, &mut other_index)
- .await
- .unwrap();
- assert_eq!(other_index.external_path, None);
}
fn committed_path(file: &IndexFileMeta) -> String {
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 6205cd11..cf88f85d 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -1804,13 +1804,6 @@ impl<'a> PaimonTableScan<'a> {
entries.push(entry);
}
}
- if deletion_vectors_needed {
- super::index_file_path::resolve_legacy_deletion_vector_entries(
- self.table,
- &mut entries,
- )
- .await?;
- }
Ok(Some(entries))
}
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 22a42356..3325a9c6 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -411,23 +411,13 @@ impl TableWrite {
merge_engine,
)))
} else if is_dynamic_bucket {
- BucketAssignerEnum::Dynamic(Box::new(
- DynamicBucketAssigner::new(
- partition_field_indices,
- primary_key_indices.clone(),
- schema.fields().to_vec(),
- target_bucket_row_number,
- 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(),
- )
-
.with_data_file_path_directory(core_options.data_file_path_directory()),
- ))
+ BucketAssignerEnum::Dynamic(Box::new(DynamicBucketAssigner::new(
+ table.clone(),
+ partition_field_indices,
+ primary_key_indices.clone(),
+ is_overwrite,
+ partition_computer.clone(),
+ )?))
} else if total_buckets == POSTPONE_BUCKET {
BucketAssignerEnum::Constant(ConstantBucketAssigner::new(
partition_field_indices,
@@ -605,7 +595,13 @@ impl TableWrite {
return Ok(());
};
- let grouped = self.divide_by_partition_bucket(&batch).await?;
+ let grouped = match self.divide_by_partition_bucket(&batch).await {
+ Ok(grouped) => grouped,
+ Err(error) => {
+ self.fail_write().await;
+ return Err(error);
+ }
+ };
for ((partition_bytes, bucket), sub_batch) in grouped {
self.write_bucket(partition_bytes, bucket, sub_batch)
.await?;
@@ -1008,12 +1004,18 @@ impl TableWrite {
}
.await;
if result.is_err() {
- self.close().await;
- self.failed = true;
+ self.fail_write().await;
}
result
}
+ /// Routing can update persistent bucket-index state before it fails, just
+ /// as file writes can produce output. Neither failure permits a later
commit.
+ async fn fail_write(&mut self) {
+ self.close().await;
+ self.failed = true;
+ }
+
fn ensure_active(&self) -> Result<()> {
if self.failed {
return Err(crate::Error::DataInvalid {
diff --git a/crates/paimon/tests/dynamic_bucket_write_test.rs
b/crates/paimon/tests/dynamic_bucket_write_test.rs
new file mode 100644
index 00000000..8e29704a
--- /dev/null
+++ b/crates/paimon/tests/dynamic_bucket_write_test.rs
@@ -0,0 +1,202 @@
+// 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.
+
+mod common;
+
+use common::incremental_helpers::{make_batch, memory_table, pk_schema,
setup_dirs};
+
+#[tokio::test]
+async fn dynamic_bucket_limit_survives_writer_restart() {
+ let (io, table) = memory_table(
+ "memory:/dynamic_limit",
+ pk_schema(&[
+ ("bucket", "-1"),
+ ("dynamic-bucket.target-row-num", "1"),
+ ("dynamic-bucket.max-buckets", "1"),
+ ]),
+ );
+ setup_dirs(&io, table.location()).await;
+ for ids in [vec![1, 2, 3], vec![3, 4, 5]] {
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&make_batch(ids.clone(), ids))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].bucket, 0);
+ builder.new_commit().commit(messages).await.unwrap();
+ }
+}
+
+#[tokio::test]
+async fn dynamic_write_rejects_data_without_complete_hash_index() {
+ let (io, table) = memory_table(
+ "memory:/dynamic_missing",
+ pk_schema(&[("bucket", "-1"), ("dynamic-bucket.target-row-num", "1")]),
+ );
+ setup_dirs(&io, table.location()).await;
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&make_batch(vec![1, 2], vec![10, 20]))
+ .await
+ .unwrap();
+ let mut messages = writer.prepare_commit().await.unwrap();
+ for message in &mut messages {
+ message.new_index_files.clear();
+ }
+ builder.new_commit().commit(messages).await.unwrap();
+ let mut writer = table.new_write_builder().new_write().unwrap();
+ let error = writer
+ .write_arrow_batch(&make_batch(vec![2], vec![200]))
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("complete HASH index"), "{error}");
+ assert!(writer
+ .prepare_commit()
+ .await
+ .unwrap_err()
+ .to_string()
+ .contains("cannot be reused"));
+ assert!(writer
+ .write_arrow_batch(&make_batch(vec![3], vec![30]))
+ .await
+ .unwrap_err()
+ .to_string()
+ .contains("cannot be reused"));
+}
+
+#[test]
+fn dynamic_write_rejects_invalid_max_buckets() {
+ for value in ["0", "-2", "32769", "2147483648", "invalid"] {
+ let (_, table) = memory_table(
+ "memory:/dynamic_invalid",
+ pk_schema(&[("bucket", "-1"), ("dynamic-bucket.max-buckets",
value)]),
+ );
+ let result = table.new_write_builder().new_write();
+ assert!(result.is_err(), "accepted max-buckets={value}");
+ assert!(result
+ .err()
+ .unwrap()
+ .to_string()
+ .contains("dynamic-bucket.max-buckets"));
+ }
+}
+
+#[test]
+fn dynamic_write_rejects_custom_bucket_keys() {
+ for key in ["id", ""] {
+ let (_, table) = memory_table(
+ "memory:/dynamic_key",
+ pk_schema(&[("bucket", "-1"), ("bucket-key", key)]),
+ );
+ let error = table.new_write_builder().new_write().err().unwrap();
+ assert!(error.to_string().contains("Cannot define 'bucket-key'"));
+ }
+}
+
+/// Inject failure after at least one bucket HASH was written, independently of
+/// HashMap iteration order. Record physical data/index paths to verify
ownership.
+#[derive(Debug)]
+struct FailSecondHash {
+ operator: opendal::Operator,
+ indexes: std::sync::Mutex<Vec<String>>,
+ data: std::sync::Mutex<Vec<String>>,
+}
+
+#[async_trait::async_trait]
+impl paimon::io::FileIOProvider for FailSecondHash {
+ async fn create(&self, path: &str) -> paimon::Result<(opendal::Operator,
String)> {
+ let path = path.strip_prefix("file://").unwrap_or(path);
+ let name = path.rsplit('/').next().unwrap();
+ if name.starts_with("index-") {
+ let mut indexes = self.indexes.lock().unwrap();
+ if !indexes.iter().any(|item| item == path) {
+ indexes.push(path.to_string());
+ }
+ if indexes.get(1).is_some_and(|item| item == path) {
+ return Err(paimon::Error::DataInvalid {
+ message: "injected HASH write failure".into(),
+ source: None,
+ });
+ }
+ }
+ if name.ends_with(".parquet") {
+ self.data.lock().unwrap().push(path.to_string());
+ }
+ Ok((
+ self.operator.clone(),
+ path.trim_start_matches('/').to_string(),
+ ))
+ }
+}
+
+#[tokio::test]
+async fn failed_hash_prepare_cleans_indexes_and_data() {
+ use paimon::catalog::Identifier;
+ use paimon::io::FileIOBuilder;
+ use paimon::table::Table;
+ use std::sync::Arc;
+
+ let tmp = tempfile::tempdir().unwrap();
+ let mut config = opendal_service_fs::FsConfig::default();
+ config.root = Some("/".into());
+ let provider = Arc::new(FailSecondHash {
+ operator: opendal::Operator::from_config(config).unwrap(),
+ indexes: Default::default(),
+ data: Default::default(),
+ });
+ let io = FileIOBuilder::new("file")
+ .with_provider(provider.clone())
+ .build()
+ .unwrap();
+ let path = tmp.path().to_str().unwrap();
+ let table = Table::new(
+ io.clone(),
+ Identifier::new("default", "failure"),
+ path.into(),
+ pk_schema(&[("bucket", "-1"), ("dynamic-bucket.target-row-num", "1")]),
+ None,
+ );
+ setup_dirs(&io, path).await;
+ let mut writer = table.new_write_builder().new_write().unwrap();
+ writer
+ .write_arrow_batch(&make_batch(vec![1, 2], vec![10, 20]))
+ .await
+ .unwrap();
+ let error = writer.prepare_commit().await.unwrap_err();
+ assert!(
+ error.to_string().contains("injected HASH write failure"),
+ "{error}"
+ );
+ let indexes = provider.indexes.lock().unwrap().clone();
+ assert_eq!(indexes.len(), 2);
+ assert!(indexes
+ .iter()
+ .all(|path| !std::path::Path::new(path).exists()));
+ let data = provider.data.lock().unwrap().clone();
+ assert!(!data.is_empty());
+ assert!(data.iter().all(|path| !std::path::Path::new(path).exists()));
+ assert!(writer
+ .prepare_commit()
+ .await
+ .unwrap_err()
+ .to_string()
+ .contains("cannot be reused"));
+}
diff --git a/crates/paimon/tests/scan_planning_parity_test.rs
b/crates/paimon/tests/scan_planning_parity_test.rs
index a5836aaf..f08134db 100644
--- a/crates/paimon/tests/scan_planning_parity_test.rs
+++ b/crates/paimon/tests/scan_planning_parity_test.rs
@@ -734,9 +734,14 @@ async fn
float_primary_key_snapshot_merges_versions_while_delta_keeps_events() {
}
#[tokio::test]
-async fn
python_dv_manifest_layout_resolves_legacy_canonical_and_external_files() {
+async fn python_dv_manifest_resolves_java_index_locations() {
use paimon::spec::IndexManifest;
- for layout in ["legacy", "canonical", "external", "missing-external"] {
+ for layout in [
+ "missing-canonical",
+ "canonical",
+ "external",
+ "missing-external",
+ ] {
let table = evolution_table_with_options(
"memory:/planning_parity/python_dv",
&[
@@ -773,7 +778,7 @@ async fn
python_dv_manifest_layout_resolves_legacy_canonical_and_external_files(
.copy_file(&source, &canonical)
.await
.unwrap();
- let legacy = format!(
+ let table_index = format!(
"{}/index/{}",
table.location(),
entries[0].index_file.file_name
@@ -781,22 +786,22 @@ async fn
python_dv_manifest_layout_resolves_legacy_canonical_and_external_files(
let external = format!("{}/external/dv", table.location());
table
.file_io()
- .copy_file(&canonical, &legacy)
+ .copy_file(&canonical, &table_index)
.await
.unwrap();
let expected_path = match layout {
- "legacy" => {
+ "missing-canonical" => {
table.file_io().delete_file(&canonical).await.unwrap();
entries[0].index_file.external_path = None;
- legacy
+ canonical
}
"canonical" => {
- // The canonical file must win even if a stale legacy file
exists.
+ // A same-name file in table/index must not affect
bucket-local reads.
table
.file_io()
- .new_output(&legacy)
+ .new_output(&table_index)
.unwrap()
- .write(bytes::Bytes::from_static(b"stale legacy DV"))
+ .write(bytes::Bytes::from_static(b"wrong index directory"))
.await
.unwrap();
entries[0].index_file.external_path = None;
@@ -855,11 +860,15 @@ async fn
python_dv_manifest_layout_resolves_legacy_canonical_and_external_files(
.unwrap()
.try_collect::<Vec<RecordBatch>>()
.await;
- if layout == "missing-external" {
+ if matches!(layout, "missing-canonical" | "missing-external") {
+ let error = batches.expect_err("a missing DV must not fall back to
another directory");
assert!(
- batches.is_err(),
- "an explicit missing DV must not fall back to a local copy"
+ error
+ .to_string()
+ .contains(expected_path.trim_start_matches("memory:/")),
+ "{error}"
);
+ assert!(table.file_io().exists(&table_index).await.unwrap());
} else {
assert!(batches.is_ok(), "layout={layout}: {batches:?}");
assert_eq!(read_ids(&table, &plan).await, vec![0, 2]);
diff --git a/crates/paimon/tests/table_update_paths_test.rs
b/crates/paimon/tests/table_update_paths_test.rs
index 2c7590d0..18cded19 100644
--- a/crates/paimon/tests/table_update_paths_test.rs
+++ b/crates/paimon/tests/table_update_paths_test.rs
@@ -420,7 +420,7 @@ async fn
external_deletion_vectors_repeat_time_travel_and_abort() {
}
#[tokio::test]
-async fn
delete_reads_legacy_index_directory_without_changing_manifest_identity() {
+async fn delete_requires_bucket_index_and_preserves_manifest_on_failure() {
let table = table(&[("index-file-in-data-file-dir", "true")]).await;
seed(&table).await;
let ids = row_ids(&table).await;
@@ -428,7 +428,7 @@ async fn
delete_reads_legacy_index_directory_without_changing_manifest_identity(
let old = messages[0].new_index_files[0].clone();
assert!(old.external_path.is_none());
let bucket_path = format!("{}/p=a/q=1/bucket-0/{}", table.location(),
old.file_name);
- let legacy_path = format!("{}/index/{}", table.location(), old.file_name);
+ let table_index_path = format!("{}/index/{}", table.location(),
old.file_name);
let bytes = table
.file_io()
.new_input(&bucket_path)
@@ -438,18 +438,60 @@ async fn
delete_reads_legacy_index_directory_without_changing_manifest_identity(
.unwrap();
table
.file_io()
- .new_output(&legacy_path)
+ .new_output(&table_index_path)
.unwrap()
- .write(bytes)
+ .write(bytes.clone())
.await
.unwrap();
table.file_io().delete_file(&bucket_path).await.unwrap();
commit(&table, messages).await;
+
+ let mut writer = table.new_write_builder().new_delete().unwrap();
+ writer.add_row_ids(vec![ids[&1]]).unwrap();
+ let before = files(&table).await;
+ let error = writer.prepare_commit().await.unwrap_err();
+ assert!(
+ error
+ .to_string()
+ .contains(bucket_path.trim_start_matches("memory:/")),
+ "{error}"
+ );
+ assert_eq!(files(&table).await, before);
+ assert_eq!(
+ table
+ .snapshot_manager()
+ .get_latest_snapshot_id()
+ .await
+ .unwrap(),
+ Some(2)
+ );
+ let snapshot = table.snapshot_manager().get_snapshot(2).await.unwrap();
+ let entries = paimon::spec::IndexManifest::read(
+ table.file_io(),
+ &format!(
+ "{}/manifest/{}",
+ table.location(),
+ snapshot.index_manifest().unwrap()
+ ),
+ )
+ .await
+ .unwrap();
+ assert_eq!(entries.len(), 1);
+ assert_eq!(entries[0].index_file, old);
+
+ // Restoring the required Java location allows a fresh writer to proceed.
+ table
+ .file_io()
+ .new_output(&bucket_path)
+ .unwrap()
+ .write(bytes)
+ .await
+ .unwrap();
let messages = delete(&table, vec![ids[&1]]).await;
assert_eq!(messages[0].deleted_index_files, vec![old]);
commit(&table, messages).await;
assert_eq!(row_ids(&table).await.len(), 2);
- assert!(table.file_io().exists(&legacy_path).await.unwrap());
+ assert!(table.file_io().exists(&table_index_path).await.unwrap());
let historical = table
.copy_with_time_travel(HashMap::from([("scan.snapshot-id".into(),
"2".into())]))
.await
