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 eefe3ece feat(python): deserialize cross-language splits (#889)
eefe3ece is described below
commit eefe3eceb215d185fd47acf353ef8c6866f11482
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Sep 20 20:06:01 2026 +0800
feat(python): deserialize cross-language splits (#889)
---
.../python/python/pypaimon_rust/datafusion.pyi | 7 +++-
bindings/python/src/read.rs | 13 ++++++
bindings/python/tests/test_read.py | 46 +++++++++++++++++++++-
3 files changed, 64 insertions(+), 2 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index d179df9d..a91ad1da 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -39,9 +39,14 @@ class TableSchema:
class Split:
def __init__(self, state: bytes) -> None: ...
+ @staticmethod
+ def deserialize(state: bytes) -> "Split":
+ """Decode a Java-compatible SplitSerializer v1 frame."""
+ ...
def row_count(self) -> int: ...
- # Java SplitSerializer v1 binary: a DataSplit (v8), or an IndexedSplit
when the split has row ranges.
def is_streaming(self) -> bool: ...
+ # Java SplitSerializer v1 binary: DataSplit v8/v9, or score-free
+ # IndexedSplit when the native split has row ranges.
def serialize(self) -> bytes: ...
class Plan:
diff --git a/bindings/python/src/read.rs b/bindings/python/src/read.rs
index f73a8ded..22af9ee6 100644
--- a/bindings/python/src/read.rs
+++ b/bindings/python/src/read.rs
@@ -773,6 +773,19 @@ impl PySplit {
Ok(PyBytes::new(py, &bytes))
}
+ /// Reconstruct a native split from the stable, cross-language
+ /// `SplitSerializer` v1 wire format.
+ ///
+ /// This is deliberately separate from the constructor used by pickle:
+ /// pickle bytes are an opaque Rust JSON encoding, while this method
accepts
+ /// Java/Python-compatible DataSplit and score-free IndexedSplit frames.
+ #[staticmethod]
+ fn deserialize(state: &Bound<'_, PyBytes>) -> PyResult<Self> {
+ Ok(Self {
+ inner:
DataSplit::deserialize_split_v1(state.as_bytes()).map_err(to_py_err)?,
+ })
+ }
+
/// Reduce to `Split(bytes)` for pickle/copy. The bytes are an opaque,
/// implementation-detail encoding; only same/compatible-version round-trip
/// is guaranteed.
diff --git a/bindings/python/tests/test_read.py
b/bindings/python/tests/test_read.py
index fa836077..c072f32e 100644
--- a/bindings/python/tests/test_read.py
+++ b/bindings/python/tests/test_read.py
@@ -21,7 +21,7 @@ import tempfile
import pyarrow as pa
import pytest
-from pypaimon_rust.datafusion import PaimonCatalog, SQLContext
+from pypaimon_rust.datafusion import PaimonCatalog, Split, SQLContext
def _make_table_with_data(warehouse):
@@ -174,6 +174,50 @@ def test_split_pickle_roundtrip():
assert restored.row_count() == split.row_count()
+def test_split_wire_roundtrip_reads_same_rows():
+ with tempfile.TemporaryDirectory() as warehouse:
+ table = _make_table_with_data(warehouse)
+ builder = table.new_read_builder()
+ split = builder.new_scan().plan().splits()[0]
+
+ restored = Split.deserialize(split.serialize())
+
+ assert restored.row_count() == split.row_count()
+ original = pa.Table.from_batches(builder.new_read().read([split]))
+ roundtrip = pa.Table.from_batches(builder.new_read().read([restored]))
+ assert roundtrip.equals(original)
+
+
+def test_indexed_split_wire_roundtrip_preserves_row_ranges():
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.rdb")
+ ctx.sql("""CREATE TABLE paimon.rdb.de (id INT, name STRING) WITH (
+ 'row-tracking.enabled' = 'true',
+ 'data-evolution.enabled' = 'true')""")
+ ctx.sql("""INSERT INTO paimon.rdb.de (id, name)
+ VALUES (1, 'a'), (2, 'b'), (3, 'c')""")
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("rdb.de")
+ builder = table.new_read_builder().with_row_ranges([(1, 1)])
+ split = builder.new_scan().plan().splits()[0]
+
+ restored = Split.deserialize(split.serialize())
+ rows = pa.Table.from_batches(builder.new_read().read([restored]))
+
+ assert restored.row_count() == 3
+ assert rows.column("id").to_pylist() == [2]
+
+
[email protected]("payload, message", [
+ (b"", "buffer underrun"),
+ (b"not-a-split", "magic"),
+])
+def test_split_wire_deserialize_rejects_malformed_input(payload, message):
+ with pytest.raises(ValueError, match=message):
+ Split.deserialize(payload)
+
+
def _make_partitioned_table(warehouse):
ctx = SQLContext()
ctx.register_catalog("paimon", {"warehouse": warehouse})
