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 fc27163e [core] Honor external data paths across table writers (#981)
fc27163e is described below
commit fc27163eef8f999aa1f52e50e405233902fd09c5
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Sep 28 21:33:49 2026 +0800
[core] Honor external data paths across table writers (#981)
---
crates/paimon/src/spec/schema.rs | 2 +-
crates/paimon/src/table/cow_writer.rs | 2 +-
crates/paimon/src/table/data_evolution_writer.rs | 2 +-
crates/paimon/src/table/data_file_path_factory.rs | 90 +++++
crates/paimon/src/table/data_file_writer.rs | 210 ++++++-----
.../src/table/dedicated_format_file_writer.rs | 125 ++++++-
crates/paimon/src/table/external_path.rs | 82 +++--
crates/paimon/src/table/kv_file_writer.rs | 41 +--
crates/paimon/src/table/mod.rs | 1 +
crates/paimon/src/table/postpone_file_writer.rs | 48 +--
crates/paimon/src/table/table_write.rs | 22 +-
crates/paimon/tests/external_data_write_test.rs | 383 +++++++++++++++++++++
12 files changed, 834 insertions(+), 174 deletions(-)
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index 7386a625..c00b3e17 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -1580,7 +1580,7 @@ impl Schema {
Ok(())
}
- fn validate_primary_key_blob_configuration(
+ pub(crate) fn validate_primary_key_blob_configuration(
fields: &[DataField],
primary_keys: &[String],
options: &HashMap<String, String>,
diff --git a/crates/paimon/src/table/cow_writer.rs
b/crates/paimon/src/table/cow_writer.rs
index b076207e..a0ff4304 100644
--- a/crates/paimon/src/table/cow_writer.rs
+++ b/crates/paimon/src/table/cow_writer.rs
@@ -320,7 +320,7 @@ impl CopyOnWriteMergeWriter {
Some(0),
None,
None,
- )
+ )?
.with_target_file_row_num(target_file_row_num);
writer.write(&rewritten).await?;
writer.prepare_commit().await?
diff --git a/crates/paimon/src/table/data_evolution_writer.rs
b/crates/paimon/src/table/data_evolution_writer.rs
index 85e21762..cbe0b142 100644
--- a/crates/paimon/src/table/data_evolution_writer.rs
+++ b/crates/paimon/src/table/data_evolution_writer.rs
@@ -1557,7 +1557,7 @@ impl DataEvolutionPartialWriter {
Some(0), // file_source: APPEND
Some(first_row_id),
Some(write_set.write_columns.clone()),
- )
+ )?
.with_file_index(write_set.file_index_options.clone());
self.writers.insert(key.clone(), writer);
}
diff --git a/crates/paimon/src/table/data_file_path_factory.rs
b/crates/paimon/src/table/data_file_path_factory.rs
new file mode 100644
index 00000000..2695ab78
--- /dev/null
+++ b/crates/paimon/src/table/data_file_path_factory.rs
@@ -0,0 +1,90 @@
+// 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.
+
+//! Java-style per-bucket file placement shared by all physical writers.
+
+use super::external_path::ExternalPathProvider;
+use crate::spec::{bucket_path_under, data_file_path, relative_bucket_path,
CoreOptions};
+use crate::Result;
+use std::collections::HashMap;
+use std::sync::Mutex;
+
+/// Select a location once per new data/changelog/Blob file. Sidecars use the
+/// selected file's parent; they must never advance the external-path provider.
+pub(super) struct DataFilePathFactory {
+ bucket_path: String,
+ external: Option<Mutex<ExternalPathProvider>>,
+}
+
+pub(super) struct DataFilePath {
+ pub path: String,
+ pub external_path: Option<String>,
+}
+
+impl DataFilePath {
+ pub fn parent(&self) -> &str {
+ self.path
+ .rsplit_once('/')
+ .expect("data file has a parent")
+ .0
+ }
+}
+
+impl DataFilePathFactory {
+ pub fn new(
+ table: &str,
+ partition: &str,
+ bucket: i32,
+ options: &HashMap<String, String>,
+ ) -> Result<Self> {
+ let options_view = CoreOptions::new(options);
+ let directory = options_view.data_file_path_directory();
+ let bucket_path = bucket_path_under(&data_file_path(table, directory),
partition, bucket);
+ let relative = relative_bucket_path(partition, bucket, directory);
+ Ok(Self {
+ bucket_path,
+ external: ExternalPathProvider::new(options,
&relative)?.map(Mutex::new),
+ })
+ }
+
+ pub fn bucket_path(&self) -> &str {
+ &self.bucket_path
+ }
+
+ pub fn new_path(&self, name: &str) -> Result<DataFilePath> {
+ let external_path = self
+ .external
+ .as_ref()
+ .map(|provider| {
+ provider
+ .lock()
+ .map_err(|_| crate::Error::UnexpectedError {
+ message: "External path provider lock is
poisoned".into(),
+ source: None,
+ })
+ .map(|mut provider| provider.next_path(name))
+ })
+ .transpose()?;
+ let path = external_path
+ .clone()
+ .unwrap_or_else(|| format!("{}/{name}", self.bucket_path));
+ Ok(DataFilePath {
+ path,
+ external_path,
+ })
+ }
+}
diff --git a/crates/paimon/src/table/data_file_writer.rs
b/crates/paimon/src/table/data_file_writer.rs
index d2c01381..62b7da25 100644
--- a/crates/paimon/src/table/data_file_writer.rs
+++ b/crates/paimon/src/table/data_file_writer.rs
@@ -23,6 +23,7 @@
//! [`DataFileMeta`] for the commit path.
use super::data_file_index_writer::{DataFileIndexWriter, FileIndexOptions};
+use super::data_file_path_factory::{DataFilePath, DataFilePathFactory};
use crate::arrow::format::{
create_format_writer, with_write_resources, FormatFileWriter,
FormatValueStats,
};
@@ -30,7 +31,7 @@ use crate::io::FileIO;
use crate::resource::ResourceContext;
use crate::spec::data_file_to_file_index_file_name;
use crate::spec::stats::BinaryTableStats;
-use crate::spec::{bucket_path_under, CoreOptions, DataField, DataFileMeta,
EMPTY_SERIALIZED_ROW};
+use crate::spec::{CoreOptions, DataField, DataFileMeta, EMPTY_SERIALIZED_ROW};
use crate::Result;
use arrow_array::RecordBatch;
use chrono::Utc;
@@ -47,9 +48,7 @@ use tokio::task::JoinSet;
/// Call [`prepare_commit`](Self::prepare_commit) to finalize and collect file
metadata.
pub(crate) struct DataFileWriter {
file_io: FileIO,
- table_location: String,
- partition_path: String,
- bucket: i32,
+ paths: Arc<DataFilePathFactory>,
schema_id: i64,
target_file_size: i64,
target_file_row_num: i64,
@@ -70,6 +69,7 @@ pub(crate) struct DataFileWriter {
/// Current open format writer, lazily created on first write.
current_writer: Option<Box<dyn FormatFileWriter>>,
current_file_name: Option<String>,
+ current_file_path: Option<DataFilePath>,
current_row_count: i64,
index_options: Option<Arc<FileIndexOptions>>,
current_index: Option<DataFileIndexWriter>,
@@ -96,15 +96,19 @@ impl DataFileWriter {
file_source: Option<i32>,
first_row_id: Option<i64>,
write_cols: Option<Vec<String>>,
- ) -> Self {
+ ) -> Result<Self> {
+ let paths = Arc::new(DataFilePathFactory::new(
+ &table_location,
+ &partition_path,
+ bucket,
+ &format_options,
+ )?);
let data_file_prefix = CoreOptions::new(&format_options)
.data_file_prefix()
.to_string();
- Self {
+ Ok(Self {
file_io,
- table_location,
- partition_path,
- bucket,
+ paths,
schema_id,
target_file_size,
target_file_row_num: i64::MAX,
@@ -123,12 +127,18 @@ impl DataFileWriter {
in_flight_closes: JoinSet::new(),
current_writer: None,
current_file_name: None,
+ current_file_path: None,
current_row_count: 0,
index_options: None,
current_index: None,
resources: None,
created_paths: Vec::new(),
- }
+ })
+ }
+
+ pub(super) fn with_path_factory(mut self, paths: Arc<DataFilePathFactory>)
-> Self {
+ self.paths = paths;
+ self
}
pub(super) fn with_file_index(mut self, options:
Option<Arc<FileIndexOptions>>) -> Self {
@@ -205,10 +215,11 @@ impl DataFileWriter {
self.next_file_ordinal,
self.file_format,
);
- let bucket_dir = self.bucket_dir();
+ let location = self.paths.new_path(&file_name)?;
+ let bucket_dir = location.parent();
self.file_io.mkdirs(&format!("{bucket_dir}/")).await?;
- let file_path = format!("{bucket_dir}/{file_name}");
+ let file_path = location.path.clone();
self.created_paths.push(file_path.clone());
if self.index_options.is_some() {
self.created_paths.push(format!(
@@ -230,6 +241,7 @@ impl DataFileWriter {
self.current_writer = Some(with_write_resources(writer,
self.resources.as_ref()));
self.current_index = index;
self.current_file_name = Some(file_name);
+ self.current_file_path = Some(location);
self.current_row_count = 0;
Ok(())
}
@@ -260,7 +272,8 @@ impl DataFileWriter {
let writer = self.current_writer.take()?;
let index = self.current_index.take();
let file_io = self.file_io.clone();
- let bucket_dir = self.bucket_dir();
+ let location = self.current_file_path.take().unwrap();
+ let bucket_dir = location.parent().to_string();
let threshold = self
.index_options
.as_ref()
@@ -285,6 +298,7 @@ impl DataFileWriter {
write_cols,
write_result.value_stats,
);
+ meta.external_path = location.external_path;
if let Some(index) = index {
let bytes = index.serialize()?;
if bytes.len() as u64 > threshold.unwrap() as u64 {
@@ -317,20 +331,32 @@ impl DataFileWriter {
pub(super) async fn prepare_all<K>(
writers: impl IntoIterator<Item = (K, Self)>,
) -> Result<Vec<(K, Vec<DataFileMeta>)>> {
- let results =
- futures::future::join_all(writers.into_iter().map(|(key, mut
writer)| async {
- let result = writer.prepare_commit().await;
- (key, writer, result)
- }))
- .await;
- if results.iter().any(|(_, _, result)| result.is_err()) {
+ let mut writers: Vec<_> = writers.into_iter().collect();
+ let files = Self::prepare_group(writers.iter_mut().map(|(_, writer)|
writer)).await?;
+ Ok(writers
+ .into_iter()
+ .zip(files)
+ .map(|((key, _), files)| (key, files))
+ .collect())
+ }
+
+ /// Close all physical columns as one operation. Successful closes transfer
+ /// ownership here until every other column succeeds, including Blob files.
+ pub(super) async fn prepare_group<'a>(
+ writers: impl IntoIterator<Item = &'a mut Self>,
+ ) -> Result<Vec<Vec<DataFileMeta>>> {
+ let results =
futures::future::join_all(writers.into_iter().map(|writer| async {
+ let result = writer.prepare_commit().await;
+ (writer, result)
+ }))
+ .await;
+ if results.iter().any(|(_, result)| result.is_err()) {
let mut first_error = None;
- for (_, mut writer, result) in results {
+ for (writer, result) in results {
match result {
Ok(files) => {
- // Successful prepare transferred these paths to us.
for file in files {
- for path in
file.collect_files(&writer.bucket_dir()) {
+ for path in
file.collect_files(writer.bucket_dir()) {
let _ =
writer.file_io.delete_file(&path).await;
}
}
@@ -343,10 +369,17 @@ impl DataFileWriter {
}
return Err(first_error.unwrap());
}
- results
- .into_iter()
- .map(|(key, _, files)| files.map(|files| (key, files)))
- .collect()
+ results.into_iter().map(|(_, files)| files).collect()
+ }
+
+ #[cfg(test)]
+ pub(super) fn inject_close_failure(&mut self) {
+ self.in_flight_closes.spawn(async {
+ Err(crate::Error::DataInvalid {
+ message: "injected close failure".into(),
+ source: None,
+ })
+ });
}
async fn finish(&mut self) -> Result<Vec<DataFileMeta>> {
@@ -370,6 +403,7 @@ impl DataFileWriter {
}
self.current_index = None;
self.current_file_name = None;
+ self.current_file_path = None;
while self.in_flight_closes.join_next().await.is_some() {}
for path in self.created_paths.drain(..) {
let _ = self.file_io.delete_file(&path).await;
@@ -377,12 +411,8 @@ impl DataFileWriter {
self.written_files.clear();
}
- fn bucket_dir(&self) -> String {
- let data_root = crate::spec::data_file_path(
- &self.table_location,
- CoreOptions::new(&self.format_options).data_file_path_directory(),
- );
- bucket_path_under(&data_root, &self.partition_path, self.bucket)
+ fn bucket_dir(&self) -> &str {
+ self.paths.bucket_path()
}
#[allow(clippy::too_many_arguments)]
@@ -484,7 +514,8 @@ mod tests {
None,
None,
None,
- );
+ )
+ .unwrap();
let schema = Arc::new(Schema::new(vec![Field::new(
"id",
ArrowDataType::Int32,
@@ -524,6 +555,7 @@ mod tests {
None,
None,
)
+ .unwrap()
.with_target_file_row_num(2);
let schema = Arc::new(Schema::new(vec![Field::new(
"id",
@@ -547,55 +579,71 @@ mod tests {
#[tokio::test]
async fn grouped_prepare_failure_removes_successful_and_failed_outputs() {
- let file_io = FileIOBuilder::new("memory").build().unwrap();
- let mut writers = Vec::new();
- for first_row_id in [0, 1] {
- let mut writer = DataFileWriter::new(
- file_io.clone(),
- "memory:///prepare-failure".into(),
- String::new(),
- 0,
- 0,
- i64::MAX,
- "none".into(),
- 0,
- i64::MAX,
- "parquet".into(),
- vec![DataField::new(
+ for external in [false, true] {
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let mut writers = Vec::new();
+ for first_row_id in [0, 1] {
+ let mut writer = DataFileWriter::new(
+ file_io.clone(),
+ "memory:///prepare-failure".into(),
+ String::new(),
0,
- "id".into(),
- DataType::Int(IntType::new()),
- )],
- HashMap::new(),
- Some(first_row_id),
- None,
- None,
- );
- let batch = RecordBatch::try_from_iter([(
- "id",
- Arc::new(Int32Array::from(vec![1])) as arrow_array::ArrayRef,
- )])
- .unwrap();
- writer.write(&batch).await.unwrap();
- if first_row_id == 1 {
- // Simulate a background file close failing after another file
- // in the operation has already finished successfully.
- writer.in_flight_closes.spawn(async {
- Err(crate::Error::DataInvalid {
- message: "injected close failure".into(),
- source: None,
- })
- });
+ 0,
+ i64::MAX,
+ "none".into(),
+ 0,
+ i64::MAX,
+ "parquet".into(),
+ vec![DataField::new(
+ 0,
+ "id".into(),
+ DataType::Int(IntType::new()),
+ )],
+ if external {
+ HashMap::from([
+ (
+ "data-file.external-paths".into(),
+ "memory:/external-a,memory:/external-b".into(),
+ ),
+ (
+ "data-file.external-paths.strategy".into(),
+ "entropy-inject".into(),
+ ),
+ ])
+ } else {
+ HashMap::new()
+ },
+ Some(first_row_id),
+ None,
+ None,
+ )
+ .unwrap();
+ let batch = RecordBatch::try_from_iter([(
+ "id",
+ Arc::new(Int32Array::from(vec![1])) as
arrow_array::ArrayRef,
+ )])
+ .unwrap();
+ writer.write(&batch).await.unwrap();
+ if first_row_id == 1 {
+ // Simulate a background file close failing after another
file
+ // in the operation has already finished successfully.
+ writer.in_flight_closes.spawn(async {
+ Err(crate::Error::DataInvalid {
+ message: "injected close failure".into(),
+ source: None,
+ })
+ });
+ }
+ writers.push((first_row_id, writer));
}
- writers.push((first_row_id, writer));
+ let error =
DataFileWriter::prepare_all(writers).await.unwrap_err();
+ assert!(error.to_string().contains("injected close failure"));
+ assert!(file_io
+ .list_status_recursive("memory:/")
+ .await
+ .unwrap()
+ .iter()
+ .all(|entry| !entry.path.ends_with(".parquet")));
}
- let error = DataFileWriter::prepare_all(writers).await.unwrap_err();
- assert!(error.to_string().contains("injected close failure"));
- assert!(file_io
- .list_status_recursive("memory:///prepare-failure")
- .await
- .unwrap()
- .iter()
- .all(|entry| !entry.path.ends_with(".parquet")));
}
}
diff --git a/crates/paimon/src/table/dedicated_format_file_writer.rs
b/crates/paimon/src/table/dedicated_format_file_writer.rs
index fb833dfc..7fcc2d2b 100644
--- a/crates/paimon/src/table/dedicated_format_file_writer.rs
+++ b/crates/paimon/src/table/dedicated_format_file_writer.rs
@@ -82,7 +82,13 @@ impl AppendDedicatedFormatFileWriter {
format_options: &HashMap<String, String>,
blob_inline_fields: &HashSet<String>,
blob_view_fields: &HashSet<String>,
- ) -> Self {
+ ) -> Result<Self> {
+ let paths =
Arc::new(super::data_file_path_factory::DataFilePathFactory::new(
+ &table_location,
+ &partition_path,
+ bucket,
+ format_options,
+ )?);
let mut normal_column_indices = Vec::new();
let mut normal_arrow_fields = Vec::new();
let mut normal_table_fields = Vec::new();
@@ -121,7 +127,8 @@ impl AppendDedicatedFormatFileWriter {
Some(0),
None,
Some(vec![field.name().to_string()]),
- )
+ )?
+ .with_path_factory(paths.clone())
.with_target_file_row_num(target_file_row_num),
field_name: field.name().to_string(),
column_index: idx,
@@ -160,7 +167,8 @@ impl AppendDedicatedFormatFileWriter {
Some(0),
None,
Some(vector_field_names.clone()),
- )
+ )?
+ .with_path_factory(paths.clone())
.with_target_file_row_num(target_file_row_num),
field_names: vector_field_names,
column_indices: vector_column_indices,
@@ -187,10 +195,11 @@ impl AppendDedicatedFormatFileWriter {
Some(0),
None,
Some(normal_field_names),
- )
+ )?
+ .with_path_factory(paths.clone())
.with_target_file_row_num(target_file_row_num);
- Self {
+ Ok(Self {
normal_writer,
blob_writers,
vector_writer,
@@ -202,7 +211,7 @@ impl AppendDedicatedFormatFileWriter {
.filter(|(_, field)| blob_view_fields.contains(field.name()))
.map(|(idx, field)| (idx, field.name().to_string()))
.collect(),
- }
+ })
}
pub(crate) fn with_resources(mut self, resources: Option<ResourceContext>)
-> Self {
@@ -331,18 +340,102 @@ impl AppendDedicatedFormatFileWriter {
}
pub(crate) async fn prepare_commit(&mut self) -> Result<Vec<DataFileMeta>>
{
- let mut results = self.normal_writer.prepare_commit().await?;
+ let writers = std::iter::once(&mut self.normal_writer)
+ .chain(
+ self.blob_writers
+ .iter_mut()
+ .map(|writer| &mut writer.writer),
+ )
+ .chain(
+ self.vector_writer
+ .iter_mut()
+ .map(|writer| &mut writer.writer),
+ );
+ Ok(DataFileWriter::prepare_group(writers)
+ .await?
+ .into_iter()
+ .flatten()
+ .collect())
+ }
+}
- for blob_writer in &mut self.blob_writers {
- let blob_metas = blob_writer.writer.prepare_commit().await?;
- results.extend(blob_metas);
- }
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::io::FileIOBuilder;
+ use crate::spec::{BlobType, IntType};
+ use arrow_array::{ArrayRef, Int32Array, LargeBinaryArray};
- if let Some(vector_writer) = &mut self.vector_writer {
- let vector_metas = vector_writer.writer.prepare_commit().await?;
- results.extend(vector_metas);
+ #[tokio::test]
+ async fn failed_blob_close_cleans_every_physical_column() {
+ for external in [false, true] {
+ let io = FileIOBuilder::new("memory").build().unwrap();
+ let batch = RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![1])) as ArrayRef),
+ (
+ "a",
+
Arc::new(LargeBinaryArray::from(vec![Some(b"a".as_slice())])) as ArrayRef,
+ ),
+ (
+ "b",
+
Arc::new(LargeBinaryArray::from(vec![Some(b"b".as_slice())])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let fields = vec![
+ DataField::new(0, "id".into(), DataType::Int(IntType::new())),
+ DataField::new(1, "a".into(), DataType::Blob(BlobType::new())),
+ DataField::new(2, "b".into(), DataType::Blob(BlobType::new())),
+ ];
+ let options = if external {
+ HashMap::from([
+ (
+ "data-file.external-paths".into(),
+ "memory:/first,memory:/second".into(),
+ ),
+ (
+ "data-file.external-paths.strategy".into(),
+ "entropy-inject".into(),
+ ),
+ ])
+ } else {
+ HashMap::new()
+ };
+ let mut writer = AppendDedicatedFormatFileWriter::new(
+ io.clone(),
+ "memory:/table".into(),
+ "".into(),
+ 0,
+ 0,
+ i64::MAX,
+ i64::MAX,
+ i64::MAX,
+ "none".into(),
+ 0,
+ i64::MAX,
+ "parquet".into(),
+ i64::MAX,
+ None,
+ batch.schema().as_ref(),
+ &fields,
+ &options,
+ &HashSet::new(),
+ &HashSet::new(),
+ )
+ .unwrap();
+ writer.write(&batch).await.unwrap();
+ writer.blob_writers[1].writer.inject_close_failure();
+ assert!(writer
+ .prepare_commit()
+ .await
+ .unwrap_err()
+ .to_string()
+ .contains("injected close failure"));
+ assert!(io
+ .list_status_recursive("memory:/")
+ .await
+ .unwrap()
+ .is_empty());
}
-
- Ok(results)
}
}
diff --git a/crates/paimon/src/table/external_path.rs
b/crates/paimon/src/table/external_path.rs
index 9d7b730f..28e44f09 100644
--- a/crates/paimon/src/table/external_path.rs
+++ b/crates/paimon/src/table/external_path.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-//! External index placement follows Java's FileStorePathFactory. Existing
+//! External file placement follows Java's FileStorePathFactory. Existing
//! files always resolve using their recorded path, never current write
options.
use crate::Result;
@@ -46,7 +46,7 @@ pub(super) fn new_index_external_path(
/// Mirrors Java's per-bucket ExternalPathProvider. The random starting point
/// avoids concentrating single-file buckets on the first configured root.
-struct ExternalPathProvider {
+pub(super) struct ExternalPathProvider {
paths: Vec<String>,
bucket: String,
position: usize,
@@ -55,7 +55,28 @@ struct ExternalPathProvider {
}
impl ExternalPathProvider {
- fn new(options: &HashMap<String, String>, bucket: &str) ->
Result<Option<Self>> {
+ pub(super) fn new(options: &HashMap<String, String>, bucket: &str) ->
Result<Option<Self>> {
+ // Java parses configured weights even when a strategy will not use
them.
+ let weights = options
+ .get("data-file.external-paths.weights")
+ .map(|weights| {
+ weights
+ .trim_end_matches(',')
+ .split(',')
+ .map(|weight| {
+ weight
+ .trim()
+ .parse::<i32>()
+ .ok()
+ .filter(|weight| *weight > 0)
+ .map(|weight| weight as u64)
+ .ok_or_else(|| {
+ invalid("External path weights must be
positive integers")
+ })
+ })
+ .collect::<Result<Vec<_>>>()
+ })
+ .transpose()?;
let strategy = options
.get("data-file.external-paths.strategy")
.map(|value| value.to_ascii_lowercase())
@@ -66,17 +87,17 @@ impl ExternalPathProvider {
else {
return Ok(None);
};
- if strategy == "none" {
- return Ok(None);
- }
if !matches!(
strategy.as_str(),
- "round-robin" | "specific-fs" | "weight-robin" | "entropy-inject"
+ "none" | "round-robin" | "specific-fs" | "weight-robin" |
"entropy-inject"
) {
return Err(invalid(format!(
"Unsupported external path strategy: {strategy}"
)));
}
+ if strategy == "none" {
+ return Ok(None);
+ }
let specific_fs = if strategy == "specific-fs" {
Some(
options
@@ -100,22 +121,11 @@ impl ExternalPathProvider {
}
let mut cumulative_weights = Vec::new();
if strategy == "weight-robin" && roots.len() > 1 {
- if let Some(weights) = options
- .get("data-file.external-paths.weights")
- .filter(|weights| !weights.trim().is_empty())
- {
+ if let Some(weights) = weights {
let mut total = 0_u64;
- for weight in weights.trim_end_matches(',').split(',') {
- let weight = weight
- .trim()
- .parse::<i32>()
- .ok()
- .filter(|weight| *weight > 0)
- .ok_or_else(|| {
- invalid("External path weights must be positive
integers")
- })?;
+ for weight in weights {
total = total
- .checked_add(weight as u64)
+ .checked_add(weight)
.ok_or_else(|| invalid("External path weight
overflow"))?;
cumulative_weights.push(total);
}
@@ -141,7 +151,7 @@ impl ExternalPathProvider {
}))
}
- fn next_path(&mut self, file_name: &str) -> String {
+ pub(super) fn next_path(&mut self, file_name: &str) -> String {
let index = if let Some(total) = self.cumulative_weights.last() {
let value = rand::thread_rng().gen_range(0..*total);
self.cumulative_weights
@@ -277,6 +287,34 @@ mod tests {
}
}
+ #[test]
+ fn disabled_strategy_does_not_resolve_unused_external_roots() {
+ let options = HashMap::from([(
+ "data-file.external-paths".into(),
+ "/unused/local/path".into(),
+ )]);
+ assert!(ExternalPathProvider::new(&options, "bucket-0")
+ .unwrap()
+ .is_none());
+ }
+
+ #[test]
+ fn invalid_weights_are_rejected_even_without_weighted_distribution() {
+ for strategy in ["none", "round-robin", "weight-robin"] {
+ for roots in ["file:/one", "file:/one,file:/two"] {
+ for weights in ["", "0", "-1", "x", "2147483648", "1,,2"] {
+ let mut options = options(strategy);
+ options.insert("data-file.external-paths".into(),
roots.into());
+ options.insert("data-file.external-paths.weights".into(),
weights.into());
+ assert!(
+ ExternalPathProvider::new(&options,
"bucket-0").is_err(),
+ "{strategy}/{roots}/{weights}"
+ );
+ }
+ }
+ }
+ }
+
#[test]
fn comma_lists_follow_java_trailing_empty_semantics() {
let mut options = options("weight-robin");
diff --git a/crates/paimon/src/table/kv_file_writer.rs
b/crates/paimon/src/table/kv_file_writer.rs
index ebef74ab..4b47263e 100644
--- a/crates/paimon/src/table/kv_file_writer.rs
+++ b/crates/paimon/src/table/kv_file_writer.rs
@@ -34,9 +34,9 @@ use crate::io::FileIO;
use crate::resource::{MemoryReservation, ResourceContext};
use crate::spec::stats::{compute_column_stats, BinaryTableStats};
use crate::spec::{
- bucket_path_under, data_file_to_file_index_file_name,
extract_datum_from_arrow,
- AggregationConfig, BigIntType, BinaryRowBuilder, CoreOptions, DataField,
DataFileMeta,
- DataType, MergeEngine, PartialUpdateConfig, RowKind, TinyIntType,
SEQUENCE_NUMBER_FIELD_ID,
+ data_file_to_file_index_file_name, extract_datum_from_arrow,
AggregationConfig, BigIntType,
+ BinaryRowBuilder, CoreOptions, DataField, DataFileMeta, DataType,
MergeEngine,
+ PartialUpdateConfig, RowKind, TinyIntType, SEQUENCE_NUMBER_FIELD_ID,
SEQUENCE_NUMBER_FIELD_NAME, VALUE_KIND_FIELD_ID, VALUE_KIND_FIELD_NAME,
};
use crate::table::data_file_index_writer::FileIndexOptions;
@@ -59,6 +59,7 @@ use std::sync::Arc;
/// Internal writer for primary-key tables that buffers data in memory,
/// sorts by primary key on flush, and prepends `_SEQUENCE_NUMBER` and
`_VALUE_KIND` columns.
pub(crate) struct KeyValueFileWriter {
+ paths: super::data_file_path_factory::DataFilePathFactory,
file_io: FileIO,
config: KeyValueWriteConfig,
target_file_row_num: usize,
@@ -169,8 +170,15 @@ impl KeyValueFileWriter {
}
let managed_blob_writer = ManagedBlobWriteState::new(&file_io,
&config)?;
+ let paths = super::data_file_path_factory::DataFilePathFactory::new(
+ &config.table_location,
+ &config.partition_path,
+ config.bucket,
+ &config.table_options,
+ )?;
Ok(Self {
+ paths,
file_io,
config,
target_file_row_num,
@@ -490,16 +498,10 @@ impl KeyValueFileWriter {
write.file_ordinal,
write.file_format,
);
- let bucket_dir = bucket_path_under(
- &crate::spec::data_file_path(
- &self.config.table_location,
-
CoreOptions::new(&self.config.table_options).data_file_path_directory(),
- ),
- &self.config.partition_path,
- self.config.bucket,
- );
+ let location = self.paths.new_path(&file_name)?;
+ let bucket_dir = location.parent();
self.file_io.mkdirs(&format!("{bucket_dir}/")).await?;
- let file_path = format!("{bucket_dir}/{file_name}");
+ let file_path = location.path.clone();
let output = self.file_io.new_output(&file_path)?;
// The physical KV file also contains sequence and row-kind columns.
Give
// Parquet only the logical value fields so metadata stats and their
dense
@@ -685,7 +687,7 @@ impl KeyValueFileWriter {
embedded_index: None,
file_source: Some(0), // FileSource.APPEND
value_stats_cols,
- external_path: None,
+ external_path: location.external_path.clone(),
first_row_id: None,
write_cols: None,
column_max_sequence_numbers: None,
@@ -730,7 +732,7 @@ impl KeyValueFileWriter {
blob_references,
&self.file_io,
&file_path,
- &bucket_dir,
+ bucket_dir,
&mut meta,
)
.await?;
@@ -1175,20 +1177,13 @@ impl KeyValueFileWriter {
if let Some(reservation) = &mut self.buffer_reservation {
let _ = reservation.try_resize(0);
}
- let bucket_path = bucket_path_under(
- &crate::spec::data_file_path(
- &self.config.table_location,
-
CoreOptions::new(&self.config.table_options).data_file_path_directory(),
- ),
- &self.config.partition_path,
- self.config.bucket,
- );
+ let bucket_path = self.paths.bucket_path();
for file in self
.written_files
.drain(..)
.chain(self.written_changelog_files.drain(..))
{
- for path in file.collect_files(&bucket_path) {
+ for path in file.collect_files(bucket_path) {
let _ = self.file_io.delete_file(&path).await;
}
}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 5cdf174a..e779858e 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -42,6 +42,7 @@ mod data_evolution_nested;
mod data_evolution_reader;
pub mod data_evolution_writer;
mod data_file_index_writer;
+mod data_file_path_factory;
mod data_file_reader;
mod data_file_writer;
mod de_vector_read;
diff --git a/crates/paimon/src/table/postpone_file_writer.rs
b/crates/paimon/src/table/postpone_file_writer.rs
index b932de33..ea117501 100644
--- a/crates/paimon/src/table/postpone_file_writer.rs
+++ b/crates/paimon/src/table/postpone_file_writer.rs
@@ -24,13 +24,13 @@
//!
//! Reference:
[PostponeBucketWriter](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/sink/PostponeBucketWriter.java)
+use super::data_file_path_factory::{DataFilePath, DataFilePathFactory};
use crate::arrow::format::{create_format_writer, with_write_resources,
FormatFileWriter};
use crate::io::FileIO;
use crate::resource::ResourceContext;
use crate::spec::stats::BinaryTableStats;
use crate::spec::{
- bucket_path_under, data_file_to_file_index_file_name, DataFileMeta,
EMPTY_SERIALIZED_ROW,
- VALUE_KIND_FIELD_NAME,
+ data_file_to_file_index_file_name, DataFileMeta, EMPTY_SERIALIZED_ROW,
VALUE_KIND_FIELD_NAME,
};
use crate::table::data_file_index_writer::{DataFileIndexWriter,
FileIndexOptions};
use crate::table::kv_file_writer::build_physical_schema;
@@ -42,6 +42,7 @@ use tokio::task::JoinSet;
/// Configuration for [`PostponeFileWriter`].
pub(crate) struct PostponeWriteConfig {
+ pub table_options: std::collections::HashMap<String, String>,
pub table_location: String,
pub partition_path: String,
pub bucket: i32,
@@ -62,12 +63,14 @@ pub(crate) struct PostponeWriteConfig {
/// prepending `_SEQUENCE_NUMBER` and `_VALUE_KIND` columns to each batch.
/// Rolls to a new file when `target_file_size` is reached.
pub(crate) struct PostponeFileWriter {
+ paths: DataFilePathFactory,
file_io: FileIO,
config: PostponeWriteConfig,
next_sequence_number: i64,
current_writer: Option<Box<dyn FormatFileWriter>>,
current_index: Option<DataFileIndexWriter>,
current_file_name: Option<String>,
+ current_file_path: Option<DataFilePath>,
current_row_count: i64,
/// Sequence number at which the current file started.
current_file_start_seq: i64,
@@ -81,14 +84,22 @@ pub(crate) struct PostponeFileWriter {
}
impl PostponeFileWriter {
- pub(crate) fn new(file_io: FileIO, config: PostponeWriteConfig) -> Self {
- Self {
+ pub(crate) fn new(file_io: FileIO, config: PostponeWriteConfig) ->
Result<Self> {
+ let paths = DataFilePathFactory::new(
+ &config.table_location,
+ &config.partition_path,
+ config.bucket,
+ &config.table_options,
+ )?;
+ Ok(Self {
+ paths,
file_io,
config,
next_sequence_number: 0,
current_writer: None,
current_index: None,
current_file_name: None,
+ current_file_path: None,
current_row_count: 0,
current_file_start_seq: 0,
current_file_creation_time: Utc::now(),
@@ -96,7 +107,7 @@ impl PostponeFileWriter {
created_paths: Vec::new(),
in_flight_closes: JoinSet::new(),
resources: None,
- }
+ })
}
pub(crate) fn with_resources(mut self, resources: Option<ResourceContext>)
-> Self {
@@ -199,6 +210,7 @@ impl PostponeFileWriter {
}
self.current_index = None;
self.current_file_name = None;
+ self.current_file_path = None;
while self.in_flight_closes.join_next().await.is_some() {}
for path in self.created_paths.drain(..) {
let _ = self.file_io.delete_file(&path).await;
@@ -236,11 +248,8 @@ impl PostponeFileWriter {
let file_name = self.current_file_name.take().unwrap();
let index = self.current_index.take();
let file_io = self.file_io.clone();
- let bucket_dir = bucket_path_under(
- &self.config.table_location,
- &self.config.partition_path,
- self.config.bucket,
- );
+ let location = self.current_file_path.take().unwrap();
+ let bucket_dir = location.parent().to_string();
let threshold = self
.config
.file_index_options
@@ -266,6 +275,7 @@ impl PostponeFileWriter {
schema_id,
creation_time,
);
+ meta.external_path = location.external_path;
write_index(index, threshold, &file_io, &bucket_dir, &mut
meta).await?;
Ok(meta)
});
@@ -285,14 +295,11 @@ impl PostponeFileWriter {
self.written_files.len(),
self.config.file_format,
);
- let bucket_dir = bucket_path_under(
- &self.config.table_location,
- &self.config.partition_path,
- self.config.bucket,
- );
+ let location = self.paths.new_path(&file_name)?;
+ let bucket_dir = location.parent();
self.file_io.mkdirs(&format!("{bucket_dir}/")).await?;
let physical_schema = build_physical_schema(&user_schema);
- let file_path = format!("{bucket_dir}/{file_name}");
+ let file_path = location.path.clone();
self.created_paths.push(file_path.clone());
if index.is_some() {
self.created_paths.push(format!(
@@ -314,6 +321,7 @@ impl PostponeFileWriter {
self.current_writer = Some(with_write_resources(writer,
self.resources.as_ref()));
self.current_index = index;
self.current_file_name = Some(file_name);
+ self.current_file_path = Some(location);
self.current_row_count = 0;
self.current_file_start_seq = self.next_sequence_number;
self.current_file_creation_time = Utc::now();
@@ -343,16 +351,14 @@ impl PostponeFileWriter {
self.config.schema_id,
self.current_file_creation_time,
);
- let bucket_dir = bucket_path_under(
- &self.config.table_location,
- &self.config.partition_path,
- self.config.bucket,
- );
+ let location = self.current_file_path.take().unwrap();
+ let bucket_dir = location.parent().to_string();
let threshold = self
.config
.file_index_options
.as_ref()
.map(|options| options.in_manifest_threshold);
+ meta.external_path = location.external_path;
write_index(index, threshold, &self.file_io, &bucket_dir, &mut
meta).await?;
self.written_files.push(meta);
Ok(())
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 4e78c42e..80f84d4a 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -248,6 +248,11 @@ impl TableWrite {
CoreOptions::new(table.schema().options()).validate_data_file_path_directory()?;
let is_overwrite = false;
let schema = table.schema();
+ crate::spec::Schema::validate_primary_key_blob_configuration(
+ schema.fields(),
+ schema.primary_keys(),
+ schema.options(),
+ )?;
let write_schema = build_target_arrow_schema(schema.fields())?;
let core_options = CoreOptions::new(schema.options());
let blob_descriptor_fields = core_options.blob_descriptor_fields();
@@ -1119,7 +1124,7 @@ impl TableWrite {
let writer = if self.primary_key_indices.is_empty() {
self.create_append_writer(partition_path, bucket)?
} else if bucket == POSTPONE_BUCKET {
- self.create_postpone_writer(partition_path, bucket)
+ self.create_postpone_writer(partition_path, bucket)?
} else {
self.create_kv_writer(partition_path, bucket, &partition_bytes)
.await?
@@ -1167,7 +1172,7 @@ impl TableWrite {
self.table.schema().options(),
&self.blob_inline_fields,
&self.blob_view_fields,
- )
+ )?
.with_resources(self.resources.clone()),
)))
} else {
@@ -1188,7 +1193,7 @@ impl TableWrite {
Some(0),
None,
None,
- )
+ )?
.with_file_index(self.file_index_options.clone())
.with_target_file_row_num(self.target_file_row_num)
.with_resources(self.resources.clone()),
@@ -1197,16 +1202,17 @@ impl TableWrite {
}
/// Create a postpone writer (KV format, no sorting/dedup, special file
naming).
- fn create_postpone_writer(&self, partition_path: String, bucket: i32) ->
FileWriter {
+ fn create_postpone_writer(&self, partition_path: String, bucket: i32) ->
Result<FileWriter> {
let data_file_prefix = format!(
"{}-u-{}-s-{}-w-",
self.data_file_prefix, self.commit_user, self.postpone_write_id
);
- FileWriter::Postpone(
+ Ok(FileWriter::Postpone(
PostponeFileWriter::new(
self.table.file_io().clone(),
PostponeWriteConfig {
- table_location: self.table.data_file_location(),
+ table_location: self.table.location().to_string(),
+ table_options: self.table.schema().options().clone(),
partition_path,
bucket,
schema_id: self.schema_id,
@@ -1218,9 +1224,9 @@ impl TableWrite {
data_file_prefix,
file_index_options: self.file_index_options.clone(),
},
- )
+ )?
.with_resources(self.resources.clone()),
- )
+ ))
}
/// Create a key-value writer for PK tables with normal buckets.
diff --git a/crates/paimon/tests/external_data_write_test.rs
b/crates/paimon/tests/external_data_write_test.rs
new file mode 100644
index 00000000..4042ab08
--- /dev/null
+++ b/crates/paimon/tests/external_data_write_test.rs
@@ -0,0 +1,383 @@
+// 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 arrow_array::{ArrayRef, Int32Array, RecordBatch, StringArray};
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs};
+use futures::TryStreamExt;
+use paimon::spec::{DataType, IntType, Schema, TableSchema, VarCharType};
+use paimon::table::Table;
+use std::sync::Arc;
+
+async fn table(mode: &str, strategy: &str, indexes: bool) -> Table {
+ let path = "memory:/external_write";
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("pt", DataType::VarChar(VarCharType::string_type()))
+ .column("value", DataType::Int(IntType::new()))
+ .partition_keys(["pt"])
+ .option("data-file.path-directory", "data/nested")
+ .option(
+ "data-file.external-paths",
+ "memory:/external-a,memory:/external-b",
+ )
+ .option("data-file.external-paths.strategy", strategy)
+ .option("data-file.external-paths.weights", "1,2")
+ .option("data-file.external-paths.specific-fs", "memory")
+ .option("target-file-row-num", "1");
+ if mode == "pk" || mode == "postpone" {
+ schema = schema
+ .primary_key(["id", "pt"])
+ .option("bucket", if mode == "postpone" { "-2" } else { "1" })
+ .option(
+ "changelog-producer",
+ if mode == "postpone" { "none" } else { "input" },
+ );
+ }
+ if mode == "evolution" {
+ schema = schema
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .option("deletion-vectors.enabled", "true");
+ }
+ if indexes {
+ schema = schema
+ .option("file-index.bloom-filter.columns", "value")
+ .option("file-index.bloom-filter.value.items", "10")
+ .option("file-index.in-manifest-threshold", "0 B");
+ }
+ let (io, table) = memory_table(path, TableSchema::new(0,
&schema.build().unwrap()));
+ setup_dirs(&io, path).await;
+ persist_table_schema(&io, path, table.schema()).await;
+ table
+}
+
+fn batch(id: i32, value: i32) -> RecordBatch {
+ RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![id])) as ArrayRef),
+ ("pt", Arc::new(StringArray::from(vec!["a/b"])) as ArrayRef),
+ ("value", Arc::new(Int32Array::from(vec![value])) as ArrayRef),
+ ])
+ .unwrap()
+}
+
+async fn values(table: &Table) -> Vec<i32> {
+ let builder = table.new_read_builder();
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut values: Vec<_> = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column_by_name("value")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect();
+ values.sort();
+ values
+}
+
+#[tokio::test]
+async fn external_data_and_sidecars_follow_the_recorded_path() {
+ for mode in ["append", "pk", "evolution", "postpone"] {
+ for strategy in [
+ "round-robin",
+ "weight-robin",
+ "entropy-inject",
+ "specific-fs",
+ "none",
+ ] {
+ let table = table(mode, strategy, true).await;
+ let builder = table.new_write_builder();
+ let mut write = builder.new_write().unwrap();
+ for id in 1..=4 {
+ write.write_arrow_batch(&batch(id, id * 10)).await.unwrap();
+ }
+ let messages = write.prepare_commit().await.unwrap();
+ let mut external_roots = std::collections::HashSet::new();
+ for message in &messages {
+ for file in
message.new_files.iter().chain(&message.new_changelog_files) {
+ assert_eq!(
+ file.external_path.is_some(),
+ strategy != "none",
+ "{mode}/{strategy}"
+ );
+ let bucket = format!(
+ "{}/data/nested/pt=a%2Fb/bucket-{}",
+ table.location(),
+ if mode == "postpone" { "postpone" } else { "0" }
+ );
+ for path in file.collect_files(&bucket) {
+ assert!(table.file_io().exists(&path).await.unwrap(),
"{path}");
+ if strategy != "none" {
+ assert!(path.starts_with("memory:/external-"),
"{path}");
+
assert!(path.contains("/data/nested/pt=a%2Fb/bucket-"));
+
external_roots.insert(path.split('/').nth(1).unwrap().to_string());
+ assert!(!table
+ .file_io()
+ .exists(&format!("{bucket}/{}",
file.file_name))
+ .await
+ .unwrap());
+ }
+ }
+ if !file.file_name.starts_with("changelog-") {
+ assert!(!file.extra_files.is_empty());
+ }
+ }
+ }
+ if mode != "postpone"
+ && ["round-robin", "entropy-inject",
"specific-fs"].contains(&strategy)
+ {
+ assert_eq!(external_roots.len(), 2);
+ }
+ builder.new_commit().commit(messages).await.unwrap();
+ // Normal scans intentionally skip postpone files until compaction.
+ let expected = if mode == "postpone" {
+ vec![]
+ } else {
+ vec![10, 20, 30, 40]
+ };
+ assert_eq!(values(&table).await, expected, "{mode}/{strategy}");
+ }
+ }
+}
+
+#[tokio::test]
+async fn abort_removes_external_data_changelog_and_sidecars() {
+ for mode in ["append", "pk", "evolution", "postpone"] {
+ let table = table(mode, "entropy-inject", true).await;
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&batch(1, 10)).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ let mut paths = Vec::new();
+ for message in &messages {
+ for file in
message.new_files.iter().chain(&message.new_changelog_files) {
+ paths.extend(file.collect_files("unused-table-bucket"));
+ }
+ }
+ assert!(!paths.is_empty());
+ for path in &paths {
+ assert!(path.starts_with("memory:/external-"));
+ assert!(table.file_io().exists(path).await.unwrap());
+ }
+ builder.new_commit().abort(&messages).await.unwrap();
+ for path in &paths {
+ assert!(!table.file_io().exists(path).await.unwrap(), "{path}");
+ }
+ assert!(table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .is_none());
+ }
+}
+
+#[tokio::test]
+async fn updates_read_old_external_files_and_publish_to_new_roots() {
+ use std::collections::HashMap;
+ let table = table("evolution", "entropy-inject", false).await;
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ for id in 1..=3 {
+ writer.write_arrow_batch(&batch(id, id * 10)).await.unwrap();
+ }
+ builder
+ .new_commit()
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ let moved = table.copy_with_options(HashMap::from([
+ ("data-file.external-paths".into(), "memory:/new-root".into()),
+ (
+ "data-file.external-paths.strategy".into(),
+ "round-robin".into(),
+ ),
+ ]));
+ let mut update = moved.new_write_builder().new_update().unwrap();
+ update.with_update_type(vec!["value".into()]).unwrap();
+ let input = RecordBatch::try_from_iter([
+ (
+ "_ROW_ID",
+ Arc::new(arrow_array::Int64Array::from(vec![0])) as ArrayRef,
+ ),
+ ("value", Arc::new(Int32Array::from(vec![11])) as ArrayRef),
+ ])
+ .unwrap();
+ let messages = update
+ .update_by_arrow_with_row_id(vec![input])
+ .await
+ .unwrap();
+ for file in messages.iter().flat_map(|m| &m.new_files) {
+ assert!(file
+ .external_path
+ .as_ref()
+ .unwrap()
+ .starts_with("memory:/new-root/"));
+ }
+ moved
+ .new_write_builder()
+ .new_commit()
+ .commit(messages)
+ .await
+ .unwrap();
+ assert_eq!(values(&moved).await, vec![11, 20, 30]);
+ let messages = update
+ .upsert_by_arrow_with_key(vec![batch(2, 22), batch(4, 40)],
vec!["id".into()])
+ .await
+ .unwrap();
+ moved
+ .new_write_builder()
+ .new_commit()
+ .commit(messages)
+ .await
+ .unwrap();
+ assert_eq!(values(&moved).await, vec![11, 22, 30, 40]);
+ for (row_id, expected) in [(0, vec![22, 30, 40]), (1, vec![30, 40])] {
+ let messages = update.delete_by_row_id(vec![row_id]).await.unwrap();
+ moved
+ .new_write_builder()
+ .new_commit()
+ .commit(messages)
+ .await
+ .unwrap();
+ assert_eq!(values(&moved).await, expected);
+ }
+ let historical =
+ moved.copy_with_options(HashMap::from([("scan.snapshot-id".into(),
"1".into())]));
+ assert_eq!(values(&historical).await, vec![10, 20, 30]);
+}
+
+#[tokio::test]
+async fn dedicated_blob_fields_share_external_rotation_with_parquet() {
+ use arrow_array::LargeBinaryArray;
+ use paimon::spec::BlobType;
+ let path = "memory:/external_blob";
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("first", DataType::Blob(BlobType::new()))
+ .column("second", DataType::Blob(BlobType::new()))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .option(
+ "data-file.external-paths",
+ "memory:/external-a,memory:/external-b",
+ )
+ .option("data-file.external-paths.strategy", "entropy-inject")
+ .build()
+ .unwrap();
+ let (io, table) = memory_table(path, TableSchema::new(0, &schema));
+ setup_dirs(&io, path).await;
+ persist_table_schema(&io, path, table.schema()).await;
+ let input = RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef),
+ (
+ "first",
+ Arc::new(LargeBinaryArray::from(vec![
+ Some(b"hello".as_slice()),
+ None,
+ Some(b"".as_slice()),
+ ])) as ArrayRef,
+ ),
+ (
+ "second",
+ Arc::new(LargeBinaryArray::from(vec![
+ None,
+ Some(b"world".as_slice()),
+ Some(b"!".as_slice()),
+ ])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&input).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ let files: Vec<_> = messages.iter().flat_map(|m| &m.new_files).collect();
+ assert_eq!(files.len(), 3);
+ for (column, root) in [
+ ("id", "external-b"),
+ ("first", "external-a"),
+ ("second", "external-b"),
+ ] {
+ let file = files
+ .iter()
+ .find(|file| {
+ file.write_cols
+ .as_ref()
+ .unwrap()
+ .contains(&column.to_string())
+ })
+ .unwrap();
+ let location = file.external_path.as_ref().unwrap();
+ assert!(
+ location.starts_with(&format!("memory:/{root}/bucket-0/")),
+ "{column}: {location}"
+ );
+ assert!(io.exists(location).await.unwrap());
+ }
+ builder.new_commit().commit(messages).await.unwrap();
+ let read = table.new_read_builder();
+ let plan = read.new_scan().plan().await.unwrap();
+ let batches: Vec<RecordBatch> = read
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+ assert_eq!(batches[0].columns(), input.columns());
+}
+
+#[tokio::test]
+async fn copied_pk_blob_options_cannot_enable_external_writes() {
+ use paimon::spec::BlobType;
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("payload", DataType::Blob(BlobType::new()))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .build()
+ .unwrap();
+ let (_, table) = memory_table("memory:/pk-blob", TableSchema::new(0,
&schema));
+ let table = table.copy_with_options(std::collections::HashMap::from([(
+ "data-file.external-paths".into(),
+ "memory:/outside".into(),
+ )]));
+ let error = table
+ .new_write_builder()
+ .new_write()
+ .err()
+ .expect("invalid copied schema must fail");
+ assert!(error.to_string().contains("external-paths"));
+}
