[core] Preserve append writer configuration per bucket across prepares

bindings/python/tests/test_blob_consumer.py

@@ -24,20 +24,25 @@
 from pypaimon_rust.datafusion import PaimonCatalog, SQLContext
 
 
-def _table(tmp_path):
+def _table(tmp_path, partitioned=False):
     context = SQLContext()
     context.register_catalog('paimon', {'warehouse': str(tmp_path)})
     context.sql('CREATE SCHEMA paimon.consumer')
-    context.sql("""CREATE TABLE paimon.consumer.t (id INT, payload BLOB) WITH (
+    partition = ', pt INT) PARTITIONED BY (pt)' if partitioned else ')'
+    context.sql("""CREATE TABLE paimon.consumer.t (id INT, payload BLOB{} WITH (
         'row-tracking.enabled' = 'true', 'data-evolution.enabled' = 'true',
-        'blob.copy-buffer-size' = '2 B')""")
+        'blob.copy-buffer-size' = '2 B')""".format(partition))
     return context, PaimonCatalog({'warehouse': str(tmp_path)}).get_table('consumer.t')
 
 
-def _batch():
-    return pa.record_batch([pa.array([1, 2, 3], type=pa.int32()),
-                            pa.array([b'payload', None, b''], type=pa.large_binary())],
-                           names=['id', 'payload'])
+def _batch(partition=None, first_id=1):
+    columns = [pa.array(range(first_id, first_id + 3), type=pa.int32()),
+               pa.array([b'payload', None, b''], type=pa.large_binary())]
+    names = ['id', 'payload']
+    if partition is not None:
+        columns.append(pa.array([partition] * 3, type=pa.int32()))
+        names.append('pt')
+    return pa.record_batch(columns, names=names)
 
 
 def _descriptor_payload(encoded):
@@ -54,7 +59,7 @@ def _descriptor_payload(encoded):
 
 @pytest.mark.parametrize('stream', [False, True])
 def test_blob_consumer_descriptor_bytes_and_nulls(tmp_path, stream):
-    context, table = _table(tmp_path)
+    context, table = _table(tmp_path, partitioned=True)
     builder = table.new_stream_write_builder() if stream else table.new_batch_write_builder()
     writer = builder.new_write()
     received = []
@@ -65,18 +70,22 @@ def callback(name, encoded):
 
     try:
         assert writer.with_blob_consumer(callback) is writer
-        writer.write_arrow(_batch())
-        with pytest.raises(ValueError, match='before any write'):
-            writer.with_blob_consumer(None)
+        writer.write_arrow(_batch(1))
+        assert writer.with_blob_consumer(None) is writer
+        # The first partition retains its callback, while the new partition
+        # captures the cleared factory setting.
+        writer.write_arrow(_batch(1, 4))
+        writer.write_arrow(_batch(2, 7))
         messages = writer.prepare_commit(True, 1) if stream else writer.prepare_commit()
         commit = builder.new_commit()
         commit.commit(1, messages) if stream else commit.commit(messages)
-        assert [name for name, _ in received] == ['payload'] * 3
+        assert [name for name, _ in received] == ['payload'] * 6
         assert [_descriptor_payload(encoded) if encoded is not None else None
-                for _, encoded in received] == [b'payload', None, b'']
+                for _, encoded in received] == [b'payload', None, b''] * 2
         assert pa.Table.from_batches(context.sql(
-            'SELECT id, payload FROM paimon.consumer.t ORDER BY id')).to_pydict() == {
-            'id': [1, 2, 3], 'payload': [b'payload', None, b'']}
+            'SELECT id, payload, pt FROM paimon.consumer.t ORDER BY id')).to_pydict() == {
+            'id': list(range(1, 10)), 'payload': [b'payload', None, b''] * 3,
+            'pt': [1] * 6 + [2] * 3}
     finally:
         writer.close()
     with pytest.raises(RuntimeError, match='closed'):

bindings/python/tests/test_write.py

@@ -66,27 +66,71 @@ def test_write_commit_read_roundtrip():
 
 @pytest.mark.parametrize("streaming", [False, True])
 def test_partial_write_type_roundtrip_and_lifecycle(tmp_path, streaming):
-    ctx = _make_empty_table(str(tmp_path))
+    ctx = SQLContext()
+    ctx.register_catalog("paimon", {"warehouse": str(tmp_path)})
+    ctx.sql("CREATE SCHEMA paimon.wdb")
+    ctx.sql("CREATE TABLE paimon.wdb.t (id INT, name STRING, pt STRING) PARTITIONED BY (pt)")
     table = _get_table(str(tmp_path))
     builder = table.new_stream_write_builder() if streaming else table.new_batch_write_builder()
     writer = builder.new_write()
-    assert writer.with_write_type(["id"]) is writer
-    writer.write_arrow(pa.record_batch([pa.array([], type=pa.int32())], names=["id"]))
-    assert writer.with_write_type(["id"]) is writer
+    assert writer.with_write_type(["id", "pt"]) is writer
+    empty = pa.record_batch(
+        [pa.array([], type=pa.int32()), pa.array([], type=pa.string())], names=["id", "pt"])
+    writer.write_arrow(empty)
+    assert writer.with_write_type(["name", "pt"]) is writer
+    assert writer.with_write_type(["id", "pt"]) is writer
     for columns in [[], ["missing"], ["id", "id"]]:
         with pytest.raises(ValueError):
             writer.with_write_type(columns)
