This is an automated email from the ASF dual-hosted git repository.
jerry-024 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 5a6a5ce7 fix(write): avoid double separators in bucket paths (#860)
5a6a5ce7 is described below
commit 5a6a5ce7bba2e53b6d8c147a1e06da5c988ae980
Author: shyjsarah <[email protected]>
AuthorDate: Thu Sep 17 17:17:34 2026 +0800
fix(write): avoid double separators in bucket paths (#860)
---
crates/paimon/src/table/data_file_writer.rs | 88 +++++++++++++++++++++----
crates/paimon/src/table/kv_file_writer.rs | 20 ++----
crates/paimon/src/table/postpone_file_writer.rs | 21 ++----
3 files changed, 90 insertions(+), 39 deletions(-)
diff --git a/crates/paimon/src/table/data_file_writer.rs
b/crates/paimon/src/table/data_file_writer.rs
index 76f49a0d..b283c5f7 100644
--- a/crates/paimon/src/table/data_file_writer.rs
+++ b/crates/paimon/src/table/data_file_writer.rs
@@ -25,7 +25,7 @@
use crate::arrow::format::{create_format_writer, FormatFileWriter,
FormatValueStats};
use crate::io::FileIO;
use crate::spec::stats::BinaryTableStats;
-use crate::spec::{bucket_dir_name, DataField, DataFileMeta,
EMPTY_SERIALIZED_ROW};
+use crate::spec::{bucket_path_under, DataField, DataFileMeta,
EMPTY_SERIALIZED_ROW};
use crate::Result;
use arrow_array::RecordBatch;
use chrono::Utc;
@@ -142,16 +142,7 @@ impl DataFileWriter {
self.written_files.len(),
self.file_format,
);
- let bucket_dir = if self.partition_path.is_empty() {
- format!("{}/{}", self.table_location, bucket_dir_name(self.bucket))
- } else {
- format!(
- "{}/{}/{}",
- self.table_location,
- self.partition_path,
- bucket_dir_name(self.bucket)
- )
- };
+ let bucket_dir = bucket_path_under(&self.table_location,
&self.partition_path, self.bucket);
self.file_io.mkdirs(&format!("{bucket_dir}/")).await?;
let file_path = format!("{bucket_dir}/{file_name}");
@@ -280,3 +271,78 @@ impl DataFileWriter {
}
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::io::{FileIOBuilder, FileIOProvider};
+ use crate::spec::{DataType, IntType};
+ use arrow_schema::{DataType as ArrowDataType, Field, Schema};
+ use opendal::Operator;
+ use std::sync::{Arc, Mutex};
+
+ #[derive(Debug)]
+ struct RecordingProvider {
+ operator: Operator,
+ paths: Mutex<Vec<String>>,
+ }
+
+ #[async_trait::async_trait]
+ impl FileIOProvider for RecordingProvider {
+ async fn create(&self, path: &str) -> Result<(Operator, String)> {
+ let relative_path = path
+ .strip_prefix("s3://bucket/")
+ .expect("writer path must use the configured bucket")
+ .to_string();
+ self.paths.lock().unwrap().push(relative_path.clone());
+ Ok((self.operator.clone(), relative_path))
+ }
+ }
+
+ #[tokio::test]
+ async fn partitioned_writer_does_not_create_double_slash_bucket_paths() {
+ let provider = Arc::new(RecordingProvider {
+ operator:
Operator::from_config(opendal::services::MemoryConfig::default()).unwrap(),
+ paths: Mutex::new(Vec::new()),
+ });
+ let file_io = FileIOBuilder::new("unused")
+ .with_provider(provider.clone())
+ .build()
+ .unwrap();
+ let mut writer = DataFileWriter::new(
+ file_io,
+ "s3://bucket/table".to_string(),
+ "dt=2026-09-17/".to_string(),
+ 0,
+ 0,
+ i64::MAX,
+ "none".to_string(),
+ 0,
+ 1024,
+ "parquet".to_string(),
+ vec![DataField::new(
+ 0,
+ "id".to_string(),
+ DataType::Int(IntType::new()),
+ )],
+ HashMap::new(),
+ None,
+ None,
+ None,
+ );
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "id",
+ ArrowDataType::Int32,
+ false,
+ )]));
+
+ writer.open_new_file(schema).await.unwrap();
+
+ assert!(provider
+ .paths
+ .lock()
+ .unwrap()
+ .iter()
+ .all(|path| !path.contains("//")));
+ }
+}
diff --git a/crates/paimon/src/table/kv_file_writer.rs
b/crates/paimon/src/table/kv_file_writer.rs
index 514d7524..a4ca6830 100644
--- a/crates/paimon/src/table/kv_file_writer.rs
+++ b/crates/paimon/src/table/kv_file_writer.rs
@@ -31,8 +31,8 @@ use crate::arrow::format::create_format_writer;
use crate::io::FileIO;
use crate::spec::stats::{compute_column_stats, BinaryTableStats};
use crate::spec::{
- extract_datum_from_arrow, AggregationConfig, BinaryRowBuilder,
CoreOptions, DataFileMeta,
- DataType, MergeEngine, PartialUpdateConfig, RowKind, EMPTY_SERIALIZED_ROW,
+ bucket_path_under, extract_datum_from_arrow, AggregationConfig,
BinaryRowBuilder, CoreOptions,
+ DataFileMeta, DataType, MergeEngine, PartialUpdateConfig, RowKind,
EMPTY_SERIALIZED_ROW,
SEQUENCE_NUMBER_FIELD_NAME, VALUE_KIND_FIELD_NAME,
};
use crate::table::prepared_files::PreparedFiles;
@@ -408,17 +408,11 @@ impl KeyValueFileWriter {
write.file_ordinal,
write.file_format,
);
- let bucket_dir = if self.config.partition_path.is_empty() {
- format!(
- "{}/bucket-{}",
- self.config.table_location, self.config.bucket
- )
- } else {
- format!(
- "{}/{}/bucket-{}",
- self.config.table_location, self.config.partition_path,
self.config.bucket
- )
- };
+ let bucket_dir = bucket_path_under(
+ &self.config.table_location,
+ &self.config.partition_path,
+ self.config.bucket,
+ );
self.file_io.mkdirs(&format!("{bucket_dir}/")).await?;
let file_path = format!("{bucket_dir}/{file_name}");
let output = self.file_io.new_output(&file_path)?;
diff --git a/crates/paimon/src/table/postpone_file_writer.rs
b/crates/paimon/src/table/postpone_file_writer.rs
index 978649cd..e651b3ed 100644
--- a/crates/paimon/src/table/postpone_file_writer.rs
+++ b/crates/paimon/src/table/postpone_file_writer.rs
@@ -27,7 +27,7 @@
use crate::arrow::format::{create_format_writer, FormatFileWriter};
use crate::io::FileIO;
use crate::spec::stats::BinaryTableStats;
-use crate::spec::{bucket_dir_name, DataFileMeta, EMPTY_SERIALIZED_ROW,
VALUE_KIND_FIELD_NAME};
+use crate::spec::{bucket_path_under, DataFileMeta, EMPTY_SERIALIZED_ROW,
VALUE_KIND_FIELD_NAME};
use crate::table::kv_file_writer::build_physical_schema;
use crate::Result;
use arrow_array::{Int64Array, Int8Array, RecordBatch};
@@ -205,20 +205,11 @@ impl PostponeFileWriter {
self.written_files.len(),
self.config.file_format,
);
- let bucket_dir = if self.config.partition_path.is_empty() {
- format!(
- "{}/{}",
- self.config.table_location,
- bucket_dir_name(self.config.bucket)
- )
- } else {
- format!(
- "{}/{}/{}",
- self.config.table_location,
- self.config.partition_path,
- bucket_dir_name(self.config.bucket)
- )
- };
+ let bucket_dir = bucket_path_under(
+ &self.config.table_location,
+ &self.config.partition_path,
+ self.config.bucket,
+ );
self.file_io.mkdirs(&format!("{bucket_dir}/")).await?;
let physical_schema = build_physical_schema(&user_schema);
let file_path = format!("{bucket_dir}/{file_name}");
