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 aa60a6ac Honor data-file prefix in Python native writes (#932)
aa60a6ac is described below
commit aa60a6ac7cbe9299aca19effd9ba8301c1bc55da
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Sep 24 17:10:45 2026 +0800
Honor data-file prefix in Python native writes (#932)
---
bindings/python/src/table.rs | 5 ++
bindings/python/tests/test_write.py | 25 +++++++++
crates/paimon/src/api/rest_api.rs | 4 +-
crates/paimon/src/spec/core_options.rs | 12 +++++
crates/paimon/src/spec/manifest_file_meta.rs | 15 +++---
crates/paimon/src/spec/objects_file.rs | 60 ++++++++--------------
crates/paimon/src/spec/snapshot.rs | 8 +++
crates/paimon/src/table/data_file_writer.rs | 10 +++-
crates/paimon/src/table/kv_file_writer.rs | 4 +-
crates/paimon/src/table/postpone_file_writer.rs | 4 +-
crates/paimon/src/table/rest_env.rs | 5 ++
crates/paimon/src/table/snapshot_commit.rs | 18 +++++--
crates/paimon/src/table/table_commit.rs | 29 ++++++++++-
.../paimon/src/table/table_commit/parity_tests.rs | 2 +-
.../src/table/table_commit/recovery_tests.rs | 2 +-
crates/paimon/src/table/table_write.rs | 10 ++--
crates/paimon/src/table/write_builder.rs | 2 +-
17 files changed, 151 insertions(+), 64 deletions(-)
diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs
index 06c6316c..47701b5c 100644
--- a/bindings/python/src/table.rs
+++ b/bindings/python/src/table.rs
@@ -110,6 +110,11 @@ impl PyTable {
self.inner.location().to_string()
}
+ /// REST catalog table identity used by callers with an already resolved
schema.
+ fn rest_table_uuid(&self) -> Option<&str> {
+ self.inner.rest_env().map(|env| env.uuid())
+ }
+
fn schema(&self) -> PyTableSchema {
PyTableSchema::new(self.inner.schema().clone())
}
diff --git a/bindings/python/tests/test_write.py
b/bindings/python/tests/test_write.py
index f36fbfdd..052a2c11 100644
--- a/bindings/python/tests/test_write.py
+++ b/bindings/python/tests/test_write.py
@@ -64,6 +64,31 @@ def test_write_commit_read_roundtrip():
assert result == {"id": [1, 2, 3], "name": ["a", "b", "c"]}
+
[email protected]("primary_key", [False, True])
+def test_custom_data_file_prefix_matches_table_option(tmp_path, primary_key):
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": str(tmp_path)})
+ ctx.sql("CREATE SCHEMA paimon.wdb")
+ key = ", PRIMARY KEY (id)" if primary_key else ""
+ bucket = ", 'bucket' = '1'" if primary_key else ""
+ ctx.sql(
+ "CREATE TABLE paimon.wdb.t (id INT, name STRING{}) "
+ "WITH ('data-file.prefix' = 'custom-'{})".format(key, bucket)
+ )
+ table = _get_table(str(tmp_path))
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(_batch([1], ["a"]))
+ builder.new_commit().commit(writer.prepare_commit())
+
+ files = list(tmp_path.rglob("*.parquet"))
+ assert len(files) == 1
+ assert files[0].name.startswith("custom-")
+ assert pa.Table.from_batches(ctx.sql("SELECT id, name FROM
paimon.wdb.t")).to_pydict() == {
+ "id": [1], "name": ["a"]}
+
+
def test_write_multiple_batches():
with tempfile.TemporaryDirectory() as warehouse:
ctx = _make_empty_table(warehouse)
diff --git a/crates/paimon/src/api/rest_api.rs
b/crates/paimon/src/api/rest_api.rs
index 34d76f46..e664ace9 100644
--- a/crates/paimon/src/api/rest_api.rs
+++ b/crates/paimon/src/api/rest_api.rs
@@ -1040,6 +1040,7 @@ impl RESTApi {
&self,
identifier: &Identifier,
table_uuid: &str,
+ base_snapshot_uuid: Option<&str>,
snapshot: &Snapshot,
statistics: &[PartitionStatistics],
) -> Result<bool> {
@@ -1048,7 +1049,8 @@ impl RESTApi {
validate_non_empty_multi(&[(database, "database name"), (table, "table
name")])?;
let path = self.resource_paths.commit_table(database, table);
let request = serde_json::json!({
- "tableUuid": table_uuid,
+ "tableId": table_uuid,
+ "baseSnapshotUuid": base_snapshot_uuid,
"snapshot": snapshot,
"statistics": statistics,
});
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index 477b7891..86c28c1e 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -65,6 +65,7 @@ const COMMIT_MAX_RETRY_WAIT_OPTION: &str =
"commit.max-retry-wait";
const FILE_COMPRESSION_OPTION: &str = "file.compression";
const FILE_COMPRESSION_ZSTD_LEVEL_OPTION: &str = "file.compression.zstd-level";
const FILE_FORMAT_OPTION: &str = "file.format";
+const DATA_FILE_PREFIX_OPTION: &str = "data-file.prefix";
const VECTOR_FILE_FORMAT_OPTION: &str = "vector.file.format";
const VECTOR_TARGET_FILE_SIZE_OPTION: &str = "vector.target-file-size";
const CHANGELOG_FILE_PREFIX_OPTION: &str = "changelog-file.prefix";
@@ -1347,6 +1348,14 @@ impl<'a> CoreOptions<'a> {
.unwrap_or(1)
}
+ /// File name prefix for data files. Default is `"data-"`.
+ pub fn data_file_prefix(&self) -> &str {
+ self.options
+ .get(DATA_FILE_PREFIX_OPTION)
+ .map(String::as_str)
+ .unwrap_or("data-")
+ }
+
/// File name prefix for changelog files. Default is `"changelog-"`.
pub fn changelog_file_prefix(&self) -> &str {
self.options
@@ -2831,12 +2840,14 @@ mod tests {
]);
let default_core = CoreOptions::new(&default_options);
+ assert_eq!(default_core.data_file_prefix(), "data-");
assert_eq!(default_core.changelog_file_prefix(), "changelog-");
assert_eq!(default_core.changelog_file_format(), "avro");
assert_eq!(default_core.changelog_file_compression(), "snappy");
assert_eq!(default_core.changelog_file_stats_mode(), None);
let custom_options = HashMap::from([
+ (DATA_FILE_PREFIX_OPTION.to_string(), "files-".to_string()),
(
CHANGELOG_FILE_PREFIX_OPTION.to_string(),
"custom-".to_string(),
@@ -2856,6 +2867,7 @@ mod tests {
]);
let custom_core = CoreOptions::new(&custom_options);
+ assert_eq!(custom_core.data_file_prefix(), "files-");
assert_eq!(custom_core.changelog_file_prefix(), "custom-");
assert_eq!(custom_core.changelog_file_format(), "parquet");
assert_eq!(custom_core.changelog_file_compression(), "zstd");
diff --git a/crates/paimon/src/spec/manifest_file_meta.rs
b/crates/paimon/src/spec/manifest_file_meta.rs
index 0d37b714..c6ac3f26 100644
--- a/crates/paimon/src/spec/manifest_file_meta.rs
+++ b/crates/paimon/src/spec/manifest_file_meta.rs
@@ -104,11 +104,12 @@ pub struct ManifestFileMeta {
)]
max_row_id: Option<i64>,
- /// Common positive bucket count recorded by an external manifest writer.
- ///
- /// Rust consumes this field for manifest pruning but intentionally does
not
- /// serialize it into manifest lists.
- #[serde(rename = "_TOTAL_BUCKETS", default, skip_serializing)]
+ /// Common positive bucket count for entries in this manifest.
+ #[serde(
+ rename = "_TOTAL_BUCKETS",
+ default,
+ skip_serializing_if = "Option::is_none"
+ )]
total_buckets: Option<i32>,
/// Files owned by this manifest and sharing its lifecycle.
@@ -244,8 +245,7 @@ impl ManifestFileMeta {
self
}
- /// Attach external manifest metadata in read-path tests.
- #[cfg(test)]
+ /// Record a common positive bucket count for manifest pruning.
#[inline]
#[must_use]
pub(crate) fn with_total_buckets(mut self, total_buckets: Option<i32>) ->
Self {
@@ -355,6 +355,7 @@ pub const MANIFEST_FILE_META_SCHEMA: &str = r#"["null", {
{"name": "_MAX_LEVEL", "type": ["null", "int"], "default": null},
{"name": "_MIN_ROW_ID", "type": ["null", "long"], "default": null},
{"name": "_MAX_ROW_ID", "type": ["null", "long"], "default": null},
+ {"name": "_TOTAL_BUCKETS", "type": ["null", "int"], "default": null},
{"name": "_EXTRA_FILES", "type": ["null", {"type": "array", "items":
"string"}], "default": null}
]
}]"#;
diff --git a/crates/paimon/src/spec/objects_file.rs
b/crates/paimon/src/spec/objects_file.rs
index 8f5bfaad..c30ecc8f 100644
--- a/crates/paimon/src/spec/objects_file.rs
+++ b/crates/paimon/src/spec/objects_file.rs
@@ -199,6 +199,7 @@ mod tests {
"_MAX_LEVEL",
"_MIN_ROW_ID",
"_MAX_ROW_ID",
+ "_TOTAL_BUCKETS",
"_EXTRA_FILES",
],
);
@@ -244,9 +245,7 @@ mod tests {
}
#[test]
- fn test_read_manifest_file_meta_total_buckets_without_writing_it() {
- assert!(!MANIFEST_FILE_META_SCHEMA.contains("_TOTAL_BUCKETS"));
-
+ fn test_roundtrip_total_buckets_and_read_legacy_manifest_meta() {
let original = vec![ManifestFileMeta::new(
"manifest-java-0".to_string(),
1024,
@@ -255,48 +254,29 @@ mod tests {
BinaryTableStats::empty(),
0,
)
- .with_bucket_level_stats(Some(2), Some(2), Some(0), Some(0))];
+ .with_bucket_level_stats(Some(2), Some(2), Some(0), Some(0))
+ .with_total_buckets(Some(8))];
let bytes = to_avro_bytes(MANIFEST_FILE_META_SCHEMA,
&original).unwrap();
- let mut value = Reader::new(bytes.as_slice())
- .unwrap()
- .next()
- .unwrap()
- .unwrap();
- let fields = match &mut value {
- Value::Union(_, record) => match record.as_mut() {
- Value::Record(fields) => fields,
- other => panic!("Expected an Avro record, got {other:?}"),
- },
- other => panic!("Expected an Avro union, got {other:?}"),
- };
- let extra_files_index = fields
- .iter()
- .position(|(name, _)| name == "_EXTRA_FILES")
- .unwrap();
- fields.insert(
- extra_files_index,
- (
- "_TOTAL_BUCKETS".to_string(),
- Value::Union(1, Box::new(Value::Int(8))),
- ),
- );
+ let decoded =
from_avro_bytes_fast::<ManifestFileMeta>(&bytes).unwrap();
+ assert_eq!(decoded[0].total_buckets(), Some(8));
- let java_schema = MANIFEST_FILE_META_SCHEMA.replacen(
- r#"{"name": "_EXTRA_FILES", "type": ["null", {"type": "array",
"items": "string"}], "default": null}"#,
- concat!(
- r#"{"name": "_TOTAL_BUCKETS", "type": ["null", "int"],
"default": null},"#,
- "\n ",
- r#"{"name": "_EXTRA_FILES", "type": ["null", {"type": "array",
"items": "string"}], "default": null}"#
- ),
+ let legacy_schema = MANIFEST_FILE_META_SCHEMA.replacen(
+ " {\"name\": \"_TOTAL_BUCKETS\", \"type\": [\"null\",
\"int\"], \"default\": null},\n",
+ "",
1,
);
- let schema = Schema::parse_str(&java_schema).unwrap();
- let mut writer = Writer::new(&schema, Vec::new());
- writer.append(value).unwrap();
- let bytes = writer.into_inner().unwrap();
-
+ assert!(!legacy_schema.contains("_TOTAL_BUCKETS"));
+ let legacy = vec![ManifestFileMeta::new(
+ "manifest-legacy-0".to_string(),
+ 1024,
+ 5,
+ 0,
+ BinaryTableStats::empty(),
+ 0,
+ )];
+ let bytes = to_avro_bytes(&legacy_schema, &legacy).unwrap();
let decoded =
from_avro_bytes_fast::<ManifestFileMeta>(&bytes).unwrap();
- assert_eq!(decoded[0].total_buckets(), Some(8));
+ assert_eq!(decoded[0].total_buckets(), None);
}
#[test]
diff --git a/crates/paimon/src/spec/snapshot.rs
b/crates/paimon/src/spec/snapshot.rs
index 652d58a1..70b366db 100644
--- a/crates/paimon/src/spec/snapshot.rs
+++ b/crates/paimon/src/spec/snapshot.rs
@@ -54,6 +54,10 @@ impl std::fmt::Display for CommitKind {
pub struct Snapshot {
/// version of snapshot
version: i32,
+ /// Unique identity for optimistic REST catalog publication.
+ #[builder(default = None)]
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ uuid: Option<String>,
id: i64,
schema_id: i64,
/// a manifest list recording all changes from the previous snapshots
@@ -118,6 +122,10 @@ pub struct Snapshot {
}
impl Snapshot {
+ pub fn uuid(&self) -> Option<&str> {
+ self.uuid.as_deref()
+ }
+
/// Get the version of this snapshot.
#[inline]
pub fn version(&self) -> i32 {
diff --git a/crates/paimon/src/table/data_file_writer.rs
b/crates/paimon/src/table/data_file_writer.rs
index b25c4a04..6039da60 100644
--- a/crates/paimon/src/table/data_file_writer.rs
+++ b/crates/paimon/src/table/data_file_writer.rs
@@ -30,7 +30,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, DataField, DataFileMeta,
EMPTY_SERIALIZED_ROW};
+use crate::spec::{bucket_path_under, CoreOptions, DataField, DataFileMeta,
EMPTY_SERIALIZED_ROW};
use crate::Result;
use arrow_array::RecordBatch;
use chrono::Utc;
@@ -56,6 +56,7 @@ pub(crate) struct DataFileWriter {
file_compression_zstd_level: i32,
write_buffer_size: i64,
file_format: String,
+ data_file_prefix: String,
write_fields: Vec<DataField>,
format_options: HashMap<String, String>,
file_source: Option<i32>,
@@ -94,6 +95,9 @@ impl DataFileWriter {
first_row_id: Option<i64>,
write_cols: Option<Vec<String>>,
) -> Self {
+ let data_file_prefix = CoreOptions::new(&format_options)
+ .data_file_prefix()
+ .to_string();
Self {
file_io,
table_location,
@@ -105,6 +109,7 @@ impl DataFileWriter {
file_compression_zstd_level,
write_buffer_size,
file_format,
+ data_file_prefix,
write_fields,
format_options,
file_source,
@@ -182,7 +187,8 @@ impl DataFileWriter {
.map(|options| options.create_writer())
.transpose()?;
let file_name = format!(
- "data-{}-{}.{}",
+ "{}{}-{}.{}",
+ self.data_file_prefix,
uuid::Uuid::new_v4(),
self.written_files.len(),
self.file_format,
diff --git a/crates/paimon/src/table/kv_file_writer.rs
b/crates/paimon/src/table/kv_file_writer.rs
index 8cc7efbf..4f2987bb 100644
--- a/crates/paimon/src/table/kv_file_writer.rs
+++ b/crates/paimon/src/table/kv_file_writer.rs
@@ -80,6 +80,7 @@ pub(crate) struct KeyValueWriteConfig {
pub file_compression_zstd_level: i32,
pub write_buffer_size: i64,
pub file_format: String,
+ pub data_file_prefix: String,
pub input_changelog: bool,
pub changelog_file_prefix: String,
pub changelog_file_compression: String,
@@ -324,7 +325,7 @@ impl KeyValueFileWriter {
data_seq.as_ref(),
&data_indices,
IndexedFileWrite {
- file_prefix: "data-",
+ file_prefix: &self.config.data_file_prefix,
file_ordinal: self.written_files.len(),
file_format: &self.config.file_format,
file_compression: &self.config.file_compression,
@@ -1014,6 +1015,7 @@ mod tests {
file_compression_zstd_level: 0,
write_buffer_size: 1024,
file_format: "parquet".to_string(),
+ data_file_prefix: "data-".to_string(),
input_changelog: false,
changelog_file_prefix: "changelog-".to_string(),
changelog_file_compression: "none".to_string(),
diff --git a/crates/paimon/src/table/postpone_file_writer.rs
b/crates/paimon/src/table/postpone_file_writer.rs
index c683a99c..0329a421 100644
--- a/crates/paimon/src/table/postpone_file_writer.rs
+++ b/crates/paimon/src/table/postpone_file_writer.rs
@@ -20,7 +20,7 @@
//! Writes data in KV format (`_SEQUENCE_NUMBER`, `_VALUE_KIND` + user columns)
//! but without sorting or deduplication — compaction assigns real buckets
later.
//!
-//! Uses a special file naming prefix: `data-u-{commitUser}-s-0-w-`.
+//! Uses a special file naming prefix: `data--u-{commitUser}-s-0-w-`.
//!
//! Reference:
[PostponeBucketWriter](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/sink/PostponeBucketWriter.java)
@@ -47,7 +47,7 @@ pub(crate) struct PostponeWriteConfig {
pub file_compression_zstd_level: i32,
pub write_buffer_size: i64,
pub file_format: String,
- /// Data file name prefix: `"data-u-{commitUser}-s-0-w-"`.
+ /// Data file name prefix: `"data--u-{commitUser}-s-0-w-"`.
pub data_file_prefix: String,
}
diff --git a/crates/paimon/src/table/rest_env.rs
b/crates/paimon/src/table/rest_env.rs
index e814f966..6440536f 100644
--- a/crates/paimon/src/table/rest_env.rs
+++ b/crates/paimon/src/table/rest_env.rs
@@ -86,6 +86,11 @@ impl RESTEnv {
&self.identifier
}
+ /// Catalog identity used to reject commits against a replaced table.
+ pub fn uuid(&self) -> &str {
+ &self.uuid
+ }
+
/// Load a table through the same REST catalog environment.
pub async fn get_table(&self, identifier: &Identifier) -> Result<Table> {
Self::load_table(
diff --git a/crates/paimon/src/table/snapshot_commit.rs
b/crates/paimon/src/table/snapshot_commit.rs
index 9d141f79..942e6d77 100644
--- a/crates/paimon/src/table/snapshot_commit.rs
+++ b/crates/paimon/src/table/snapshot_commit.rs
@@ -36,8 +36,12 @@ use std::sync::Arc;
pub trait SnapshotCommit: Send + Sync {
/// Commit the given snapshot. Returns true if successful, false if
/// another writer won the race.
- async fn commit(&self, snapshot: &Snapshot, statistics:
&[PartitionStatistics])
- -> Result<bool>;
+ async fn commit(
+ &self,
+ base_snapshot_uuid: Option<&str>,
+ snapshot: &Snapshot,
+ statistics: &[PartitionStatistics],
+ ) -> Result<bool>;
}
/// A SnapshotCommit using file renaming to commit.
@@ -57,6 +61,7 @@ impl RenamingSnapshotCommit {
impl SnapshotCommit for RenamingSnapshotCommit {
async fn commit(
&self,
+ _base_snapshot_uuid: Option<&str>,
snapshot: &Snapshot,
_statistics: &[PartitionStatistics],
) -> Result<bool> {
@@ -88,11 +93,18 @@ impl RESTSnapshotCommit {
impl SnapshotCommit for RESTSnapshotCommit {
async fn commit(
&self,
+ base_snapshot_uuid: Option<&str>,
snapshot: &Snapshot,
statistics: &[PartitionStatistics],
) -> Result<bool> {
self.api
- .commit_snapshot(&self.identifier, &self.uuid, snapshot,
statistics)
+ .commit_snapshot(
+ &self.identifier,
+ &self.uuid,
+ base_snapshot_uuid,
+ snapshot,
+ statistics,
+ )
.await
}
}
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index b1d399b2..183f2926 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -1137,7 +1137,12 @@ impl TableCommit {
};
// Once publication starts its outcome may be unknown. These files must
// remain available even if the response is lost or a later retry
fails.
- let publication_error = match self.snapshot_commit.commit(&snapshot,
&statistics).await {
+ let base_snapshot_uuid =
latest_snapshot.as_ref().and_then(Snapshot::uuid);
+ let publication_error = match self
+ .snapshot_commit
+ .commit(base_snapshot_uuid, &snapshot, &statistics)
+ .await
+ {
Ok(true) => return Ok(CommitAttemptResult::Success),
Ok(false) => None,
Err(error) => Some(error),
@@ -1297,6 +1302,7 @@ impl TableCommit {
.await?;
let snapshot = Snapshot::builder()
.version(3)
+ .uuid(Some(uuid::Uuid::new_v4().to_string()))
.id(new_snapshot_id)
.schema_id(schema_id)
.base_manifest_list(base_manifest_list_name)
@@ -1525,6 +1531,8 @@ impl TableCommit {
let mut min_row_id: Option<i64> = None;
let mut max_row_id: Option<i64> = None;
let mut all_entries_have_row_id = !entries.is_empty();
+ let mut total_buckets: Option<i32> = None;
+ let mut total_buckets_known = true;
let mut schema_id = self.table.schema().id();
for entry in entries {
match entry.kind() {
@@ -1533,6 +1541,12 @@ impl TableCommit {
}
schema_id = schema_id.max(entry.file().schema_id);
let b = entry.bucket();
+ let candidate = entry.total_buckets();
+ if candidate <= 0 || total_buckets.is_some_and(|value| value !=
candidate) {
+ total_buckets_known = false;
+ } else {
+ total_buckets = Some(candidate);
+ }
min_bucket = Some(min_bucket.map_or(b, |cur| cur.min(b)));
max_bucket = Some(max_bucket.map_or(b, |cur| cur.max(b)));
let l = entry.file().level;
@@ -1561,6 +1575,11 @@ impl TableCommit {
schema_id,
)
.with_bucket_level_stats(min_bucket, max_bucket, min_level, max_level)
+ .with_total_buckets(if total_buckets_known {
+ total_buckets
+ } else {
+ None
+ })
.with_row_id_stats(min_row_id, max_row_id)
.with_extra_files(sidecar_name.map(|name| vec![name])))
}
@@ -6741,7 +6760,12 @@ mod tests {
let table_path = "memory:/test_commit_bucket_level_stats";
setup_dirs(&file_io, table_path).await;
- let commit = setup_commit(&file_io, table_path);
+ let table = test_table_with_options(
+ &file_io,
+ table_path,
+ HashMap::from([("bucket".to_string(), "8".to_string())]),
+ );
+ let commit = TableCommit::new(table, "test-user".to_string());
fn data_file_at_level(name: &str, level: i32) -> DataFileMeta {
let mut f = test_data_file(name, 1);
@@ -6768,6 +6792,7 @@ mod tests {
);
assert_eq!(metas[0].min_bucket(), Some(0));
assert_eq!(metas[0].max_bucket(), Some(3));
+ assert_eq!(metas[0].total_buckets(), Some(commit.total_buckets));
assert_eq!(metas[0].min_level(), Some(0));
assert_eq!(metas[0].max_level(), Some(2));
}
diff --git a/crates/paimon/src/table/table_commit/parity_tests.rs
b/crates/paimon/src/table/table_commit/parity_tests.rs
index 8799eddd..bdd9c4d2 100644
--- a/crates/paimon/src/table/table_commit/parity_tests.rs
+++ b/crates/paimon/src/table/table_commit/parity_tests.rs
@@ -307,7 +307,7 @@ struct LostResponseCommit {
#[async_trait::async_trait]
impl SnapshotCommit for LostResponseCommit {
- async fn commit(&self, snapshot: &Snapshot, _: &[PartitionStatistics]) ->
Result<bool> {
+ async fn commit(&self, _: Option<&str>, snapshot: &Snapshot, _:
&[PartitionStatistics]) -> Result<bool> {
let attempt = self.calls.fetch_add(1,
std::sync::atomic::Ordering::SeqCst);
if attempt == 0 {
if self.publish_first {
diff --git a/crates/paimon/src/table/table_commit/recovery_tests.rs
b/crates/paimon/src/table/table_commit/recovery_tests.rs
index 631a5ce9..b202f100 100644
--- a/crates/paimon/src/table/table_commit/recovery_tests.rs
+++ b/crates/paimon/src/table/table_commit/recovery_tests.rs
@@ -399,7 +399,7 @@ struct ConcurrentDvCommit {
#[async_trait::async_trait]
impl SnapshotCommit for ConcurrentDvCommit {
- async fn commit(&self, snapshot: &Snapshot, _: &[PartitionStatistics]) ->
Result<bool> {
+ async fn commit(&self, _: Option<&str>, snapshot: &Snapshot, _:
&[PartitionStatistics]) -> Result<bool> {
if self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
let other = TableCommit::new(self.table.clone(),
"concurrent".into());
match self.change {
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 54eb6375..ced9a781 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -136,6 +136,7 @@ pub struct TableWrite {
file_compression_zstd_level: i32,
write_buffer_size: i64,
file_format: String,
+ data_file_prefix: String,
primary_key_indices: Vec<usize>,
primary_key_types: Vec<DataType>,
sequence_field_indices: Vec<usize>,
@@ -251,6 +252,7 @@ impl TableWrite {
let file_compression = core_options.file_compression().to_string();
let file_compression_zstd_level =
core_options.file_compression_zstd_level();
let file_format = core_options.file_format().to_string();
+ let data_file_prefix = core_options.data_file_prefix().to_string();
let vector_file_format = core_options.vector_file_format();
let changelog_file_prefix =
core_options.changelog_file_prefix().to_string();
let changelog_file_format =
core_options.changelog_file_format().to_string();
@@ -419,6 +421,7 @@ impl TableWrite {
file_compression_zstd_level,
write_buffer_size,
file_format,
+ data_file_prefix,
primary_key_indices,
primary_key_types,
sequence_field_indices,
@@ -1118,7 +1121,7 @@ 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 {
- let data_file_prefix = format!("data-u-{}-s-0-w-", self.commit_user);
+ let data_file_prefix = format!("{}-u-{}-s-0-w-",
self.data_file_prefix, self.commit_user);
FileWriter::Postpone(
PostponeFileWriter::new(
self.table.file_io().clone(),
@@ -1178,6 +1181,7 @@ impl TableWrite {
file_compression_zstd_level:
self.file_compression_zstd_level,
write_buffer_size: self.write_buffer_size,
file_format: self.file_format.clone(),
+ data_file_prefix: self.data_file_prefix.clone(),
input_changelog: self.changelog_producer ==
ChangelogProducer::Input
&& !self.is_overwrite,
changelog_file_prefix: self.changelog_file_prefix.clone(),
@@ -4281,9 +4285,9 @@ pub(in crate::table) mod tests {
let messages = table_write.prepare_commit().await.unwrap();
let file = &messages[0].new_files[0];
- // Verify postpone file naming:
data-u-{commitUser}-s-{writeId}-w-{uuid}-{index}.parquet
+ // Verify postpone file naming:
data--u-{commitUser}-s-{writeId}-w-{uuid}-{index}.parquet
assert!(
- file.file_name.starts_with("data-u-my-commit-user-s-"),
+ file.file_name.starts_with("data--u-my-commit-user-s-"),
"Expected postpone file prefix, got: {}",
file.file_name
);
diff --git a/crates/paimon/src/table/write_builder.rs
b/crates/paimon/src/table/write_builder.rs
index 07ff11e0..ea365600 100644
--- a/crates/paimon/src/table/write_builder.rs
+++ b/crates/paimon/src/table/write_builder.rs
@@ -443,7 +443,7 @@ mod tests {
assert!(
messages[0].new_files[0]
.file_name
- .starts_with("data-u-my-commit-user-s-"),
+ .starts_with("data--u-my-commit-user-s-"),
"Expected custom commit user in file name, got: {}",
messages[0].new_files[0].file_name
);