-    writer.write_arrow(pa.record_batch([pa.array([1, 2], type=pa.int32())], names=["id"]))
-    with pytest.raises(ValueError, match="before any data"):
-        writer.with_write_type(["id", "name"])
+    schema = pa.schema([("id", pa.int32()), ("name", pa.string()), ("pt", pa.string())])
+    writer.write_arrow(pa.record_batch([[1, 2], ["ignored", "ignored"], ["old", "old"]], schema=schema))
+    assert writer.with_write_type(["id", "pt"]) is writer
+    for columns in [["name", "pt"], ["pt", "id"]]:
+        with pytest.raises(ValueError, match="before any write"):
+            writer.with_write_type(columns)
+    writer.write_arrow(pa.record_batch(
+        [pa.array([3, 4], type=pa.int32()), pa.array(["old", "new"])], names=["id", "pt"]))
     messages = writer.prepare_commit(True, 7) if streaming else writer.prepare_commit()
     commit = builder.new_commit()
     commit.commit(7, messages) if streaming else commit.commit(messages)
+    if streaming:
+        assert writer.prepare_commit(True, 8) == []
+    assert writer.with_write_type(["id", "pt"]) is writer
+    with pytest.raises(ValueError, match="before any write"):
+        writer.with_write_type(["name", "pt"])
     writer.close()
     with pytest.raises(RuntimeError, match="closed"):
         writer.with_write_type(["id"])
-    assert pa.Table.from_batches(ctx.sql("SELECT id, name FROM paimon.wdb.t ORDER BY id")).to_pydict() == {
-        "id": [1, 2], "name": [None, None]}
+    assert pa.Table.from_batches(ctx.sql("SELECT id, name, pt FROM paimon.wdb.t ORDER BY pt, id")).to_pydict() == {
+        "id": [4, 1, 2, 3], "name": [None] * 4, "pt": ["new", "old", "old", "old"]}
+
+
+@pytest.mark.parametrize("streaming", [False, True])
+def test_ignored_rows_do_not_lock_write_type(tmp_path, streaming):
+    ctx = SQLContext()
+    ctx.register_catalog("paimon", {"warehouse": str(tmp_path)})
+    ctx.sql("CREATE SCHEMA paimon.wdb")
+    ctx.sql("CREATE TABLE paimon.wdb.t (id INT, name STRING, op STRING) "
+            "WITH ('rowkind.field' = 'op', 'ignore-delete' = 'true')")
+    table = _get_table(str(tmp_path))
+    builder = table.new_stream_write_builder() if streaming else table.new_batch_write_builder()
+    writer = builder.new_write()
+    schema = pa.schema([("id", pa.int32()), ("name", pa.string()), ("op", pa.string())])
+    try:
+        writer.with_write_type(["id", "op"])
+        writer.write_arrow(pa.record_batch([[1], ["ignored"], ["-D"]], schema=schema))
+        assert writer.with_write_type(["name", "op"]) is writer
+        writer.write_arrow(pa.record_batch([[2], ["accepted"], ["+I"]], schema=schema))
+        assert writer.with_write_type(["name", "op"]) is writer
+        with pytest.raises(ValueError, match="before any write"):
+            writer.with_write_type(["id", "op"])
+        messages = writer.prepare_commit(True, 1) if streaming else writer.prepare_commit()
+        commit = builder.new_commit()
+        commit.commit(1, messages) if streaming else commit.commit(messages)
+    finally:
+        writer.close()
+    assert pa.Table.from_batches(ctx.sql("SELECT id, name FROM paimon.wdb.t")).to_pydict() == {
+        "id": [None], "name": ["accepted"]}
 
 
 @pytest.mark.parametrize("streaming", [False, True])

crates/paimon/src/table/blob_consumer_tests.rs

@@ -94,7 +94,9 @@ async fn append_consumer_survives_file_rolls_and_checkpoint_recreation() {
                 .sum::<i64>(),
             2
         );
-        assert!(writer.with_blob_consumer(None).is_err());
+        // Java changes the factory setting while an existing bucket keeps the
+        // consumer it captured, including across consecutive prepares.
+        writer.with_blob_consumer(None).unwrap();
     }
     writer.close().await;
     let events = events.lock().unwrap().clone();

crates/paimon/src/table/internal.rs

@@ -15,15 +15,17 @@
 // specific language governing permissions and limitations
 // under the License.
 
-//! Unstable scheduling bridge for the repository's language bindings.
+//! Unstable execution bridge for the repository's language bindings.
 //!
 //! Source plans are the public input to search readers. Candidate and scoring
 //! requests below are internal execution stages: Core owns their ordering and
-//! global result coordination. This module is available only with `internal-api`.
+//! global result coordination. Writer adapters forward their existing commit
+//! identifiers through the same bridge. This module is available only with `internal-api`.
 
 pub use super::search_task::{SearchTask, SearchTaskExecutor, SearchTaskResult};
 use super::{
-    BatchVectorRead, GlobalIndexResult, HybridSearchBuilder, Table, VectorRead, VectorScanPlan,
+    BatchVectorRead, CommitMessage, GlobalIndexResult, HybridSearchBuilder, Table, TableWrite,
+    VectorRead, VectorScanPlan,
 };
 
 pub async fn execute_search_task(
@@ -65,3 +67,13 @@ pub async fn execute_hybrid_with_executor(
 ) -> crate::Result<GlobalIndexResult> {
     builder.execute_with_executor(executor).await
 }
+
+/// Forward an identifier already supplied by the batch or stream facade.
+pub async fn prepare_commit_with_identifier(
+    writer: &mut TableWrite,
+    commit_identifier: i64,
+) -> crate::Result<Vec<CommitMessage>> {
+    writer
+        .prepare_commit_with_identifier(commit_identifier)
+        .await
+}