feat(write): route postpone batch writes to fixed buckets

bindings/c/src/tests.rs

@@ -73,6 +73,17 @@ fn not_null_table_schema() -> TableSchema {
     TableSchema::new(0, &schema)
 }
 
+fn postpone_table_schema() -> TableSchema {
+    let schema = Schema::builder()
+        .column("id", DataType::Int(IntType::new()))
+        .column("name", DataType::VarChar(VarCharType::string_type()))
+        .primary_key(["id"])
+        .option("bucket", "-2")
+        .build()
+        .unwrap();
+    TableSchema::new(0, &schema)
+}
+
 unsafe fn wrap_table(table: Table) -> *mut paimon_table {
     let inner = Box::into_raw(Box::new(table)) as *mut c_void;
     Box::into_raw(Box::new(paimon_table { inner }))
@@ -950,6 +961,47 @@ fn test_write_new_builder_and_free() {
     }
 }
 
+#[test]
+fn test_postpone_fixed_bucket_builder_is_explicit() {
+    let path = "memory:/test_explicit_postpone_fixed_bucket_builder";
+    let file_io = memory_file_io();
+    setup_table_dirs(&file_io, path);
+    let table = Table::new(
+        file_io,
+        Identifier::new("default", "test"),
+        path.to_string(),
+        postpone_table_schema(),
+        None,
+    );
+    let handle = unsafe { wrap_table(table) };
+
+    unsafe {
+        let normal = paimon_table_new_write_builder(handle);
+        assert!(normal.error.is_null());
+        let normal_state = &*((*normal.write_builder).inner as *const WriteBuilderState);
+        assert!(!normal_state.postpone_fixed_bucket);
+        paimon_write_builder_free(normal.write_builder);
+
+        let fixed = paimon_table_new_postpone_fixed_bucket_write_builder(handle);
+        assert!(fixed.error.is_null());
+        let fixed_state = &*((*fixed.write_builder).inner as *const WriteBuilderState);
+        assert!(fixed_state.postpone_fixed_bucket);
+        paimon_write_builder_free(fixed.write_builder);
+
+        let commit_user = CString::new("fixed-user").unwrap();
+        let fixed = paimon_table_new_postpone_fixed_bucket_write_builder_with_commit_user(
+            handle,
+            commit_user.as_ptr(),
+        );
+        assert!(fixed.error.is_null());
+        let fixed_state = &*((*fixed.write_builder).inner as *const WriteBuilderState);
+        assert!(fixed_state.postpone_fixed_bucket);
+        assert_eq!(fixed_state.commit_user, "fixed-user");
+        paimon_write_builder_free(fixed.write_builder);
+        unwrap_table(handle);
+    }
+}
+
 #[test]
 fn test_write_commit_read_roundtrip() {
     let path = "memory:/test_write_roundtrip";
@@ -1721,6 +1773,11 @@ fn test_null_pointer_handling() {
         assert!(result.write_builder.is_null());
         paimon_error_free(result.error);
 
+        let result = paimon_table_new_postpone_fixed_bucket_write_builder(ptr::null());
+        assert!(!result.error.is_null());
+        assert!(result.write_builder.is_null());
+        paimon_error_free(result.error);
+
         let result = paimon_write_builder_new_write(ptr::null());
         assert!(!result.error.is_null());
         assert!(result.write.is_null());

bindings/c/src/types.rs

@@ -210,6 +210,7 @@ pub(crate) struct WriteBuilderState {
     pub table: Table,
     pub commit_user: String,
     pub overwrite: bool,
+    pub postpone_fixed_bucket: bool,
 }
 
 pub(crate) struct TableWriteState {

bindings/c/src/write.rs

@@ -37,6 +37,7 @@ use crate::types::*;
 unsafe fn new_write_builder(
     table: *const paimon_table,
     commit_user: Option<String>,
+    postpone_fixed_bucket: bool,
 ) -> paimon_result_write_builder {
     if let Err(e) = check_non_null(table, "table") {
         return paimon_result_write_builder {
@@ -45,7 +46,19 @@ unsafe fn new_write_builder(
         };
     }
     let table_ref = &*((*table).inner as *const Table);
-    let builder = table_ref.new_write_builder();
+    let builder = if postpone_fixed_bucket {
+        match table_ref.new_postpone_fixed_bucket_write_builder() {
+            Ok(builder) => builder,
+            Err(e) => {
+                return paimon_result_write_builder {
+                    write_builder: ptr::null_mut(),
+                    error: paimon_error::from_paimon(e),
+                }
+            }
+        }
+    } else {
+        table_ref.new_write_builder()
+    };
     let commit_user = match commit_user {
         Some(commit_user) => match builder.with_commit_user(commit_user) {
             Ok(builder) => builder.commit_user().to_string(),
@@ -62,6 +75,7 @@ unsafe fn new_write_builder(
         table: table_ref.clone(),
         commit_user,
         overwrite: false,
+        postpone_fixed_bucket,
     };
     let inner = Box::into_raw(Box::new(state)) as *mut c_void;
     paimon_result_write_builder {
@@ -82,7 +96,22 @@ unsafe fn new_write_builder(
 pub unsafe extern "C" fn paimon_table_new_write_builder(
     table: *const paimon_table,
 ) -> paimon_result_write_builder {
-    new_write_builder(table, None)
+    new_write_builder(table, None, false)
+}
+
+/// Create an explicit one-shot fixed-bucket WriteBuilder for a postpone table.
+///
+/// Normal write builders retain `bucket = -2` postpone semantics. Use this
+/// entry point when the batch must be routed to immediately visible real
+/// buckets.
+///
+/// # Safety
+/// `table` must be a valid table pointer, or null (returns error).
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_new_postpone_fixed_bucket_write_builder(
+    table: *const paimon_table,
+) -> paimon_result_write_builder {
+    new_write_builder(table, None, true)
 }
 
 /// Create a WriteBuilder with a caller-provided stable commit identity.
@@ -107,7 +136,29 @@ pub unsafe extern "C" fn paimon_table_new_write_builder_with_commit_user(
             }
         }
     };
-    new_write_builder(table, Some(commit_user))
+    new_write_builder(table, Some(commit_user), false)
+}
+
+/// Create an explicit postpone fixed-bucket WriteBuilder with a stable commit identity.
+///
+/// # Safety
+/// `table` must be a valid table pointer. `commit_user` must be a valid UTF-8
+/// C string and a safe file-name segment.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_new_postpone_fixed_bucket_write_builder_with_commit_user(
+    table: *const paimon_table,
+    commit_user: *const c_char,
+) -> paimon_result_write_builder {
+    let commit_user = match validate_cstr(commit_user, "commit_user") {
+        Ok(commit_user) => commit_user,
+        Err(error) => {
+            return paimon_result_write_builder {
+                write_builder: ptr::null_mut(),
+                error,
+            }
+        }
+    };
+    new_write_builder(table, Some(commit_user), true)
 }
 
 /// Free a paimon_write_builder.
@@ -245,19 +296,21 @@ pub unsafe extern "C" fn paimon_write_builder_new_write(
     }
     let state = &*((*wb).inner as *const WriteBuilderState);
 
-    let mut builder = match state
-        .table
-        .new_write_builder()
-        .with_commit_user(state.commit_user.clone())
-    {
-        Ok(b) => b,
-        Err(e) => {
-            return paimon_result_table_write {
-                write: ptr::null_mut(),
-                error: paimon_error::from_paimon(e),
-            }
-        }
+    let builder = if state.postpone_fixed_bucket {
+        state.table.new_postpone_fixed_bucket_write_builder()
+    } else {
+        Ok(state.table.new_write_builder())
     };
+    let mut builder =
+        match builder.and_then(|builder| builder.with_commit_user(state.commit_user.clone())) {
+            Ok(b) => b,
+            Err(e) => {
+                return paimon_result_table_write {
+                    write: ptr::null_mut(),
+                    error: paimon_error::from_paimon(e),
+                }
+            }
+        };
 
     if state.overwrite {
         builder = builder.with_overwrite();
@@ -358,8 +411,9 @@ pub unsafe extern "C" fn paimon_table_write_write_arrow_batch(
 /// Close file writers and produce CommitMessages.
 ///
 /// Consumes the open file writers (they are flushed and closed). After this
-/// call, the TableWrite can be reused — `write_arrow_batch` may be called
-/// again to start a new round of writes.
+/// call, the TableWrite can normally be reused — `write_arrow_batch` may be
+/// called again to start a new round of writes. Fixed-bucket postpone batch
+/// writers are one-shot; create a new TableWrite for the next batch.
 ///
 /// The returned `paimon_commit_messages` must be passed to a
 /// `paimon_table_commit_*` function and then freed with
@@ -781,8 +835,12 @@ pub unsafe extern "C" fn paimon_table_commit_abort(
 
 const _: unsafe extern "C" fn(*const paimon_table) -> paimon_result_write_builder =
     paimon_table_new_write_builder;
+const _: unsafe extern "C" fn(*const paimon_table) -> paimon_result_write_builder =
+    paimon_table_new_postpone_fixed_bucket_write_builder;
 const _: unsafe extern "C" fn(*const paimon_table, *const c_char) -> paimon_result_write_builder =
     paimon_table_new_write_builder_with_commit_user;
+const _: unsafe extern "C" fn(*const paimon_table, *const c_char) -> paimon_result_write_builder =
+    paimon_table_new_postpone_fixed_bucket_write_builder_with_commit_user;
 const _: unsafe extern "C" fn(*const paimon_write_builder) -> paimon_result_table_write =
     paimon_write_builder_new_write;
 const _: unsafe extern "C" fn(*const paimon_write_builder) -> paimon_result_table_commit =

crates/paimon/src/spec/core_options.rs

@@ -121,6 +121,13 @@ const DEFAULT_PARQUET_ROW_GROUP_PARALLELISM: usize = 8;
 const DEFAULT_PARQUET_ROW_GROUP_MAX_INFLIGHT_BYTES: i64 = 256 * 1024 * 1024;
 const DYNAMIC_BUCKET_TARGET_ROW_NUM_OPTION: &str = "dynamic-bucket.target-row-num";
 const DEFAULT_DYNAMIC_BUCKET_TARGET_ROW_NUM: i64 = 200_000;
+const POSTPONE_BATCH_WRITE_FIXED_BUCKET_OPTION: &str = "postpone.batch-write-fixed-bucket";
+const POSTPONE_BATCH_WRITE_FIXED_BUCKET_MAX_PARALLELISM_OPTION: &str =
+    "postpone.batch-write-fixed-bucket.max-parallelism";
+const POSTPONE_TARGET_ROW_NUM_PER_BUCKET_OPTION: &str = "postpone.target-row-num-per-bucket";
+const POSTPONE_TARGET_SIZE_PER_BUCKET_OPTION: &str = "postpone.target-size-per-bucket";
+const DEFAULT_POSTPONE_BATCH_WRITE_FIXED_BUCKET_MAX_PARALLELISM: i32 = 2048;
+const DEFAULT_POSTPONE_TARGET_SIZE_PER_BUCKET: i64 = 1024 * 1024 * 1024;
 const DEFAULT_GLOBAL_INDEX_ROW_COUNT_PER_SHARD: i64 = 100_000;
 const DEFAULT_GLOBAL_INDEX_THREAD_NUM: i64 = 32;
 const DEFAULT_GLOBAL_INDEX_FALLBACK_SCAN_MAX_SIZE: i64 = 256 * 1024 * 1024;
@@ -1148,6 +1155,75 @@ impl<'a> CoreOptions<'a> {
             .unwrap_or(DEFAULT_DYNAMIC_BUCKET_TARGET_ROW_NUM)
     }
 
+    /// Whether high-level batch integrations should select the explicit
+    /// postpone fixed-bucket writer. Direct `new_write_builder` calls retain
+    /// normal postpone semantics regardless of this option.
+    pub fn postpone_batch_write_fixed_bucket(&self) -> bool {
+        self.options
+            .get(POSTPONE_BATCH_WRITE_FIXED_BUCKET_OPTION)
+            .map(|value| value.eq_ignore_ascii_case("true"))
+            .unwrap_or(true)
+    }
+
+    pub fn postpone_batch_write_fixed_bucket_max_parallelism(&self) -> crate::Result<i32> {
+        let value = self
+            .options
+            .get(POSTPONE_BATCH_WRITE_FIXED_BUCKET_MAX_PARALLELISM_OPTION)
+            .map(|value| value.parse::<i32>())
+            .transpose()
+            .map_err(|error| crate::Error::DataInvalid {
+                message: format!(
+                    "Option '{POSTPONE_BATCH_WRITE_FIXED_BUCKET_MAX_PARALLELISM_OPTION}' must be a positive integer"
+                ),
+                source: Some(Box::new(error)),
+            })?
+            .unwrap_or(DEFAULT_POSTPONE_BATCH_WRITE_FIXED_BUCKET_MAX_PARALLELISM);
+        if value <= 0 {
+            return Err(crate::Error::DataInvalid {
+                message: format!(
+                    "Option '{POSTPONE_BATCH_WRITE_FIXED_BUCKET_MAX_PARALLELISM_OPTION}' must be positive, got: {value}"
+                ),
+                source: None,
+            });
+        }
+        Ok(value)
+    }
+
+    pub fn postpone_target_row_num_per_bucket(&self) -> crate::Result<Option<i64>> {
+        let value = self.parse_i64_option(POSTPONE_TARGET_ROW_NUM_PER_BUCKET_OPTION)?;
+        if value.is_some_and(|value| value <= 0) {
+            return Err(crate::Error::DataInvalid {
+                message: format!(
+                    "Option '{POSTPONE_TARGET_ROW_NUM_PER_BUCKET_OPTION}' must be positive, got: {}",
+                    value.unwrap()
+                ),
+                source: None,
+            });
+        }
+        Ok(value)
+    }
+
+    pub fn postpone_target_size_per_bucket(&self) -> crate::Result<i64> {
+        let value = match self.options.get(POSTPONE_TARGET_SIZE_PER_BUCKET_OPTION) {
+            Some(raw) => parse_memory_size(raw).ok_or_else(|| crate::Error::DataInvalid {
+                message: format!(
+                    "Option '{POSTPONE_TARGET_SIZE_PER_BUCKET_OPTION}' must be a valid positive memory size, got: {raw}"
+                ),
+                source: None,
+            })?,
+            None => DEFAULT_POSTPONE_TARGET_SIZE_PER_BUCKET,
+        };
+        if value <= 0 {
+            return Err(crate::Error::DataInvalid {
+                message: format!(
+                    "Option '{POSTPONE_TARGET_SIZE_PER_BUCKET_OPTION}' must be positive, got: {value}"
+                ),
+                source: None,
+            });
+        }
+        Ok(value)
+    }
+
     /// When true, blob field reads return serialized BlobDescriptor bytes
     /// instead of actual blob bytes. Default is false.
     pub fn blob_as_descriptor(&self) -> bool {

crates/paimon/src/table/commit_message.rs

@@ -27,6 +27,9 @@ pub struct CommitMessage {
     pub partition: Vec<u8>,
     /// Bucket id.
     pub bucket: i32,
+    /// Per-partition bucket count. Set by fixed-bucket postpone batch writes;
+    /// ordinary writes use the table-level bucket option.
+    pub total_buckets: Option<i32>,
     /// New data files to be added.
     pub new_files: Vec<DataFileMeta>,
     /// Snapshot id from which row-id/column conflicts should be checked.
@@ -46,6 +49,7 @@ impl CommitMessage {
         Self {
             partition,
             bucket,
+            total_buckets: None,
             new_files,
             check_from_snapshot: None,
             new_changelog_files: Vec::new(),

crates/paimon/src/table/mod.rs

@@ -72,6 +72,7 @@ mod pk_vector_indexed_split_read;
 mod pk_vector_orchestrator;
 mod pk_vector_position_read;
 mod pk_vector_scan;
+mod postpone_bucket;
 mod postpone_file_writer;
 mod prepared_files;
 mod read_builder;
@@ -359,6 +360,14 @@ impl Table {
         WriteBuilder::new(self)
     }
 
+    /// Create an explicit one-shot fixed-bucket writer for a postpone table.
+    ///
+    /// Normal write builders retain postpone-bucket semantics. Callers which
+    /// want immediately visible real buckets must opt in through this builder.
+    pub fn new_postpone_fixed_bucket_write_builder(&self) -> Result<WriteBuilder<'_>> {
+        WriteBuilder::new_postpone_fixed_bucket(self)
+    }
+
     /// Create a copy of this table with extra options merged into the schema.
     ///
     /// This never switches the schema version; it corresponds to Java