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 8cbafd4d Support native Format Table append and overwrite writes (#948)
8cbafd4d is described below
commit 8cbafd4d68b6ebd65ef2dedd1a5f8a60bec5227a
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Sep 25 08:06:11 2026 +0800
Support native Format Table append and overwrite writes (#948)
---
.../datafusion/tests/rest_format_table_dml.rs | 205 +++++-
crates/paimon/src/api/api_request.rs | 10 +
crates/paimon/src/api/rest_api.rs | 33 +
crates/paimon/src/io/file_io.rs | 56 ++
crates/paimon/src/spec/schema.rs | 12 +
crates/paimon/src/table/commit_message.rs | 22 +
crates/paimon/src/table/format_partition.rs | 56 ++
crates/paimon/src/table/format_table_commit.rs | 605 +++++++++++++++++
crates/paimon/src/table/format_table_scan.rs | 10 +-
.../paimon/src/table/format_table_write_tests.rs | 717 +++++++++++++++++++++
crates/paimon/src/table/format_table_writer.rs | 475 ++++++++++++++
crates/paimon/src/table/format_write_builder.rs | 19 +-
crates/paimon/src/table/mod.rs | 4 +
crates/paimon/src/table/table_commit.rs | 26 +
crates/paimon/src/table/table_write.rs | 80 +++
crates/paimon/tests/mock_server.rs | 34 +-
crates/paimon/tests/rest_catalog_test.rs | 2 +-
17 files changed, 2349 insertions(+), 17 deletions(-)
diff --git a/crates/integrations/datafusion/tests/rest_format_table_dml.rs
b/crates/integrations/datafusion/tests/rest_format_table_dml.rs
index 4d08fa80..3494b2ba 100644
--- a/crates/integrations/datafusion/tests/rest_format_table_dml.rs
+++ b/crates/integrations/datafusion/tests/rest_format_table_dml.rs
@@ -15,7 +15,8 @@
// specific language governing permissions and limitations
// under the License.
-//! Row-level DML on a Format Table, which the Rust client cannot write.
+//! Format Table inserts and the row-level mutations that still need a
+//! dedicated copy-on-write implementation.
mod common;
@@ -45,7 +46,7 @@ const WAREHOUSE: &str = "test_warehouse";
/// A Format Table with one file in each of `dt=a` and `dt=b`, whose
partitions a mock REST
/// catalog manages. Its data directory is the root of `temp_dir`.
-async fn catalog_managed_format_table(temp_dir: &TempDir) -> SQLContext {
+async fn catalog_managed_format_table(temp_dir: &TempDir) -> (SQLContext,
mock_server::RESTServer) {
let server = start_mock_server(
WAREHOUSE.to_string(),
temp_dir.path().to_string_lossy().into_owned(),
@@ -81,7 +82,7 @@ async fn catalog_managed_format_table(temp_dir: &TempDir) ->
SQLContext {
let catalog = Arc::new(RESTCatalog::new(options, true).await.unwrap());
let mut context = SQLContext::new();
context.register_catalog("paimon", catalog).await.unwrap();
- context
+ (context, server)
}
/// The same table with its partitions discovered from the directory layout,
in a filesystem
@@ -205,7 +206,7 @@ fn statements(table_name: &str) -> [String; 3] {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_row_level_dml_is_refused_on_a_catalog_managed_format_table() {
let temp_dir = tempfile::tempdir().unwrap();
- let context = catalog_managed_format_table(&temp_dir).await;
+ let (context, _server) = catalog_managed_format_table(&temp_dir).await;
let seeded = files(temp_dir.path());
for statement in statements(TABLE_NAME) {
@@ -230,3 +231,199 @@ async fn
test_row_level_dml_is_refused_on_a_format_table_without_catalog_managed
assert_eq!(ids(&context, &table_name).await, [1, 2, 3], "{statement}");
}
}
+
+#[cfg(not(windows))]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_insert_into_catalog_managed_format_table_registers_partition() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let (context, _server) = catalog_managed_format_table(&temp_dir).await;
+ context
+ .sql(&format!(
+ "INSERT INTO {TABLE_NAME} (dt, id) VALUES ('a', 10), ('c', 11)"
+ ))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(ids(&context, TABLE_NAME).await, [1, 2, 3, 10, 11]);
+ assert_eq!(
+ files(&temp_dir.path().join("dt=c"))
+ .iter()
+ .filter(|file| file.ends_with(".parquet"))
+ .count(),
+ 1
+ );
+}
+
+#[cfg(not(windows))]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_append_keeps_visible_files_when_statistics_report_fails() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let (context, server) = catalog_managed_format_table(&temp_dir).await;
+ server.set_create_partitions_statistics_error_status(Some(
+ axum::http::StatusCode::INTERNAL_SERVER_ERROR,
+ ));
+ context
+ .sql(&format!(
+ "INSERT INTO {TABLE_NAME} (dt, id) VALUES ('c', 11)"
+ ))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(ids(&context, TABLE_NAME).await, [1, 2, 3, 11]);
+ let calls = server.create_partitions_calls();
+ assert_eq!(calls.len(), 2);
+ assert!(calls[0].2.partition_statistics.is_none());
+ assert!(calls[1].2.partition_statistics.is_some());
+ assert_eq!(
+ files(&temp_dir.path().join("dt=c"))
+ .iter()
+ .filter(|file| file.ends_with(".parquet"))
+ .count(),
+ 1
+ );
+}
+
+#[cfg(not(windows))]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_failed_partition_preflight_discards_staged_files() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let (context, server) = catalog_managed_format_table(&temp_dir).await;
+ let seeded = files(temp_dir.path());
+ server.set_list_partitions_by_names_error_status(Some(
+ axum::http::StatusCode::SERVICE_UNAVAILABLE,
+ ));
+ let error = context
+ .sql(&format!(
+ "INSERT INTO {TABLE_NAME} (dt, id) VALUES ('c', 11)"
+ ))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .err()
+ .unwrap();
+ assert!(error.to_string().contains("Service unavailable"), "{error}");
+ assert_eq!(files(temp_dir.path()), seeded);
+}
+
+#[cfg(not(windows))]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn
test_overwrite_rebinds_custom_partition_without_deleting_external_data() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let (context, server) = catalog_managed_format_table(&temp_dir).await;
+ let external = temp_dir.path().join("external-a");
+ write_ids(&external, &[99]);
+ let spec = HashMap::from([("dt".to_string(), "a".to_string())]);
+ server.set_table_partition_options(
+ DATABASE,
+ TABLE,
+ &spec,
+ HashMap::from([("path".to_string(), format!("file://{}",
external.display()))]),
+ );
+ context
+ .sql(&format!("INSERT OVERWRITE {TABLE_NAME} VALUES ('a', 10)"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(ids(&context, TABLE_NAME).await, [3, 10]);
+ assert_eq!(files(&external), ["part-0.parquet"]);
+ let partition = server
+ .table_partitions(DATABASE, TABLE)
+ .into_iter()
+ .find(|partition| partition.spec == spec)
+ .unwrap();
+ let path = partition.options.unwrap().remove("path").unwrap();
+ assert!(path.ends_with("/dt=a"), "{path}");
+ let calls = server.create_partitions_calls();
+ assert_eq!(calls.len(), 1);
+ assert_eq!(calls[0].2.replace_statistics, Some(true));
+ assert!(calls[0].2.partition_options.is_some());
+}
+
+#[cfg(not(windows))]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_insert_into_directory_format_table_publishes_visible_files() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let (table_dir, context) =
directory_partitioned_format_table(&temp_dir).await;
+ let table_name = format!("paimon.{DATABASE}.{TABLE}");
+ context
+ .sql(&format!(
+ "INSERT INTO {table_name} (dt, id) VALUES ('b', 10), ('c', 11)"
+ ))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(ids(&context, &table_name).await, [1, 2, 3, 10, 11]);
+ assert_eq!(
+ files(&table_dir.join("dt=b"))
+ .iter()
+ .filter(|file| file.ends_with(".parquet"))
+ .count(),
+ 2
+ );
+ assert_eq!(
+ files(&table_dir.join("dt=c"))
+ .iter()
+ .filter(|file| file.ends_with(".parquet"))
+ .count(),
+ 1
+ );
+}
+
+#[cfg(not(windows))]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_overwrite_only_replaces_touched_catalog_partition() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let (context, _server) = catalog_managed_format_table(&temp_dir).await;
+ context
+ .sql(&format!(
+ "INSERT OVERWRITE {TABLE_NAME} VALUES ('a', 10), ('a', 11)"
+ ))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(ids(&context, TABLE_NAME).await, [3, 10, 11]);
+ assert_eq!(
+ files(&temp_dir.path().join("dt=a"))
+ .iter()
+ .filter(|file| file.ends_with(".parquet"))
+ .count(),
+ 1
+ );
+ assert_eq!(
+ files(&temp_dir.path().join("dt=b"))
+ .iter()
+ .filter(|file| file.ends_with(".parquet"))
+ .count(),
+ 1
+ );
+}
+
+#[cfg(not(windows))]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_static_partition_overwrite_with_empty_source_keeps_other_data() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let (_table_dir, context) =
directory_partitioned_format_table(&temp_dir).await;
+ let table_name = format!("paimon.{DATABASE}.{TABLE}");
+ context
+ .sql(&format!(
+ "INSERT OVERWRITE {table_name} PARTITION (dt = 'a') \
+ SELECT CAST(0 AS BIGINT) AS id WHERE FALSE"
+ ))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(ids(&context, &table_name).await, [3]);
+}
diff --git a/crates/paimon/src/api/api_request.rs
b/crates/paimon/src/api/api_request.rs
index 7cc1febd..679ddeb8 100644
--- a/crates/paimon/src/api/api_request.rs
+++ b/crates/paimon/src/api/api_request.rs
@@ -209,6 +209,9 @@ pub struct CreatePartitionsRequest {
/// present only together with `partition_statistics`.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub replace_statistics: Option<bool>,
+ /// Per-partition catalog options, aligned with `partition_specs`.
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub partition_options: Option<Vec<HashMap<String, String>>>,
}
impl CreatePartitionsRequest {
@@ -219,6 +222,7 @@ impl CreatePartitionsRequest {
ignore_if_exists,
partition_statistics: None,
replace_statistics: None,
+ partition_options: None,
}
}
@@ -228,6 +232,12 @@ impl CreatePartitionsRequest {
self.replace_statistics = Some(replace);
self
}
+
+ /// Set catalog options for each partition in the request.
+ pub fn with_partition_options(mut self, options: Vec<HashMap<String,
String>>) -> Self {
+ self.partition_options = Some(options);
+ self
+ }
}
/// Request to drop (unregister) table partitions.
diff --git a/crates/paimon/src/api/rest_api.rs
b/crates/paimon/src/api/rest_api.rs
index e664ace9..7e435ae1 100644
--- a/crates/paimon/src/api/rest_api.rs
+++ b/crates/paimon/src/api/rest_api.rs
@@ -624,15 +624,48 @@ impl RESTApi {
ignore_if_exists: bool,
statistics: Option<Vec<PartitionStatistics>>,
replace_statistics: bool,
+ ) -> Result<()> {
+ self.create_partitions_with_options(
+ identifier,
+ partition_specs,
+ ignore_if_exists,
+ statistics,
+ replace_statistics,
+ None,
+ )
+ .await
+ }
+
+ /// Create partitions with aligned statistics and per-partition catalog
options.
+ pub async fn create_partitions_with_options(
+ &self,
+ identifier: &Identifier,
+ partition_specs: Vec<HashMap<String, String>>,
+ ignore_if_exists: bool,
+ statistics: Option<Vec<PartitionStatistics>>,
+ replace_statistics: bool,
+ partition_options: Option<Vec<HashMap<String, String>>>,
) -> Result<()> {
let database = identifier.database();
let table = identifier.object();
validate_non_empty_multi(&[(database, "database name"), (table, "table
name")])?;
+ if partition_options
+ .as_ref()
+ .is_some_and(|options| options.len() != partition_specs.len())
+ {
+ return Err(crate::Error::DataInvalid {
+ message: "Partition options must align with partition
specs".into(),
+ source: None,
+ });
+ }
let path = self.resource_paths.partitions(database, table);
let mut request = CreatePartitionsRequest::new(partition_specs,
ignore_if_exists);
if let Some(statistics) = statistics {
request = request.with_statistics(statistics, replace_statistics);
}
+ if let Some(options) = partition_options {
+ request = request.with_partition_options(options);
+ }
let _resp: serde_json::Value = self.client.post(&path,
&request).await?;
Ok(())
}
diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs
index 720225eb..bc8e2fb5 100644
--- a/crates/paimon/src/io/file_io.rs
+++ b/crates/paimon/src/io/file_io.rs
@@ -459,6 +459,37 @@ impl FileIO {
Ok(())
}
+ /// Copy a large file without materializing its entire contents in memory.
+ /// Format Table publication uses this when a backend does not support
+ /// rename, as is common for object stores and the in-memory test backend.
+ pub async fn copy_file_streaming(&self, src: &str, dst: &str) ->
Result<()> {
+ const CHUNK_SIZE: u64 = 8 * 1024 * 1024;
+ let input = self.new_input(src)?;
+ let size = input.metadata().await?.size;
+ let reader = input.reader().await?;
+ let output = self.new_output(dst)?;
+ let mut writer = output.writer().await?;
+ let mut position = 0;
+ while position < size {
+ let end = (position + CHUNK_SIZE).min(size);
+ let bytes = reader.read(position..end).await?;
+ if bytes.len() as u64 != end - position {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Short read while copying '{src}': expected {} bytes,
got {}",
+ end - position,
+ bytes.len()
+ ),
+ source: None,
+ });
+ }
+ writer.write(bytes).await?;
+ position = end;
+ }
+ writer.close().await?;
+ Ok(())
+ }
+
/// Renames the file/directory src to dst.
///
/// Reference:
<https://github.com/apache/paimon/blob/release-0.8.2/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java#L159>
@@ -1901,6 +1932,31 @@ mod input_output_test {
);
}
+ #[tokio::test]
+ async fn test_streaming_copy_crosses_chunk_boundary() {
+ let file_io = setup_memory_file_io();
+ let source = "memory:/format-copy/source.parquet";
+ let target = "memory:/format-copy/target.parquet";
+ let mut payload = vec![0u8; 8 * 1024 * 1024 + 17];
+ payload[0] = 3;
+ payload[8 * 1024 * 1024 - 1] = 7;
+ payload[8 * 1024 * 1024] = 11;
+ payload[8 * 1024 * 1024 + 16] = 13;
+ file_io
+ .new_output(source)
+ .unwrap()
+ .write(Bytes::from(payload.clone()))
+ .await
+ .unwrap();
+
+ file_io.copy_file_streaming(source, target).await.unwrap();
+ assert_eq!(
+ file_io.new_input(target).unwrap().read().await.unwrap(),
+ Bytes::from(payload)
+ );
+ assert!(file_io.exists(source).await.unwrap());
+ }
+
#[cfg(not(windows))]
#[tokio::test]
async fn
test_file_io_local_cache_invalidates_source_and_target_after_rename() {
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index 11dc4365..56651d9f 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -1075,6 +1075,8 @@ pub struct DataField {
typ: DataType,
#[serde(skip_serializing_if = "Option::is_none")]
description: Option<String>,
+ #[serde(rename = "defaultValue", skip_serializing_if = "Option::is_none")]
+ default_value: Option<String>,
}
impl DataField {
@@ -1084,6 +1086,7 @@ impl DataField {
name,
typ,
description: None,
+ default_value: None,
}
}
@@ -1103,6 +1106,10 @@ impl DataField {
self.description.as_deref()
}
+ pub fn default_value(&self) -> Option<&str> {
+ self.default_value.as_deref()
+ }
+
pub fn with_id(mut self, new_id: i32) -> Self {
self.id = new_id;
self
@@ -1117,6 +1124,11 @@ impl DataField {
self.description = new_description;
self
}
+
+ pub fn with_default_value(mut self, default_value: Option<String>) -> Self
{
+ self.default_value = default_value;
+ self
+ }
}
/// Quote an identifier the way Java `EncodingUtils.escapeIdentifier` does:
wrap it
diff --git a/crates/paimon/src/table/commit_message.rs
b/crates/paimon/src/table/commit_message.rs
index f4e86ba0..29c904d1 100644
--- a/crates/paimon/src/table/commit_message.rs
+++ b/crates/paimon/src/table/commit_message.rs
@@ -17,6 +17,7 @@
use super::source::{read_i32, read_i64, take};
use crate::spec::{BinaryRow, DataFileMeta, DataFileMetaRowLayout,
IndexFileMeta};
+use std::collections::HashMap;
/// Current Java `CommitMessageSerializer` body version. The version is carried
/// by an enclosing serializer, not embedded in the body.
@@ -71,6 +72,9 @@ fn read_rows<T>(
/// Reference:
[org.apache.paimon.table.sink.CommitMessage](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/sink/CommitMessageImpl.java)
#[derive(Debug, Clone)]
pub struct CommitMessage {
+ /// A staged Format Table file. Format Tables have no manifest or snapshot;
+ /// this message is published by the Format Table committer instead.
+ pub(crate) format_file: Option<FormatFileCommit>,
/// Binary row bytes for the partition.
pub partition: Vec<u8>,
/// Bucket id.
@@ -102,6 +106,7 @@ pub struct CommitMessage {
impl CommitMessage {
pub fn new(partition: Vec<u8>, bucket: i32, new_files: Vec<DataFileMeta>)
-> Self {
Self {
+ format_file: None,
partition,
bucket,
total_buckets: None,
@@ -132,6 +137,12 @@ impl CommitMessage {
/// Write the unframed Java v14 `CommitMessageSerializer.serialize` body.
pub fn serialize(&self) -> crate::Result<Vec<u8>> {
+ if self.format_file.is_some() {
+ return Err(crate::Error::Unsupported {
+ message: "Format Table two-phase file messages use a different
Java serializer"
+ .into(),
+ });
+ }
let mut out = Vec::new();
// The partition normally is SerializationUtils.serializeBinaryRow:
// i32 arity followed by a raw BinaryRow. Internal unpartitioned
writers
@@ -225,6 +236,7 @@ impl CommitMessage {
)));
}
Ok(Self {
+ format_file: None,
partition,
bucket,
total_buckets,
@@ -256,6 +268,16 @@ impl CommitMessage {
}
}
+/// A file prepared below `_temporary`, awaiting publish into its partition.
+#[derive(Debug, Clone)]
+pub(crate) struct FormatFileCommit {
+ pub staged_path: String,
+ pub target_path: String,
+ pub partition: HashMap<String, String>,
+ pub record_count: i64,
+ pub file_size: i64,
+}
+
#[cfg(test)]
mod tests {
use super::*;
diff --git a/crates/paimon/src/table/format_partition.rs
b/crates/paimon/src/table/format_partition.rs
index 0f27e9a0..748c8b28 100644
--- a/crates/paimon/src/table/format_partition.rs
+++ b/crates/paimon/src/table/format_partition.rs
@@ -111,6 +111,44 @@ impl FormatTablePartitionPaths {
.join("/"))
}
+ /// Physical path for a leading static partition prefix. Java creates this
+ /// directory even when an overwrite writes no rows. An empty spec denotes
+ /// the table root; a gap or an unknown key is rejected.
+ pub(crate) fn relative_prefix_path(
+ &self,
+ spec: &HashMap<String, String>,
+ ) -> crate::Result<String> {
+ if spec.len() > self.partition_keys.len() {
+ return Err(crate::Error::DataInvalid {
+ message: "Static partition is not a leading prefix".into(),
+ source: None,
+ });
+ }
+ let mut segments = Vec::with_capacity(spec.len());
+ for key in self.partition_keys.iter().take(spec.len()) {
+ let value = spec.get(key).ok_or_else(|| crate::Error::DataInvalid {
+ message: format!("Static partition is missing leading key
'{key}'"),
+ source: None,
+ })?;
+ if value.is_empty() || (self.only_value_in_path &&
matches!(value.as_str(), "." | ".."))
+ {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Partition value {value:?} cannot be used as a path
component"
+ ),
+ source: None,
+ });
+ }
+ let segment = if self.only_value_in_path {
+ escape_path_name(value)
+ } else {
+ format!("{}={}", escape_path_name(key),
escape_path_name(value))
+ };
+ segments.push(segment);
+ }
+ Ok(segments.join("/"))
+ }
+
/// Discover complete raw partition specs from the table directory, sorted
and deduplicated.
/// Skips hidden or non-matching entries; a malformed or non-canonical
segment is an error.
pub async fn discover(
@@ -421,6 +459,24 @@ mod tests {
assert!(value_only.relative_path(&traversal).is_err());
}
+ #[test]
+ fn test_relative_static_partition_prefix_path() {
+ let keyed = FormatTablePartitionPaths::new(["dt", "hour"], false);
+ let values = FormatTablePartitionPaths::new(["dt", "hour"], true);
+ let empty = HashMap::new();
+ assert_eq!(keyed.relative_prefix_path(&empty).unwrap(), "");
+ assert_eq!(values.relative_prefix_path(&empty).unwrap(), "");
+ let prefix = HashMap::from([("dt".to_string(), "a/b".to_string())]);
+ assert_eq!(keyed.relative_prefix_path(&prefix).unwrap(), "dt=a%2Fb");
+ assert_eq!(values.relative_prefix_path(&prefix).unwrap(), "a%2Fb");
+ let gap = HashMap::from([("hour".to_string(), "12".to_string())]);
+ assert!(keyed.relative_prefix_path(&gap).is_err());
+ let unknown = HashMap::from([("other".to_string(), "x".to_string())]);
+ assert!(values.relative_prefix_path(&unknown).is_err());
+ let invalid = HashMap::from([("dt".to_string(), "..".to_string())]);
+ assert!(values.relative_prefix_path(&invalid).is_err());
+ }
+
#[test]
fn test_name_prefix_pattern() {
let paths = FormatTablePartitionPaths::new(["dt".to_string(),
"hh".to_string()], false);
diff --git a/crates/paimon/src/table/format_table_commit.rs
b/crates/paimon/src/table/format_table_commit.rs
new file mode 100644
index 00000000..ee4a7822
--- /dev/null
+++ b/crates/paimon/src/table/format_table_commit.rs
@@ -0,0 +1,605 @@
+// 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.
+
+//! Publish prepared Format Table files without creating a Paimon snapshot.
+//! Java's `FormatTableCommit` treats published data and catalog partition
+//! registration as separate side effects, with different rollback rules before
+//! and after a partition becomes visible to readers.
+
+use std::collections::{HashMap, HashSet};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use super::commit_message::{CommitMessage, FormatFileCommit};
+use super::format_partition::FormatTablePartitionPaths;
+use super::format_table_scan::list_format_table_files;
+use super::Table;
+use crate::spec::{CoreOptions, Datum, Partition, PartitionStatistics};
+use crate::Result;
+
+pub(crate) struct FormatTableCommit<'a> {
+ table: &'a Table,
+ table_path: String,
+ paths: FormatTablePartitionPaths,
+ default_partition_name: String,
+ dynamic_partition_overwrite: bool,
+}
+
+impl<'a> FormatTableCommit<'a> {
+ pub(crate) fn new(table: &'a Table) -> Self {
+ let options = CoreOptions::new(table.schema().options());
+ Self {
+ table,
+ table_path: options
+ .path()
+ .unwrap_or_else(|| table.location())
+ .trim_end_matches('/')
+ .to_string(),
+ paths: FormatTablePartitionPaths::new(
+ table.schema().partition_keys().iter().cloned(),
+ options.format_table_partition_only_value_in_path(),
+ ),
+ default_partition_name:
options.partition_default_name().to_string(),
+ dynamic_partition_overwrite: table
+ .schema()
+ .options()
+ .get("dynamic-partition-overwrite")
+ .is_none_or(|value| value.eq_ignore_ascii_case("true")),
+ }
+ }
+
+ pub(crate) async fn append(&self, messages: &[CommitMessage]) ->
Result<()> {
+ self.apply(messages, None).await
+ }
+
+ pub(crate) async fn overwrite(
+ &self,
+ messages: &[CommitMessage],
+ static_partition: Option<&HashMap<String, Option<Datum>>>,
+ ) -> Result<()> {
+ self.apply(messages, Some(static_partition)).await
+ }
+
+ pub(crate) async fn abort(&self, messages: &[CommitMessage]) -> Result<()>
{
+ for message in messages {
+ if let Some(file) = &message.format_file {
+ self.table.file_io().delete_file(&file.staged_path).await?;
+ }
+ }
+ Ok(())
+ }
+
+ /// `None` is append; `Some(None)` is an overwrite without static
+ /// partitions; `Some(Some(spec))` selects the leading static prefix.
+ async fn apply(
+ &self,
+ messages: &[CommitMessage],
+ overwrite: Option<Option<&HashMap<String, Option<Datum>>>>,
+ ) -> Result<()> {
+ let result = self.apply_inner(messages, overwrite).await;
+ if result.is_err() {
+ let _ = self.abort(messages).await;
+ }
+ result
+ }
+
+ async fn apply_inner(
+ &self,
+ messages: &[CommitMessage],
+ overwrite: Option<Option<&HashMap<String, Option<Datum>>>>,
+ ) -> Result<()> {
+ self.table.ensure_not_branch_reference_for_write()?;
+ let files = self.validate_messages(messages).await?;
+ let managed = self.table.has_catalog_managed_partitions();
+ let requested = files
+ .iter()
+ .map(|file| file.partition.clone())
+ .collect::<Vec<_>>();
+ let selected = if let Some(static_partition) = overwrite {
+ self.selected_overwrite_partitions(&requested, static_partition)
+ .await?
+ } else {
+ Vec::new()
+ };
+ let static_prefix = match overwrite {
+ Some(Some(spec)) => self.static_prefix(Some(spec))?,
+ _ => HashMap::new(),
+ };
+ if files.iter().any(|file| {
+ !static_prefix
+ .iter()
+ .all(|(key, value)| file.partition.get(key) == Some(value))
+ }) {
+ return Err(crate::Error::DataInvalid {
+ message: "Format Table output is outside the static overwrite
partition".into(),
+ source: None,
+ });
+ }
+
+ // Validate registry paths before deleting or publishing anything. A
+ // partition with a custom location cannot be written through the table
+ // directory; Java rejects it for the same reason.
+ if managed {
+ let mut touched = requested.clone();
+ touched.extend(selected.iter().cloned());
+ self.validate_registered_partitions(&touched, overwrite.is_none())
+ .await?;
+ }
+
+ if overwrite.is_some() {
+ for spec in &selected {
+ let directory = self.partition_directory(spec)?;
+ for status in list_format_table_files(self.table.file_io(),
&directory, 0, None)
+ .await?
+ .into_iter()
+ .filter(|status| !status.is_dir)
+ {
+ self.table.file_io().delete_file(&status.path).await?;
+ }
+ }
+ if !static_prefix.is_empty() {
+ let relative =
self.paths.relative_prefix_path(&static_prefix)?;
+ self.table
+ .file_io()
+ .mkdirs(&format!("{}/{relative}/", self.table_path))
+ .await?;
+ }
+ }
+
+ let mut published: Vec<String> = Vec::new();
+ for file in &files {
+ let result = self.publish(file).await;
+ if let Err(error) = result {
+ // An overwrite has already removed old files, so replacement
+ // files must survive an uncertain partial publish. Append can
+ // safely roll back files this attempt uniquely named.
+ if overwrite.is_none() {
+ for path in
published.iter().chain(std::iter::once(&file.target_path)) {
+ let _ = self.table.file_io().delete_file(path).await;
+ }
+ }
+ self.discard_staging(&files).await;
+ return Err(error);
+ }
+ published.push(file.target_path.clone());
+ }
+ self.discard_staging(&files).await;
+
+ if managed {
+ let stats = self.partition_statistics(&files, &selected,
overwrite.is_some());
+ let specs = stats
+ .iter()
+ .map(|stat| stat.spec.clone())
+ .collect::<Vec<_>>();
+ if !specs.is_empty() {
+ let env = self
+ .table
+ .rest_env()
+ .expect("managed partition REST environment");
+ if overwrite.is_some() {
+ let options = specs
+ .iter()
+ .map(|spec| {
+ Ok(HashMap::from([(
+ "path".to_string(),
+ self.catalog_partition_path(spec)?,
+ )]))
+ })
+ .collect::<Result<Vec<_>>>()?;
+ // Replacement data survives a metadata failure because old
+ // data has already been removed.
+ env.api()
+ .create_partitions_with_options(
+ env.identifier(),
+ specs,
+ true,
+ Some(stats),
+ true,
+ Some(options),
+ )
+ .await?;
+ } else {
+ // Java registers first, then preserves visible files even
+ // if the later additive statistics report fails.
+ if let Err(error) = env
+ .api()
+ .create_partitions(env.identifier(), specs.clone(),
true)
+ .await
+ {
+ for path in &published {
+ let _ =
self.table.file_io().delete_file(path).await;
+ }
+ return Err(error);
+ }
+ if let Err(error) = env
+ .api()
+ .create_partitions_with_statistics(
+ env.identifier(),
+ specs,
+ true,
+ Some(stats),
+ false,
+ )
+ .await
+ {
+ log::warn!(
+ "Committed Format Table data but failed to report
append statistics: {error}"
+ );
+ }
+ }
+ }
+ }
+ Ok(())
+ }
+
+ async fn validate_messages(&self, messages: &[CommitMessage]) ->
Result<Vec<FormatFileCommit>> {
+ let mut files = Vec::with_capacity(messages.len());
+ let mut targets = HashSet::new();
+ for message in messages {
+ let file = message
+ .format_file
+ .as_ref()
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: "Format Table commit requires staged Format Table
file messages"
+ .into(),
+ source: None,
+ })?;
+ let expected_directory =
self.partition_directory(&file.partition)?;
+ let parent = file.target_path.rsplit_once('/').map(|(parent, _)|
parent);
+ if parent != Some(expected_directory.as_str()) {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Format Table target is outside partition: {}",
+ file.target_path
+ ),
+ source: None,
+ });
+ }
+ if !file
+ .staged_path
+ .starts_with(&format!("{}/_temporary/", self.table_path))
+ {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Format Table staging path is outside table: {}",
+ file.staged_path
+ ),
+ source: None,
+ });
+ }
+ if !targets.insert(file.target_path.as_str()) {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Duplicate Format Table target: {}",
file.target_path),
+ source: None,
+ });
+ }
+ if !self.table.file_io().exists(&file.staged_path).await? {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Staged Format Table file is missing:
{}", file.staged_path),
+ source: None,
+ });
+ }
+ if self.table.file_io().exists(&file.target_path).await? {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Format Table target already exists: {}",
file.target_path),
+ source: None,
+ });
+ }
+ files.push(file.clone());
+ }
+ Ok(files)
+ }
+
+ async fn validate_registered_partitions(
+ &self,
+ specs: &[HashMap<String, String>],
+ reject_custom_location: bool,
+ ) -> Result<()> {
+ if specs.is_empty() {
+ return Ok(());
+ }
+ let env = self
+ .table
+ .rest_env()
+ .expect("managed partition REST environment");
+ let registered = env
+ .api()
+ .list_partitions_by_names(env.identifier(), specs.to_vec())
+ .await?;
+ let requested = specs.iter().map(spec_key).collect::<HashSet<_>>();
+ for partition in registered {
+ if !requested.contains(&spec_key(&partition.spec)) {
+ return Err(crate::Error::DataInvalid {
+ message: "Catalog returned an unrequested Format Table
partition".into(),
+ source: None,
+ });
+ }
+ if reject_custom_location
+ && partition
+ .options
+ .as_ref()
+ .is_some_and(|options| options.contains_key("path"))
+ {
+ return Err(crate::Error::Unsupported {
+ message:
+ "Writing a Format Table partition with a custom
location is not supported"
+ .into(),
+ });
+ }
+ }
+ Ok(())
+ }
+
+ fn partition_directory(&self, spec: &HashMap<String, String>) ->
Result<String> {
+ let relative = self.paths.relative_path(spec)?;
+ Ok(if relative.is_empty() {
+ self.table_path.clone()
+ } else {
+ format!("{}/{relative}", self.table_path)
+ })
+ }
+
+ fn catalog_partition_path(&self, spec: &HashMap<String, String>) ->
Result<String> {
+ let path = self.partition_directory(spec)?;
+ if url::Url::parse(&path).is_ok() {
+ return Ok(path);
+ }
+ let absolute = std::path::Path::new(&path);
+ let absolute = if absolute.is_absolute() {
+ absolute.to_path_buf()
+ } else {
+ std::env::current_dir()
+ .map_err(|error| crate::Error::ConfigInvalid {
+ message: format!("Cannot resolve Format Table partition
path: {error}"),
+ })?
+ .join(absolute)
+ };
+ url::Url::from_file_path(&absolute)
+ .map(|url| url.to_string())
+ .map_err(|_| crate::Error::DataInvalid {
+ message: format!("Invalid Format Table partition path:
{path}"),
+ source: None,
+ })
+ }
+
+ async fn publish(&self, file: &FormatFileCommit) -> Result<()> {
+ let directory = self.partition_directory(&file.partition)?;
+ self.table
+ .file_io()
+ .mkdirs(&format!("{directory}/"))
+ .await?;
+ // Java opens each target with overwrite=false. The UUID-based name and
+ // preflight existence check make a collision extremely unlikely; test
+ // once more immediately before moving a file into place.
+ if self.table.file_io().exists(&file.target_path).await? {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Format Table target already exists: {}",
file.target_path),
+ source: None,
+ });
+ }
+ match self
+ .table
+ .file_io()
+ .rename(&file.staged_path, &file.target_path)
+ .await
+ {
+ Ok(()) => Ok(()),
+ Err(crate::Error::IoUnexpected { source, .. })
+ if source.kind() == opendal::ErrorKind::Unsupported =>
+ {
+ let result = self
+ .table
+ .file_io()
+ .copy_file_streaming(&file.staged_path, &file.target_path)
+ .await;
+ if result.is_err() {
+ let _ =
self.table.file_io().delete_file(&file.target_path).await;
+ }
+ result
+ }
+ Err(error) => Err(error),
+ }
+ }
+
+ async fn discard_staging(&self, files: &[FormatFileCommit]) {
+ for file in files {
+ let _ = self.table.file_io().delete_file(&file.staged_path).await;
+ }
+ }
+
+ async fn selected_overwrite_partitions(
+ &self,
+ written: &[HashMap<String, String>],
+ static_partition: Option<&HashMap<String, Option<Datum>>>,
+ ) -> Result<Vec<HashMap<String, String>>> {
+ let keys = self.table.schema().partition_keys();
+ if keys.is_empty() {
+ if static_partition.is_some_and(|spec| !spec.is_empty()) {
+ return Err(crate::Error::DataInvalid {
+ message: "An unpartitioned Format Table cannot have a
static partition".into(),
+ source: None,
+ });
+ }
+ return Ok(vec![HashMap::new()]);
+ }
+ let prefix = self.static_prefix(static_partition)?;
+ if prefix.is_empty() && self.dynamic_partition_overwrite {
+ return Ok(deduplicate_specs(written));
+ }
+ let candidates = if self.table.has_catalog_managed_partitions() {
+ let env = self
+ .table
+ .rest_env()
+ .expect("managed partition REST environment");
+ env.api()
+ .list_partitions(env.identifier())
+ .await?
+ .into_iter()
+ .map(|part| part.spec)
+ .collect::<Vec<_>>()
+ } else {
+ self.paths
+ .discover(
+ self.table.file_io(),
+ &self.table_path,
+ &self.default_partition_name,
+ )
+ .await?
+ };
+ let mut selected = candidates
+ .into_iter()
+ .filter(|spec| {
+ prefix
+ .iter()
+ .all(|(key, value)| spec.get(key) == Some(value))
+ })
+ .collect::<Vec<_>>();
+ selected.extend(
+ written
+ .iter()
+ .filter(|spec| {
+ prefix
+ .iter()
+ .all(|(key, value)| spec.get(key) == Some(value))
+ })
+ .cloned(),
+ );
+ if prefix.len() == keys.len() {
+ selected.push(prefix);
+ }
+ Ok(deduplicate_specs(&selected))
+ }
+
+ fn static_prefix(
+ &self,
+ static_partition: Option<&HashMap<String, Option<Datum>>>,
+ ) -> Result<HashMap<String, String>> {
+ let Some(static_partition) = static_partition else {
+ return Ok(HashMap::new());
+ };
+ let keys = self.table.schema().partition_keys();
+ let mut prefix = HashMap::new();
+ let mut missing = false;
+ for key in keys {
+ match static_partition.get(key) {
+ Some(_) if missing => {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Static partition '{key}' lacks its
leading partition"),
+ source: None,
+ });
+ }
+ Some(value) => {
+ let field = self
+ .table
+ .schema()
+ .fields()
+ .iter()
+ .find(|field| field.name() == key)
+ .expect("partition field");
+ let text = match value {
+ None => self.default_partition_name.clone(),
+ Some(datum) =>
super::format_partition::format_partition_value(
+ datum,
+ field.data_type(),
+ &self.default_partition_name,
+ CoreOptions::new(self.table.schema().options())
+ .legacy_partition_name(),
+ )
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!(
+ "Static partition '{key}' has a value
incompatible with its data type"
+ ),
+ source: None,
+ })?,
+ };
+ prefix.insert(key.clone(), text);
+ }
+ None => missing = true,
+ }
+ }
+ if static_partition.keys().any(|key| !keys.contains(key)) {
+ return Err(crate::Error::DataInvalid {
+ message: "Unknown static Format Table partition column".into(),
+ source: None,
+ });
+ }
+ Ok(prefix)
+ }
+
+ fn partition_statistics(
+ &self,
+ files: &[FormatFileCommit],
+ selected: &[HashMap<String, String>],
+ overwrite: bool,
+ ) -> Vec<PartitionStatistics> {
+ let now = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .unwrap_or_default()
+ .as_millis() as i64;
+ let mut by_spec: HashMap<Vec<(String, String)>, PartitionStatistics> =
HashMap::new();
+ if overwrite {
+ for spec in selected {
+ let key = spec_key(spec);
+ by_spec.insert(
+ key,
+ PartitionStatistics {
+ spec: spec.clone(),
+ record_count: 0,
+ file_size_in_bytes: 0,
+ file_count: 0,
+ last_file_creation_time: now,
+ total_buckets: Partition::UNKNOWN_TOTAL_BUCKETS,
+ },
+ );
+ }
+ }
+ for file in files {
+ let stat =
+ by_spec
+ .entry(spec_key(&file.partition))
+ .or_insert_with(|| PartitionStatistics {
+ spec: file.partition.clone(),
+ record_count: 0,
+ file_size_in_bytes: 0,
+ file_count: 0,
+ last_file_creation_time: now,
+ total_buckets: Partition::UNKNOWN_TOTAL_BUCKETS,
+ });
+ stat.record_count += file.record_count;
+ stat.file_size_in_bytes += file.file_size;
+ stat.file_count += 1;
+ }
+ by_spec.into_values().collect()
+ }
+}
+
+fn deduplicate_specs(specs: &[HashMap<String, String>]) -> Vec<HashMap<String,
String>> {
+ let mut seen = HashSet::new();
+ specs
+ .iter()
+ .filter(|spec| seen.insert(spec_key(spec)))
+ .cloned()
+ .collect()
+}
+
+fn spec_key(spec: &HashMap<String, String>) -> Vec<(String, String)> {
+ let mut key = spec
+ .iter()
+ .map(|(key, value)| (key.clone(), value.clone()))
+ .collect::<Vec<_>>();
+ key.sort();
+ key
+}
diff --git a/crates/paimon/src/table/format_table_scan.rs
b/crates/paimon/src/table/format_table_scan.rs
index 7297b1c5..227fbc5b 100644
--- a/crates/paimon/src/table/format_table_scan.rs
+++ b/crates/paimon/src/table/format_table_scan.rs
@@ -295,12 +295,16 @@ impl<'a> FormatTableScan<'a> {
if !self.partition_matches(&row)? {
continue;
}
- // The Rust reader cannot resolve a partition's own location yet,
and reading the
- // default directory in its place would return whatever happens to
be there.
+ // An overwrite explicitly rebinds a formerly custom partition to
+ // its default directory. That path option is safe to read here;
+ // other custom locations still need their own resolver.
if partition
.options
.as_ref()
- .is_some_and(|options| options.contains_key(PATH_OPTION))
+ .and_then(|options| options.get(PATH_OPTION))
+ .is_some_and(|location| {
+ location.trim_end_matches('/') !=
path.trim_end_matches('/')
+ })
{
return Err(crate::Error::Unsupported {
message: format!(
diff --git a/crates/paimon/src/table/format_table_write_tests.rs
b/crates/paimon/src/table/format_table_write_tests.rs
new file mode 100644
index 00000000..9b4bd883
--- /dev/null
+++ b/crates/paimon/src/table/format_table_write_tests.rs
@@ -0,0 +1,717 @@
+// 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.
+
+//! Format Table write behavior is tested through the same public builder,
+//! scan and read calls used by native callers.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use arrow_array::{Int32Array, RecordBatch, StringArray};
+use arrow_schema::{DataType as ArrowType, Field, Schema as ArrowSchema};
+use futures::TryStreamExt;
+
+use super::Table;
+use crate::catalog::Identifier;
+use crate::io::{FileIO, FileIOBuilder};
+use crate::spec::{DataType, IntType, Schema, TableSchema, VarCharType};
+
+fn table(io: FileIO, location: &str, partitioned: bool, options: &[(&str,
&str)]) -> Table {
+ let mut builder = Schema::builder();
+ if partitioned {
+ builder = builder.column("dt",
DataType::VarChar(VarCharType::string_type()));
+ }
+ builder = builder
+ .column("id", DataType::Int(IntType::new()))
+ .option("type", "format-table")
+ .option("file.format", "parquet");
+ if partitioned {
+ builder = builder.partition_keys(["dt"]);
+ }
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ Table::new(
+ io,
+ Identifier::new("default", "format_write"),
+ location.to_string(),
+ TableSchema::new(0, &builder.build().unwrap()),
+ None,
+ )
+}
+
+fn memory_table(name: &str, partitioned: bool, options: &[(&str, &str)]) ->
Table {
+ table(
+ FileIOBuilder::new("memory").build().unwrap(),
+ &format!("memory:/{name}"),
+ partitioned,
+ options,
+ )
+}
+
+fn two_partition_table(name: &str) -> Table {
+ let schema = Schema::builder()
+ .column("dt", DataType::VarChar(VarCharType::string_type()))
+ .column("hour", DataType::VarChar(VarCharType::string_type()))
+ .column("id", DataType::Int(IntType::new()))
+ .partition_keys(["dt", "hour"])
+ .option("type", "format-table")
+ .option("file.format", "parquet")
+ .build()
+ .unwrap();
+ Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "format_two_partition"),
+ format!("memory:/{name}"),
+ TableSchema::new(0, &schema),
+ None,
+ )
+}
+
+fn two_partition_batch(rows: &[(&str, &str, i32)]) -> RecordBatch {
+ RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ Field::new("dt", ArrowType::Utf8, true),
+ Field::new("hour", ArrowType::Utf8, true),
+ Field::new("id", ArrowType::Int32, true),
+ ])),
+ vec![
+ Arc::new(StringArray::from(
+ rows.iter().map(|row| row.0).collect::<Vec<_>>(),
+ )),
+ Arc::new(StringArray::from(
+ rows.iter().map(|row| row.1).collect::<Vec<_>>(),
+ )),
+ Arc::new(Int32Array::from(
+ rows.iter().map(|row| row.2).collect::<Vec<_>>(),
+ )),
+ ],
+ )
+ .unwrap()
+}
+
+fn batch(rows: &[(&str, i32)]) -> RecordBatch {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ Field::new("dt", ArrowType::Utf8, true),
+ Field::new("id", ArrowType::Int32, true),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(StringArray::from(
+ rows.iter().map(|row| row.0).collect::<Vec<_>>(),
+ )),
+ Arc::new(Int32Array::from(
+ rows.iter().map(|row| row.1).collect::<Vec<_>>(),
+ )),
+ ],
+ )
+ .unwrap()
+}
+
+fn unpartitioned_batch(ids: &[i32]) -> RecordBatch {
+ RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![Field::new(
+ "id",
+ ArrowType::Int32,
+ true,
+ )])),
+ vec![Arc::new(Int32Array::from(ids.to_vec()))],
+ )
+ .unwrap()
+}
+
+async fn ids(table: &Table) -> Vec<i32> {
+ let plan = table.new_read_builder().new_scan().plan().await.unwrap();
+ let read = table.new_read_builder().new_read().unwrap();
+ let batches: Vec<RecordBatch> = read
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut ids = Vec::new();
+ for batch in batches {
+ let index = batch.schema().index_of("id").unwrap();
+ let column = batch
+ .column(index)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ ids.extend(column.iter().flatten());
+ }
+ ids.sort_unstable();
+ ids
+}
+
+async fn visible_files(table: &Table, partition: &str) -> Vec<String> {
+ let path = if partition.is_empty() {
+ table.location().to_string()
+ } else {
+ format!("{}/{partition}", table.location())
+ };
+ let mut files =
+ super::format_table_scan::list_format_table_files(table.file_io(),
&path, 0, None)
+ .await
+ .unwrap()
+ .into_iter()
+ .filter(|status| !status.is_dir)
+ .map(|status| status.path)
+ .collect::<Vec<_>>();
+ files.sort();
+ files
+}
+
+async fn append(table: &Table, batch: &RecordBatch) {
+ let builder = table.new_write_builder();
+ let mut write = builder.new_write().unwrap();
+ write.write_arrow_batch(batch).await.unwrap();
+ let messages = write.prepare_commit().await.unwrap();
+ builder.new_commit().commit(messages).await.unwrap();
+}
+
+#[tokio::test]
+async fn append_partitioned_table_writes_data_without_partition_column() {
+ let table = memory_table("format_append_partitioned", true, &[]);
+ let builder = table.new_write_builder();
+ let mut write = builder.new_write().unwrap();
+ write
+ .write_arrow_batch(&batch(&[("a", 1), ("b", 2), ("a", 3)]))
+ .await
+ .unwrap();
+ let messages = write.prepare_commit().await.unwrap();
+ assert_eq!(messages.len(), 2);
+ for message in &messages {
+ let file = message.format_file.as_ref().unwrap();
+ assert!(file.staged_path.contains("/_temporary/"));
+ assert!(!table.file_io().exists(&file.target_path).await.unwrap());
+ assert_eq!(file.partition.len(), 1);
+ }
+ builder.new_commit().commit(messages).await.unwrap();
+ assert_eq!(ids(&table).await, [1, 2, 3]);
+ assert_eq!(visible_files(&table, "dt=a").await.len(), 1);
+ assert_eq!(visible_files(&table, "dt=b").await.len(), 1);
+}
+
+#[tokio::test]
+async fn append_unpartitioned_table_has_no_snapshot() {
+ let table = memory_table("format_append_plain", false, &[]);
+ append(&table, &unpartitioned_batch(&[4, 5])).await;
+ append(&table, &unpartitioned_batch(&[6])).await;
+ assert_eq!(ids(&table).await, [4, 5, 6]);
+ assert_eq!(visible_files(&table, "").await.len(), 2);
+ assert!(table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .is_none());
+}
+
+#[tokio::test]
+async fn format_table_ignores_paimon_bucket_configuration() {
+ let table = memory_table("format_bucket_ignored", true, &[("bucket",
"2")]);
+ append(&table, &batch(&[("a", 1)])).await;
+ assert_eq!(ids(&table).await, [1]);
+}
+
+#[test]
+fn declared_column_default_is_rejected_before_opening_a_writer() {
+ let table = memory_table("format_column_default", true, &[]);
+ let mut schema = serde_json::to_value(table.schema()).unwrap();
+ schema["fields"][0]["defaultValue"] = serde_json::json!("'fallback'");
+ let schema = serde_json::from_value(schema).unwrap();
+ let table = Table::new(
+ table.file_io().clone(),
+ table.identifier().clone(),
+ table.location().to_string(),
+ schema,
+ None,
+ );
+ let error = table.new_write_builder().new_write().err().unwrap();
+ assert!(error.to_string().contains("column default"));
+}
+
+#[tokio::test]
+async fn file_rolls_at_target_row_count_even_inside_one_batch() {
+ let table = memory_table("format_roll_rows", true,
&[("target-file-row-num", "2")]);
+ append(
+ &table,
+ &batch(&[("a", 1), ("a", 2), ("a", 3), ("a", 4), ("a", 5)]),
+ )
+ .await;
+ assert_eq!(ids(&table).await, [1, 2, 3, 4, 5]);
+ assert_eq!(visible_files(&table, "dt=a").await.len(), 3);
+}
+
+#[tokio::test]
+async fn overwrite_replaces_only_written_partitions_by_default() {
+ let table = memory_table("format_overwrite_dynamic", true, &[]);
+ append(&table, &batch(&[("a", 1), ("b", 2)])).await;
+ let builder = table.new_write_builder().with_overwrite();
+ let mut write = builder.new_write().unwrap();
+ write.write_arrow_batch(&batch(&[("a", 7)])).await.unwrap();
+ let messages = write.prepare_commit().await.unwrap();
+ builder
+ .new_commit()
+ .overwrite(messages, None)
+ .await
+ .unwrap();
+ assert_eq!(ids(&table).await, [2, 7]);
+ assert_eq!(visible_files(&table, "dt=a").await.len(), 1);
+}
+
+#[tokio::test]
+async fn static_partition_overwrite_without_output_clears_partition() {
+ let table = memory_table("format_overwrite_empty", true, &[]);
+ append(&table, &batch(&[("a", 1), ("b", 2)])).await;
+ let builder = table.new_write_builder().with_overwrite();
+ builder
+ .new_commit()
+ .overwrite(
+ Vec::new(),
+ Some(HashMap::from([(
+ "dt".to_string(),
+ Some(crate::spec::Datum::String("a".into())),
+ )])),
+ )
+ .await
+ .unwrap();
+ assert_eq!(ids(&table).await, [2]);
+ assert!(visible_files(&table, "dt=a").await.is_empty());
+}
+
+#[tokio::test]
+async fn prepared_files_are_hidden_until_commit_and_abort_discards_them() {
+ let table = memory_table("format_abort_prepared", true, &[]);
+ let builder = table.new_write_builder();
+ let mut write = builder.new_write().unwrap();
+ write
+ .write_arrow_batch(&batch(&[("a", 1), ("a", 2)]))
+ .await
+ .unwrap();
+ let messages = write.prepare_commit().await.unwrap();
+ assert_eq!(messages.len(), 1);
+ let staged = &messages[0].format_file.as_ref().unwrap().staged_path;
+ assert!(table.file_io().exists(staged).await.unwrap());
+ assert!(visible_files(&table, "dt=a").await.is_empty());
+ assert!(table
+ .new_read_builder()
+ .new_scan()
+ .plan()
+ .await
+ .unwrap()
+ .splits()
+ .is_empty());
+ builder.new_commit().abort(&messages).await.unwrap();
+ assert!(!table.file_io().exists(staged).await.unwrap());
+ assert!(visible_files(&table, "dt=a").await.is_empty());
+}
+
+#[tokio::test]
+async fn closing_a_writer_discards_pending_data_without_publishing() {
+ let table = memory_table("format_close_pending", true, &[]);
+ let builder = table.new_write_builder();
+ let mut write = builder.new_write().unwrap();
+ write.write_arrow_batch(&batch(&[("a", 1)])).await.unwrap();
+ write.close().await;
+ assert!(visible_files(&table, "dt=a").await.is_empty());
+ assert!(table
+ .new_read_builder()
+ .new_scan()
+ .plan()
+ .await
+ .unwrap()
+ .splits()
+ .is_empty());
+}
+
+#[tokio::test]
+async fn rejected_batch_does_not_stage_or_publish_a_file() {
+ let table = memory_table("format_rejected_batch", true, &[]);
+ let builder = table.new_write_builder();
+ let mut write = builder.new_write().unwrap();
+ let wrong = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ Field::new("dt", ArrowType::Utf8, true),
+ Field::new("id", ArrowType::Utf8, true),
+ ])),
+ vec![
+ Arc::new(StringArray::from(vec!["a"])),
+ Arc::new(StringArray::from(vec!["invalid"])),
+ ],
+ )
+ .unwrap();
+ let error = write.write_arrow_batch(&wrong).await.err().unwrap();
+ assert!(error.to_string().contains("expects"));
+ assert!(write.prepare_commit().await.is_err());
+ assert!(visible_files(&table, "dt=a").await.is_empty());
+}
+
+#[tokio::test]
+async fn one_writer_can_prepare_more_than_one_append() {
+ let table = memory_table("format_writer_reuse", true, &[]);
+ let builder = table.new_write_builder();
+ let mut write = builder.new_write().unwrap();
+ write.write_arrow_batch(&batch(&[("a", 1)])).await.unwrap();
+ builder
+ .new_commit()
+ .commit(write.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ write
+ .write_arrow_batch(&batch(&[("a", 2), ("b", 3)]))
+ .await
+ .unwrap();
+ builder
+ .new_commit()
+ .commit(write.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ assert_eq!(ids(&table).await, [1, 2, 3]);
+ assert_eq!(visible_files(&table, "dt=a").await.len(), 2);
+ assert_eq!(visible_files(&table, "dt=b").await.len(), 1);
+}
+
+#[tokio::test]
+async fn value_only_partition_path_is_written_and_read() {
+ let table = memory_table(
+ "format_value_only_write",
+ true,
+ &[("format-table.partition-path-only-value", "true")],
+ );
+ append(&table, &batch(&[("a", 1), ("b", 2)])).await;
+ assert_eq!(ids(&table).await, [1, 2]);
+ assert_eq!(visible_files(&table, "a").await.len(), 1);
+ assert_eq!(visible_files(&table, "b").await.len(), 1);
+ assert!(visible_files(&table, "dt=a").await.is_empty());
+}
+
+#[tokio::test]
+async fn overwrite_without_dynamic_partition_mode_replaces_all_partitions() {
+ let table = memory_table(
+ "format_overwrite_all",
+ true,
+ &[("dynamic-partition-overwrite", "false")],
+ );
+ append(&table, &batch(&[("a", 1), ("b", 2)])).await;
+ let builder = table.new_write_builder().with_overwrite();
+ let mut write = builder.new_write().unwrap();
+ write.write_arrow_batch(&batch(&[("a", 7)])).await.unwrap();
+ builder
+ .new_commit()
+ .overwrite(write.prepare_commit().await.unwrap(), None)
+ .await
+ .unwrap();
+ assert_eq!(ids(&table).await, [7]);
+ assert!(visible_files(&table, "dt=b").await.is_empty());
+}
+
+#[tokio::test]
+async fn target_collision_is_rejected_before_any_overwrite_cleanup() {
+ let table = memory_table("format_target_collision", true, &[]);
+ append(&table, &batch(&[("a", 1)])).await;
+ let builder = table.new_write_builder().with_overwrite();
+ let mut write = builder.new_write().unwrap();
+ write.write_arrow_batch(&batch(&[("a", 2)])).await.unwrap();
+ let messages = write.prepare_commit().await.unwrap();
+ let file = messages[0].format_file.as_ref().unwrap();
+ table
+ .file_io()
+ .copy_file_streaming(&file.staged_path, &file.target_path)
+ .await
+ .unwrap();
+ let old_files = visible_files(&table, "dt=a").await;
+ let error = builder
+ .new_commit()
+ .overwrite(messages.clone(), None)
+ .await
+ .err()
+ .unwrap();
+ assert!(error.to_string().contains("target already exists"));
+ assert_eq!(visible_files(&table, "dt=a").await, old_files);
+ builder.new_commit().abort(&messages).await.unwrap();
+ assert_eq!(visible_files(&table, "dt=a").await, old_files);
+}
+
+#[tokio::test]
+async fn missing_staged_file_cannot_erase_an_overwritten_partition() {
+ let table = memory_table("format_missing_stage", true, &[]);
+ append(&table, &batch(&[("a", 1)])).await;
+ let builder = table.new_write_builder().with_overwrite();
+ let mut write = builder.new_write().unwrap();
+ write.write_arrow_batch(&batch(&[("a", 2)])).await.unwrap();
+ let messages = write.prepare_commit().await.unwrap();
+ let staged = &messages[0].format_file.as_ref().unwrap().staged_path;
+ table.file_io().delete_file(staged).await.unwrap();
+ let error = builder
+ .new_commit()
+ .overwrite(messages, None)
+ .await
+ .err()
+ .unwrap();
+ assert!(error.to_string().contains("is missing"));
+ assert_eq!(ids(&table).await, [1]);
+}
+
+#[tokio::test]
+async fn format_file_message_refuses_snapshot_wire_serialization() {
+ let table = memory_table("format_message_serialize", false, &[]);
+ let builder = table.new_write_builder();
+ let mut write = builder.new_write().unwrap();
+ write
+ .write_arrow_batch(&unpartitioned_batch(&[1]))
+ .await
+ .unwrap();
+ let messages = write.prepare_commit().await.unwrap();
+ let error = messages[0].serialize().err().unwrap();
+ assert!(error.to_string().contains("different Java serializer"));
+ builder.new_commit().abort(&messages).await.unwrap();
+}
+
+#[tokio::test]
+async fn data_file_name_honors_prefix_and_compression_suffix() {
+ let table = memory_table(
+ "format_file_name",
+ true,
+ &[
+ ("data-file.prefix", "event-"),
+ ("file.suffix.include.compression", "true"),
+ ],
+ );
+ append(&table, &batch(&[("a", 1)])).await;
+ let files = visible_files(&table, "dt=a").await;
+ assert_eq!(files.len(), 1);
+ let name = files[0].rsplit('/').next().unwrap();
+ assert!(name.starts_with("event-"), "{name}");
+ assert!(name.ends_with(".snappy.parquet"), "{name}");
+ assert_eq!(ids(&table).await, [1]);
+}
+
+#[tokio::test]
+async fn static_prefix_overwrite_removes_only_matching_descendants() {
+ let table = two_partition_table("format_static_prefix");
+ append(
+ &table,
+ &two_partition_batch(&[("a", "00", 1), ("a", "01", 2), ("b", "00",
3)]),
+ )
+ .await;
+ let builder = table.new_write_builder().with_overwrite();
+ let mut write = builder.new_write().unwrap();
+ write
+ .write_arrow_batch(&two_partition_batch(&[("a", "01", 7)]))
+ .await
+ .unwrap();
+ builder
+ .new_commit()
+ .overwrite(
+ write.prepare_commit().await.unwrap(),
+ Some(HashMap::from([(
+ "dt".to_string(),
+ Some(crate::spec::Datum::String("a".into())),
+ )])),
+ )
+ .await
+ .unwrap();
+ assert_eq!(ids(&table).await, [3, 7]);
+ assert!(visible_files(&table, "dt=a/hour=00").await.is_empty());
+ assert_eq!(visible_files(&table, "dt=a/hour=01").await.len(), 1);
+ assert_eq!(visible_files(&table, "dt=b/hour=00").await.len(), 1);
+}
+
+#[tokio::test]
+async fn empty_static_overwrite_creates_the_selected_partition_directory() {
+ let table = two_partition_table("format_static_empty");
+ let prefix = HashMap::from([
+ (
+ "dt".to_string(),
+ Some(crate::spec::Datum::String("new".into())),
+ ),
+ (
+ "hour".to_string(),
+ Some(crate::spec::Datum::String("09".into())),
+ ),
+ ]);
+ table
+ .new_write_builder()
+ .with_overwrite()
+ .new_commit()
+ .overwrite(Vec::new(), Some(prefix))
+ .await
+ .unwrap();
+ assert!(table
+ .file_io()
+ .exists_dir("memory:/format_static_empty/dt=new/hour=09")
+ .await
+ .unwrap());
+ assert!(ids(&table).await.is_empty());
+}
+
+#[tokio::test]
+async fn output_outside_static_prefix_is_rejected_before_deleting_old_files() {
+ let table = two_partition_table("format_static_mismatch");
+ append(&table, &two_partition_batch(&[("a", "00", 1)])).await;
+ let builder = table.new_write_builder().with_overwrite();
+ let mut write = builder.new_write().unwrap();
+ write
+ .write_arrow_batch(&two_partition_batch(&[("b", "00", 2)]))
+ .await
+ .unwrap();
+ let messages = write.prepare_commit().await.unwrap();
+ let error = builder
+ .new_commit()
+ .overwrite(
+ messages.clone(),
+ Some(HashMap::from([(
+ "dt".to_string(),
+ Some(crate::spec::Datum::String("a".into())),
+ )])),
+ )
+ .await
+ .err()
+ .unwrap();
+ assert!(error.to_string().contains("outside the static overwrite"));
+ assert_eq!(ids(&table).await, [1]);
+ builder.new_commit().abort(&messages).await.unwrap();
+}
+
+#[tokio::test]
+async fn static_partition_must_be_a_leading_prefix() {
+ let table = two_partition_table("format_static_gap");
+ let error = table
+ .new_write_builder()
+ .with_overwrite()
+ .new_commit()
+ .overwrite(
+ Vec::new(),
+ Some(HashMap::from([(
+ "hour".to_string(),
+ Some(crate::spec::Datum::String("00".into())),
+ )])),
+ )
+ .await
+ .err()
+ .unwrap();
+ assert!(error.to_string().contains("lacks its leading partition"));
+ assert!(ids(&table).await.is_empty());
+}
+
+#[tokio::test]
+async fn physical_parquet_file_omits_partition_columns() {
+ let table = two_partition_table("format_physical_columns");
+ append(&table, &two_partition_batch(&[("a", "00", 7)])).await;
+ let files = visible_files(&table, "dt=a/hour=00").await;
+ assert_eq!(files.len(), 1);
+ let bytes = table
+ .file_io()
+ .new_input(&files[0])
+ .unwrap()
+ .read()
+ .await
+ .unwrap();
+ let reader =
+
parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder::try_new(bytes).unwrap();
+ let fields = reader.schema().fields();
+ assert_eq!(fields.len(), 1);
+ assert_eq!(fields[0].name(), "id");
+ assert_eq!(ids(&table).await, [7]);
+}
+
+#[tokio::test]
+async fn row_format_round_trips_through_native_read() {
+ for format in ["row"] {
+ let name = format!("format_write_{format}");
+ let table = memory_table(&name, true, &[("file.format", format)]);
+ append(&table, &batch(&[("a", 1), ("b", 2), ("a", 3)])).await;
+ assert_eq!(ids(&table).await, [1, 2, 3], "format={format}");
+ assert_eq!(
+ visible_files(&table, "dt=a")
+ .await
+ .iter()
+ .filter(|path| path.ends_with(&format!(".{format}")))
+ .count(),
+ 1,
+ "format={format}"
+ );
+ }
+}
+
+#[tokio::test]
+async fn null_partition_uses_the_configured_default_directory() {
+ let table = memory_table(
+ "format_null_partition",
+ true,
+ &[("partition.default-name", "_empty_")],
+ );
+ let batch = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ Field::new("dt", ArrowType::Utf8, true),
+ Field::new("id", ArrowType::Int32, true),
+ ])),
+ vec![
+ Arc::new(StringArray::from(vec![None::<&str>, Some("a")])),
+ Arc::new(Int32Array::from(vec![1, 2])),
+ ],
+ )
+ .unwrap();
+ append(&table, &batch).await;
+ assert_eq!(ids(&table).await, [1, 2]);
+ assert_eq!(visible_files(&table, "dt=_empty_").await.len(), 1);
+ assert_eq!(visible_files(&table, "dt=a").await.len(), 1);
+}
+
+#[test]
+fn invalid_target_row_count_is_rejected_before_a_writer_is_opened() {
+ let table = memory_table(
+ "format_bad_row_target",
+ false,
+ &[("target-file-row-num", "0")],
+ );
+ let error = table.new_write_builder().new_write().err().unwrap();
+ assert!(error
+ .to_string()
+ .contains("target-file-row-num must be positive"));
+}
+
+#[test]
+fn unsupported_format_is_rejected_before_a_writer_is_opened() {
+ let table = memory_table(
+ "format_unsupported_write",
+ false,
+ &[("file.format", "json")],
+ );
+ let error = table.new_write_builder().new_write().err().unwrap();
+ assert!(error.to_string().contains("not supported"));
+}
+
+#[test]
+fn readable_but_unwritable_formats_are_rejected_before_staging() {
+ for format in ["avro", "orc", "mosaic"] {
+ let table = memory_table(
+ &format!("format_no_{format}_writer"),
+ false,
+ &[("file.format", format)],
+ );
+ let error = table.new_write_builder().new_write().err().unwrap();
+ assert!(
+ error
+ .to_string()
+ .contains("can be read but cannot be written"),
+ "format={format}, error={error}"
+ );
+ }
+}
diff --git a/crates/paimon/src/table/format_table_writer.rs
b/crates/paimon/src/table/format_table_writer.rs
new file mode 100644
index 00000000..8490a6f9
--- /dev/null
+++ b/crates/paimon/src/table/format_table_writer.rs
@@ -0,0 +1,475 @@
+// 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.
+
+//! File writer for a Format Table. Java's `FormatTableWrite` puts partition
columns in
+//! the directory name and writes only the remaining columns to each data file.
+//! Files stay below a hidden staging directory until `FormatTableCommit`
publishes them.
+
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+
+use arrow_array::RecordBatch;
+use arrow_schema::{Field, Schema as ArrowSchema};
+
+use super::commit_message::{CommitMessage, FormatFileCommit};
+use super::format_partition::FormatTablePartitionPaths;
+use super::format_table_scan::supported_format_table_extension;
+use super::table_write::take_rows;
+use super::Table;
+use crate::arrow::build_target_arrow_schema;
+use crate::arrow::format::{create_format_writer, with_write_resources,
FormatFileWriter};
+use crate::resource::ResourceContext;
+use crate::spec::{BinaryRow, CoreOptions, DataField};
+use crate::Result;
+
+struct OpenFile {
+ writer: Box<dyn FormatFileWriter>,
+ staged_path: String,
+ target_path: String,
+ record_count: i64,
+}
+
+struct PartitionWriter {
+ spec: HashMap<String, String>,
+ directory: String,
+ open: Option<OpenFile>,
+}
+
+/// Mirrors the file side of Java `FormatTableWrite` and
+/// `FormatTableRollingFileWriter`. The table schema is the input schema; a
+/// file's schema omits partition columns. A failed write owns and deletes its
+/// staged files, while prepared files are transferred to the committer.
+pub(crate) struct FormatTableWriter {
+ table: Table,
+ full_schema: Arc<ArrowSchema>,
+ file_schema: Arc<ArrowSchema>,
+ file_fields: Vec<DataField>,
+ partition_indices: Vec<usize>,
+ data_indices: Vec<usize>,
+ partition_paths: FormatTablePartitionPaths,
+ table_path: String,
+ data_file_prefix: String,
+ extension: String,
+ compression: String,
+ zstd_level: i32,
+ target_file_size: i64,
+ target_file_rows: i64,
+ resources: Option<ResourceContext>,
+ staging_root: String,
+ writers: HashMap<Vec<u8>, PartitionWriter>,
+ prepared: Vec<CommitMessage>,
+ owned_staging: HashSet<String>,
+ failed: bool,
+}
+
+impl FormatTableWriter {
+ pub(crate) fn set_resources(&mut self, resources: ResourceContext) {
+ self.resources = Some(resources);
+ }
+
+ pub(crate) fn new(table: &Table, resources: Option<ResourceContext>) ->
Result<Self> {
+ table.ensure_not_branch_reference_for_write()?;
+ let schema = table.schema();
+ if let Some(field) = schema
+ .fields()
+ .iter()
+ .find(|field| field.default_value().is_some())
+ {
+ return Err(crate::Error::Unsupported {
+ message: format!(
+ "Format Table column default for '{}' is not supported by
the Rust writer",
+ field.name()
+ ),
+ });
+ }
+ let options = CoreOptions::new(schema.options());
+ let format = options.file_format();
+ let extension = supported_format_table_extension(&format)?.to_string();
+ // Readers cover more external file types than native writers. Reject
+ // unsupported formats before staging any files, rather than failing
+ // after the first RecordBatch has been routed.
+ match format.as_str() {
+ "parquet" | "row" => {}
+ #[cfg(feature = "vortex")]
+ "vortex" => {}
+ _ => {
+ return Err(crate::Error::Unsupported {
+ message: format!(
+ "Format Table file.format '{format}' can be read but
cannot be written by the Rust client"
+ ),
+ });
+ }
+ }
+ let full_schema = build_target_arrow_schema(schema.fields())?;
+ let partition_indices = schema
+ .partition_keys()
+ .iter()
+ .map(|key| {
+ schema
+ .fields()
+ .iter()
+ .position(|field| field.name() == key)
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!("Unknown Format Table partition
column '{key}'"),
+ source: None,
+ })
+ })
+ .collect::<Result<Vec<_>>>()?;
+ let data_indices = (0..schema.fields().len())
+ .filter(|index| !partition_indices.contains(index))
+ .collect::<Vec<_>>();
+ let file_fields = data_indices
+ .iter()
+ .map(|&index| schema.fields()[index].clone())
+ .collect::<Vec<_>>();
+ let file_schema = Arc::new(ArrowSchema::new(
+ data_indices
+ .iter()
+ .map(|&index| Arc::new(Field::clone(full_schema.field(index))))
+ .collect::<Vec<_>>(),
+ ));
+ let target_file_rows = schema
+ .options()
+ .get("target-file-row-num")
+ .and_then(|value| value.parse::<i64>().ok())
+ .unwrap_or(i64::MAX);
+ if target_file_rows <= 0 {
+ return Err(crate::Error::ConfigInvalid {
+ message: "target-file-row-num must be positive".into(),
+ });
+ }
+ let compression = format_table_compression(schema.options(), &format);
+ let extension = if schema
+ .options()
+ .get("file.suffix.include.compression")
+ .is_some_and(|value| value.eq_ignore_ascii_case("true"))
+ && !matches!(compression.as_str(), "" | "none")
+ {
+ format!(".{compression}{extension}")
+ } else {
+ extension
+ };
+ let table_path = options
+ .path()
+ .unwrap_or_else(|| table.location())
+ .trim_end_matches('/')
+ .to_string();
+ let staging_root = format!("{table_path}/_temporary/{}",
uuid::Uuid::new_v4());
+ Ok(Self {
+ table: table.clone(),
+ full_schema,
+ file_schema,
+ file_fields,
+ partition_indices,
+ data_indices,
+ partition_paths: FormatTablePartitionPaths::new(
+ schema.partition_keys().iter().cloned(),
+ options.format_table_partition_only_value_in_path(),
+ ),
+ table_path,
+ data_file_prefix: options.data_file_prefix().to_string(),
+ extension,
+ compression,
+ zstd_level: options.file_compression_zstd_level(),
+ target_file_size: options.target_file_size(),
+ target_file_rows,
+ resources,
+ staging_root,
+ writers: HashMap::new(),
+ prepared: Vec::new(),
+ owned_staging: HashSet::new(),
+ failed: false,
+ })
+ }
+
+ pub(crate) async fn write(&mut self, batch: &RecordBatch) -> Result<()> {
+ if self.failed {
+ return Err(crate::Error::DataInvalid {
+ message: "Format Table writer failed; create a new
writer".into(),
+ source: None,
+ });
+ }
+ let result = self.write_inner(batch).await;
+ if result.is_err() {
+ self.failed = true;
+ self.close().await;
+ }
+ result
+ }
+
+ async fn write_inner(&mut self, batch: &RecordBatch) -> Result<()> {
+ self.check_schema(batch)?;
+ if batch.num_rows() == 0 {
+ return Ok(());
+ }
+
+ // Validate all non-null fields before opening a file. Java validates
the
+ // input row before default-value replacement and partition extraction.
+ for (field_index, field) in
self.full_schema.fields().iter().enumerate() {
+ if !field.is_nullable() && batch.column(field_index).null_count()
> 0 {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Cannot write null to non-null
column({})", field.name()),
+ source: None,
+ });
+ }
+ }
+
+ let mut groups: HashMap<Vec<u8>, Vec<usize>> = HashMap::new();
+ let mut specs = HashMap::new();
+ let partition_fields = self.table.schema().partition_fields();
+ let options = CoreOptions::new(self.table.schema().options());
+ let computer = crate::spec::PartitionComputer::new(
+ self.table.schema().partition_keys(),
+ self.table.schema().fields(),
+ options.partition_default_name(),
+ options.legacy_partition_name(),
+ )?;
+ for row_index in 0..batch.num_rows() {
+ let partition = BinaryRow::from_arrow(
+ batch,
+ row_index,
+ &self.partition_indices,
+ &partition_fields,
+ )?;
+ let key = partition.to_serialized_bytes();
+ groups.entry(key.clone()).or_default().push(row_index);
+ if let std::collections::hash_map::Entry::Vacant(entry) =
specs.entry(key) {
+ entry.insert(
+ computer
+ .generate_part_values(&partition)?
+ .into_iter()
+ .collect(),
+ );
+ }
+ }
+
+ for (key, rows) in groups {
+ let input = take_rows(batch, &rows)?;
+ let file_batch = self.project_data_columns(&input)?;
+ if !self.writers.contains_key(&key) {
+ let spec = specs.remove(&key).unwrap();
+ let relative = self.partition_paths.relative_path(&spec)?;
+ let directory = if relative.is_empty() {
+ self.table_path.clone()
+ } else {
+ format!("{}/{relative}", self.table_path)
+ };
+ self.writers.insert(
+ key.clone(),
+ PartitionWriter {
+ spec,
+ directory,
+ open: None,
+ },
+ );
+ }
+ self.write_partition(&key, &file_batch).await?;
+ }
+ Ok(())
+ }
+
+ fn check_schema(&self, batch: &RecordBatch) -> Result<()> {
+ let expected = self.full_schema.fields();
+ let actual = batch.schema();
+ if actual.fields().len() != expected.len() {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Format Table write expects {} columns, got {}",
+ expected.len(),
+ actual.fields().len()
+ ),
+ source: None,
+ });
+ }
+ for (index, field) in expected.iter().enumerate() {
+ let given = actual.field(index);
+ if field.name() != given.name() || field.data_type() !=
given.data_type() {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Format Table column {} expects {}: {:?}, got {}:
{:?}",
+ index,
+ field.name(),
+ field.data_type(),
+ given.name(),
+ given.data_type()
+ ),
+ source: None,
+ });
+ }
+ }
+ Ok(())
+ }
+
+ fn project_data_columns(&self, batch: &RecordBatch) -> Result<RecordBatch>
{
+ RecordBatch::try_new(
+ self.file_schema.clone(),
+ self.data_indices
+ .iter()
+ .map(|&index| batch.column(index).clone())
+ .collect(),
+ )
+ .map_err(|error| crate::Error::DataInvalid {
+ message: format!("Cannot project Format Table data columns:
{error}"),
+ source: Some(Box::new(error)),
+ })
+ }
+
+ async fn write_partition(&mut self, key: &[u8], batch: &RecordBatch) ->
Result<()> {
+ let mut remaining = batch.clone();
+ while remaining.num_rows() > 0 {
+ if self.writers.get(key).unwrap().open.is_none() {
+ self.open_file(key).await?;
+ }
+ let current_rows = self
+ .writers
+ .get(key)
+ .unwrap()
+ .open
+ .as_ref()
+ .unwrap()
+ .record_count;
+ let available = (self.target_file_rows - current_rows) as usize;
+ let rows = available.min(remaining.num_rows());
+ let chunk = remaining.slice(0, rows);
+ self.writers
+ .get_mut(key)
+ .unwrap()
+ .open
+ .as_mut()
+ .unwrap()
+ .writer
+ .write(&chunk)
+ .await?;
+ remaining = remaining.slice(rows, remaining.num_rows() - rows);
+ let open =
self.writers.get_mut(key).unwrap().open.as_mut().unwrap();
+ open.record_count += rows as i64;
+ let should_roll = open.record_count >= self.target_file_rows
+ || open.writer.num_bytes() as i64 >= self.target_file_size;
+ if should_roll {
+ self.close_file(key).await?;
+ }
+ }
+ Ok(())
+ }
+
+ async fn open_file(&mut self, key: &[u8]) -> Result<()> {
+ let partition = self.writers.get(key).unwrap();
+ let file_name = format!(
+ "{}{}{}",
+ self.data_file_prefix,
+ uuid::Uuid::new_v4(),
+ self.extension
+ );
+ let target_path = format!("{}/{file_name}", partition.directory);
+ let staged_path = format!("{}/{}", self.staging_root, file_name);
+ self.table
+ .file_io()
+ .mkdirs(&format!("{}/", self.staging_root))
+ .await?;
+ self.owned_staging.insert(staged_path.clone());
+ let output = self.table.file_io().new_output(&staged_path)?;
+ let writer = create_format_writer(
+ &output,
+ self.file_schema.clone(),
+ &self.compression,
+ self.zstd_level,
+ Some(self.table.file_io().clone()),
+ Some(&self.file_fields),
+ Some(self.table.schema().options()),
+ )
+ .await?;
+ self.writers.get_mut(key).unwrap().open = Some(OpenFile {
+ writer: with_write_resources(writer, self.resources.as_ref()),
+ staged_path,
+ target_path,
+ record_count: 0,
+ });
+ Ok(())
+ }
+
+ async fn close_file(&mut self, key: &[u8]) -> Result<()> {
+ let partition = self.writers.get_mut(key).unwrap();
+ let Some(file) = partition.open.take() else {
+ return Ok(());
+ };
+ file.writer.close().await?;
+ let size = self
+ .table
+ .file_io()
+ .get_status(&file.staged_path)
+ .await?
+ .size as i64;
+ let mut message = CommitMessage::new(key.to_vec(), 0, Vec::new());
+ message.format_file = Some(FormatFileCommit {
+ staged_path: file.staged_path.clone(),
+ target_path: file.target_path,
+ partition: partition.spec.clone(),
+ record_count: file.record_count,
+ file_size: size,
+ });
+ self.owned_staging.remove(&file.staged_path);
+ self.prepared.push(message);
+ Ok(())
+ }
+
+ pub(crate) async fn prepare_commit(&mut self) ->
Result<Vec<CommitMessage>> {
+ if self.failed {
+ return Err(crate::Error::DataInvalid {
+ message: "Format Table writer failed; create a new
writer".into(),
+ source: None,
+ });
+ }
+ let keys = self.writers.keys().cloned().collect::<Vec<_>>();
+ for key in &keys {
+ if let Err(error) = self.close_file(key).await {
+ self.failed = true;
+ self.close().await;
+ return Err(error);
+ }
+ }
+ self.writers.clear();
+ Ok(std::mem::take(&mut self.prepared))
+ }
+
+ pub(crate) async fn close(&mut self) {
+ self.writers.clear();
+ for path in self.owned_staging.drain() {
+ let _ = self.table.file_io().delete_file(&path).await;
+ }
+ for message in self.prepared.drain(..) {
+ if let Some(file) = message.format_file {
+ let _ =
self.table.file_io().delete_file(&file.staged_path).await;
+ }
+ }
+ }
+}
+
+fn format_table_compression(options: &HashMap<String, String>, format: &str)
-> String {
+ options
+ .get("file.compression")
+ .or_else(|| options.get("format-table.file.compression"))
+ .or_else(|| options.get("compression"))
+ .cloned()
+ .unwrap_or_else(|| {
+ match format {
+ "parquet" => "snappy",
+ "orc" | "avro" | "mosaic" => "zstd",
+ _ => "none",
+ }
+ .to_string()
+ })
+}
diff --git a/crates/paimon/src/table/format_write_builder.rs
b/crates/paimon/src/table/format_write_builder.rs
index fc1764c1..161603ce 100644
--- a/crates/paimon/src/table/format_write_builder.rs
+++ b/crates/paimon/src/table/format_write_builder.rs
@@ -25,6 +25,8 @@ use uuid::Uuid;
pub(crate) struct FormatWriteBuilder<'a> {
table: &'a Table,
commit_user: String,
+ overwrite: bool,
+ resources: Option<ResourceContext>,
}
impl<'a> FormatWriteBuilder<'a> {
@@ -32,6 +34,8 @@ impl<'a> FormatWriteBuilder<'a> {
Self {
table,
commit_user: Uuid::new_v4().to_string(),
+ overwrite: false,
+ resources: None,
}
}
@@ -49,11 +53,13 @@ impl<'a> FormatWriteBuilder<'a> {
Ok(self)
}
- pub(crate) fn with_overwrite(self) -> Self {
+ pub(crate) fn with_overwrite(mut self) -> Self {
+ self.overwrite = true;
self
}
- pub(crate) fn with_resources(self, _resources: ResourceContext) -> Self {
+ pub(crate) fn with_resources(mut self, resources: ResourceContext) -> Self
{
+ self.resources = Some(resources);
self
}
@@ -67,9 +73,12 @@ impl<'a> FormatWriteBuilder<'a> {
}
pub(crate) fn new_write(&self) -> crate::Result<TableWrite> {
- Err(crate::Error::Unsupported {
- message: "Writing format tables is not supported by the Rust
client yet".to_string(),
- })
+ TableWrite::new_format(
+ self.table,
+ self.commit_user.clone(),
+ self.resources.clone(),
+ self.overwrite,
+ )
}
pub(crate) fn new_update(&self, _update_columns: Vec<String>) ->
crate::Result<TableUpdate> {
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index f536f3b5..231fdd86 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -49,8 +49,12 @@ mod format_partition;
mod format_partition_stats;
mod format_partition_truncate;
mod format_read_builder;
+mod format_table_commit;
mod format_table_read;
mod format_table_scan;
+#[cfg(test)]
+mod format_table_write_tests;
+mod format_table_writer;
mod format_write_builder;
#[cfg(feature = "fulltext")]
mod full_text_index_adapter;
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index 183f2926..3e4b9c08 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -31,6 +31,7 @@ use crate::spec::{
EMPTY_SERIALIZED_ROW, MANIFEST_ENTRY_SCHEMA, POSTPONE_BUCKET,
};
use crate::table::commit_message::CommitMessage;
+use crate::table::format_table_commit::FormatTableCommit;
use crate::table::global_index_build_common::same_extra_field_ids;
use crate::table::index_file_path::committed_index_file_path;
use crate::table::partition_filter::PartitionFilter;
@@ -318,6 +319,11 @@ impl TableCommit {
commit_messages: Vec<CommitMessage>,
commit_identifier: i64,
) -> Result<()> {
+ if self.table.is_format_table() {
+ return FormatTableCommit::new(&self.table)
+ .append(&commit_messages)
+ .await;
+ }
self.commit_with_identifier_impl(commit_messages, commit_identifier,
false)
.await
}
@@ -342,6 +348,11 @@ impl TableCommit {
commit_identifier: i64,
filter_committed: bool,
) -> Result<()> {
+ if self.table.is_format_table() {
+ return Err(crate::Error::Unsupported {
+ message: "Format Table commits have no checkpoint
identifiers".into(),
+ });
+ }
// A commit validates against the existing snapshot.
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
self.table.ensure_not_branch_reference_for_write()?;
@@ -446,6 +457,11 @@ impl TableCommit {
commit_messages: Vec<CommitMessage>,
static_partitions: Option<HashMap<String, Option<Datum>>>,
) -> Result<()> {
+ if self.table.is_format_table() {
+ return FormatTableCommit::new(&self.table)
+ .overwrite(&commit_messages, static_partitions.as_ref())
+ .await;
+ }
self.overwrite_impl(
commit_messages,
static_partitions,
@@ -465,6 +481,11 @@ impl TableCommit {
static_partitions: Option<HashMap<String, Option<Datum>>>,
commit_identifier: i64,
) -> Result<()> {
+ if self.table.is_format_table() {
+ return Err(crate::Error::Unsupported {
+ message: "Format Table overwrites have no checkpoint
identifiers".into(),
+ });
+ }
self.overwrite_impl(commit_messages, static_partitions,
commit_identifier, true)
.await
}
@@ -862,6 +883,11 @@ impl TableCommit {
/// files or storage errors are ignored so abort cleanup never masks the
/// original write failure.
pub async fn abort(&self, commit_messages: &[CommitMessage]) -> Result<()>
{
+ if self.table.is_format_table() {
+ return FormatTableCommit::new(&self.table)
+ .abort(commit_messages)
+ .await;
+ }
CoreOptions::new(self.table.schema().options())
.ensure_type_paimon_served(&self.table.identifier().full_name())?;
self.table.ensure_not_branch_reference_for_write()?;
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 01567d67..9901072c 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -39,6 +39,7 @@ use crate::table::commit_message::CommitMessage;
use crate::table::data_file_index_writer::FileIndexOptions;
use crate::table::data_file_writer::DataFileWriter;
use
crate::table::dedicated_format_file_writer::AppendDedicatedFormatFileWriter;
+use crate::table::format_table_writer::FormatTableWriter;
use crate::table::kv_file_writer::{KeyValueFileWriter, KeyValueWriteConfig};
use crate::table::partition_filter::PartitionFilter;
use crate::table::postpone_file_writer::{PostponeFileWriter,
PostponeWriteConfig};
@@ -123,6 +124,8 @@ impl FileWriter {
///
/// Reference: [pypaimon
BatchTableWrite](https://github.com/apache/paimon/blob/master/paimon-python/pypaimon/write/table_write.py)
pub struct TableWrite {
+ // Keep the Format Table state off ordinary Paimon write futures' stacks.
+ format_writer: Option<Box<FormatTableWriter>>,
table: Table,
write_schema: Arc<arrow_schema::Schema>,
partition_writers: HashMap<PartitionBucketKey, FileWriter>,
@@ -172,6 +175,69 @@ pub struct TableWrite {
}
impl TableWrite {
+ pub(crate) fn new_format(
+ table: &Table,
+ commit_user: String,
+ resources: Option<ResourceContext>,
+ overwrite: bool,
+ ) -> Result<Self> {
+ let format_writer = FormatTableWriter::new(table, resources.clone())?;
+ let schema = table.schema();
+ let options = CoreOptions::new(schema.options());
+ // Format Tables do not use Paimon buckets, indexes, changelogs or
+ // snapshots. Build their public TableWrite wrapper without running
+ // Paimon-only option validation or constructing stateful assigners.
+ Ok(Self {
+ format_writer: Some(Box::new(format_writer)),
+ table: table.clone(),
+ write_schema: build_target_arrow_schema(schema.fields())?,
+ partition_writers: HashMap::new(),
+ partition_computer: PartitionComputer::new(
+ schema.partition_keys(),
+ schema.fields(),
+ options.partition_default_name(),
+ options.legacy_partition_name(),
+ )?,
+ partition_keys: schema.partition_keys().to_vec(),
+ schema_id: schema.id(),
+ target_file_size: 0,
+ blob_target_file_size: 0,
+ vector_target_file_size: 0,
+ file_compression: String::new(),
+ file_compression_zstd_level: 0,
+ write_buffer_size: 0,
+ file_format: String::new(),
+ data_file_prefix: String::new(),
+ primary_key_indices: Vec::new(),
+ primary_key_types: Vec::new(),
+ sequence_field_indices: Vec::new(),
+ merge_engine: MergeEngine::Deduplicate,
+ changelog_producer: ChangelogProducer::None,
+ changelog_file_prefix: String::new(),
+ changelog_file_format: String::new(),
+ changelog_file_compression: String::new(),
+ partition_seq_cache: HashMap::new(),
+ sequence_snapshot: None,
+ commit_user,
+ postpone_write_id: 0,
+ bucket_assigner:
BucketAssignerEnum::Constant(ConstantBucketAssigner::new(
+ Vec::new(),
+ 0,
+ )),
+ is_overwrite: overwrite,
+ blob_view_fields: HashSet::new(),
+ blob_inline_fields: HashSet::new(),
+ has_blob_fields: false,
+ vector_file_format: None,
+ has_dedicated_vector_fields: false,
+ row_kind_generator: None,
+ row_kind_filter: None,
+ file_index_options: None,
+ resources,
+ failed: false,
+ })
+ }
+
pub(crate) fn new(table: &Table, commit_user: String) ->
crate::Result<Self> {
// A dynamic-bucket write reads the persisted PK hash index; the rest
are
// refused too, since their commit is blocked anyway.
@@ -407,6 +473,7 @@ impl TableWrite {
}
Ok(Self {
+ format_writer: None,
table: table.clone(),
write_schema,
partition_writers: HashMap::new(),
@@ -525,12 +592,18 @@ impl TableWrite {
/// retained key-value batches and unflushed format-writer input. Call this
/// before the first write.
pub fn with_resources(mut self, resources: ResourceContext) -> Self {
+ if let Some(writer) = self.format_writer.as_mut() {
+ writer.set_resources(resources.clone());
+ }
self.resources = Some(resources);
self
}
/// Write an Arrow RecordBatch. Rows are routed to the correct partition
and bucket.
pub async fn write_arrow_batch(&mut self, batch: &RecordBatch) ->
Result<()> {
+ if let Some(writer) = self.format_writer.as_mut() {
+ return writer.write(batch).await;
+ }
self.ensure_active()?;
let Some(batch) = self.normalize_write_batch(batch)? else {
return Ok(());
@@ -921,6 +994,10 @@ impl TableWrite {
/// Close without preparing another commit, discarding only outstanding
output.
/// Files already returned by prepare_commit belong to the caller.
pub async fn close(&mut self) {
+ if let Some(writer) = self.format_writer.as_mut() {
+ writer.close().await;
+ return;
+ }
for (_, mut writer) in self.partition_writers.drain() {
writer.abort().await;
}
@@ -938,6 +1015,9 @@ impl TableWrite {
/// commits. (`sequence_snapshot` is pinned only by the postpone path,
which
/// forbids reuse, so it is left untouched.)
pub async fn prepare_commit(&mut self) -> Result<Vec<CommitMessage>> {
+ if let Some(writer) = self.format_writer.as_mut() {
+ return writer.prepare_commit().await;
+ }
self.ensure_active()?;
self.partition_seq_cache.clear();
let writers: Vec<(PartitionBucketKey, FileWriter)> =
diff --git a/crates/paimon/tests/mock_server.rs
b/crates/paimon/tests/mock_server.rs
index 3afa0ac0..893c61ca 100644
--- a/crates/paimon/tests/mock_server.rs
+++ b/crates/paimon/tests/mock_server.rs
@@ -72,6 +72,7 @@ struct MockState {
create_partitions_calls: Vec<(String, String, CreatePartitionsRequest)>,
drop_partitions_calls: Vec<(String, String, DropPartitionsRequest)>,
create_partitions_error_status: Option<StatusCode>,
+ create_partitions_statistics_error_status: Option<StatusCode>,
list_partitions_error_status: Option<StatusCode>,
permissions: Vec<PermissionAssignment>,
list_permissions_queries: Vec<HashMap<String, String>>,
@@ -1012,6 +1013,7 @@ impl RESTServer {
return (StatusCode::NOT_FOUND, Json(error)).into_response();
}
+ let statistics_error = inner.create_partitions_statistics_error_status;
let registered_partitions = inner.partitions.entry(key).or_default();
let has_conflict = request
.partition_specs
@@ -1033,12 +1035,28 @@ impl RESTServer {
return (StatusCode::CONFLICT, Json(error)).into_response();
}
- for spec in request.partition_specs {
- if !registered_partitions
+ for (index, spec) in request.partition_specs.into_iter().enumerate() {
+ let position = registered_partitions
.iter()
- .any(|partition| partition.spec == spec)
- {
+ .position(|partition| partition.spec == spec);
+ let partition = if let Some(position) = position {
+ &mut registered_partitions[position]
+ } else {
registered_partitions.push(partition_from_spec(spec));
+ registered_partitions.last_mut().unwrap()
+ };
+ if let Some(options) = request
+ .partition_options
+ .as_ref()
+ .and_then(|options| options.get(index))
+ {
+ partition.options = Some(options.clone());
+ }
+ }
+ if request.partition_statistics.is_some() {
+ if let Some(status) = statistics_error {
+ return (status, Json(json!({"message": "Statistics update
failed"})))
+ .into_response();
}
}
// As the catalog does: a negative field was never measured and leaves
the stored value
@@ -1743,6 +1761,14 @@ impl RESTServer {
self.inner.lock().unwrap().create_partitions_error_status = status;
}
+ /// Fail only statistics-bearing partition requests, after registration.
+ pub fn set_create_partitions_statistics_error_status(&self, status:
Option<StatusCode>) {
+ self.inner
+ .lock()
+ .unwrap()
+ .create_partitions_statistics_error_status = status;
+ }
+
/// Make the list-partitions endpoint return the given status.
pub fn set_list_partitions_error_status(&self, status: Option<StatusCode>)
{
self.inner.lock().unwrap().list_partitions_error_status = status;
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index d23309da..e2dd53ec 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -1098,7 +1098,7 @@ async fn test_rest_catalog_reads_format_table() {
let table = ctx.catalog.get_table(&identifier).await.unwrap();
assert_eq!(table.location(), format_path);
assert_eq!(table.schema().options().get("path"), Some(&format_path));
- assert!(table.new_write_builder().new_write().is_err());
+ assert!(table.new_write_builder().new_write().is_ok());
let read_builder = table.new_read_builder();
let plan = read_builder.new_scan().plan().await.unwrap();
