[core] Expose postpone bucket plans through the write builder

bindings/python/project-description.md

@@ -53,6 +53,38 @@ ctx.sql("INSERT INTO paimon.my_db.users VALUES (1, 'alice'), (2, 'bob')")
 batches = ctx.sql("SELECT * FROM paimon.my_db.users")
 ```
 
+### Blob Source Streams and Commit Identity
+
+The write and update APIs accept a `UriReaderFactory` for custom Blob reference
+sources. Both stream protocols are public runtime imports and support structural
+typing; implementations do not need to inherit from them.
+
+```python
+from io import BytesIO
+from pypaimon_rust.datafusion import UriReader, UriReaderFactory
+
+payloads = {"custom://example": b"content"}
+
+class Reader:
+    def new_input_stream(self, uri: str):
+        return BytesIO(payloads[uri])
+
+class Factory:
+    def create(self, uri: str) -> UriReader:
+        return Reader()
+
+factory: UriReaderFactory = Factory()
+builder = table.new_batch_write_builder().with_commit_user("my-job")
+writer = builder.new_write().with_blob_uri_reader_factory(factory)
+update = builder.new_update().with_blob_uri_reader_factory(factory)
+```
+
+Rust copies the requested Blob ranges and closes the source streams. Pass `None`
+to clear the factory. Configure a writer before its first write. Batch and stream
+updates, including incremental row-ID updaters, expose the same chainable setter.
+`with_commit_user` is available on batch, stream, and postpone fixed bucket write
+builders, and the identity applies to both writers and committers from that builder.
+
 ### Multimodal SQL Helpers
 
 `SQLContext` also registers Python scalar UDFs for common BLOB media and vector

bindings/python/python/pypaimon_rust/__init__.py

@@ -16,3 +16,10 @@
 # under the License.
 
 from .pypaimon_rust import *
+from .pypaimon_rust import datafusion as datafusion
+from .uri_reader import UriReader as UriReader, UriReaderFactory as UriReaderFactory
+
+# The native submodule is registered during extension initialization. Export the
+# Python stream protocols beside its writers and update APIs.
+datafusion.UriReader = UriReader
+datafusion.UriReaderFactory = UriReaderFactory

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -18,22 +18,18 @@
 from datetime import datetime
 from os import PathLike
 from typing import (
-    Any, BinaryIO, Callable, Dict, Iterable, List, Literal, Optional, Protocol, Sequence, Tuple,
+    Any, Callable, Dict, Iterable, List, Literal, Optional, Sequence, Tuple,
     TypeAlias, Union,
 )
 
 import pyarrow
 
+from .uri_reader import UriReader as UriReader, UriReaderFactory as UriReaderFactory
+
 ArrowTypeLike: TypeAlias = Union[pyarrow.DataType, pyarrow.Field, str]
 InputFieldsLike: TypeAlias = Union[ArrowTypeLike, Sequence[ArrowTypeLike]]
 VolatilityLike: TypeAlias = Union[str, Any]
 
-class _UriReader(Protocol):
-    def new_input_stream(self, uri: str) -> BinaryIO: ...
-
-class _UriReaderFactory(Protocol):
-    def create(self, uri: str) -> _UriReader: ...
-
 class DataField:
     def id(self) -> int: ...
     def name(self) -> str: ...
@@ -462,7 +458,7 @@ class CommitMessage:
 
 class BatchTableWrite:
     def with_blob_uri_reader_factory(
-        self, factory: Optional[_UriReaderFactory]
+        self, factory: Optional[UriReaderFactory]
     ) -> "BatchTableWrite":
         """Use custom source streams for Blob references; Rust copies bounded
         ranges, reuses consecutive sources and closes them on completion."""
@@ -481,7 +477,9 @@ class BatchTableWrite:
         ...
 
 class BatchTableUpdate:
-    def _with_blob_uri_reader_factory(self, factory: Optional[Any]) -> None: ...
+    def with_blob_uri_reader_factory(
+        self, factory: Optional[UriReaderFactory]
+    ) -> "BatchTableUpdate": ...
     def merge_into(
         self, source: Optional[Union[pyarrow.Table, Table]], *, on: Sequence[Tuple[str, str]],
         when_matched: Sequence[Dict[str, Any]], when_not_matched: Sequence[Dict[str, Any]]
@@ -531,7 +529,9 @@ class BatchTableUpdate:
         ...
 
 class TableUpdateByRowId:
-    def _with_blob_uri_reader_factory(self, factory: Optional[Any]) -> None: ...
+    def with_blob_uri_reader_factory(
+        self, factory: Optional[UriReaderFactory]
+    ) -> "TableUpdateByRowId": ...
     def update_columns(self, data: pyarrow.Table, column_names: List[str]) -> List[CommitMessage]:
         """Stage one input table; return the cumulative messages to commit once."""
         ...
@@ -540,7 +540,9 @@ class TableUpdateByRowId:
     def _abort(self) -> None: ...
 
 class StreamTableUpdate:
-    def _with_blob_uri_reader_factory(self, factory: Optional[Any]) -> None: ...
+    def with_blob_uri_reader_factory(
+        self, factory: Optional[UriReaderFactory]
+    ) -> "StreamTableUpdate": ...
     def merge_into(
         self, source: Optional[Union[pyarrow.Table, Table]], *, on: Sequence[Tuple[str, str]],
         when_matched: Sequence[Dict[str, Any]], when_not_matched: Sequence[Dict[str, Any]],
@@ -591,7 +593,7 @@ class StreamTableUpdate:
 
 class StreamTableWrite:
     def with_blob_uri_reader_factory(
-        self, factory: Optional[_UriReaderFactory]
+        self, factory: Optional[UriReaderFactory]
     ) -> "StreamTableWrite": ...
     def with_write_type(self, columns: List[str]) -> "StreamTableWrite": ...
     def with_blob_consumer(
@@ -740,8 +742,8 @@ class StreamTableCommit:
         ...
 
 class BatchWriteBuilder:
-    def _with_commit_user(self, commit_user: str) -> "BatchWriteBuilder":
-        """Internal PyPaimon bridge for commits produced by an external Python writer."""
+    def with_commit_user(self, commit_user: str) -> "BatchWriteBuilder":
+        """Set the commit identity shared by this builder's writers and committers."""
         ...
     def with_overwrite(self, static_partition: Optional[Dict[str, Any]] = {}) -> "BatchWriteBuilder":
         """Configure both writer and committer. Explicit None restores append.
@@ -772,7 +774,16 @@ class PostponeFixedBucketWriteBuilder:
     Unknown partitions buffer until preparation; known layouts write directly.
     Rescaling and compaction are not performed.
     """
-    def _with_commit_user(self, commit_user: str) -> "PostponeFixedBucketWriteBuilder": ...
+    def with_commit_user(self, commit_user: str) -> "PostponeFixedBucketWriteBuilder": ...
+    def resolve_bucket_plan(self) -> pyarrow.RecordBatch:
+        """Known partition layouts from the latest writer snapshot, without SELECT.
+
+        Uses the with_bucket_plan schema. Explicit plans take precedence;
+        overwrite otherwise returns no existing layouts. Configured defaults
+        do not create partition rows in this metadata plan.
+        This method does not infer bucket counts for new input partitions.
+        """
+        ...
     def with_bucket_plan(self, plan: pyarrow.RecordBatch) -> "PostponeFixedBucketWriteBuilder":
         """Partition columns in schema order followed by total_buckets: int32.
 

bindings/python/python/pypaimon_rust/uri_reader.py

@@ -0,0 +1,34 @@
+# 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.
+
+"""Source stream contracts for managed Blob URI reads."""
+
+from typing import BinaryIO, Protocol
+
+__all__ = ["UriReader", "UriReaderFactory"]
+
+
+class UriReader(Protocol):
+    """Open readable binary streams; Rust closes each stream after use."""
+
+    def new_input_stream(self, uri: str) -> BinaryIO: ...
+
+
+class UriReaderFactory(Protocol):
+    """Select a URI reader while leaving payload copying to Rust Core."""
+
+    def create(self, uri: str) -> UriReader: ...