You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "JunRuiLee (via GitHub)" <gi...@apache.org> on 2026/06/29 03:24:58 UTC
[PR] [python] Expose DataFrame-style batch write + commit loop [paimon-rust]
JunRuiLee opened a new pull request, #420:
URL: https://github.com/apache/paimon-rust/pull/420
### Purpose
First PR of exposing PyPaimon's DataFrame write path to Rust. Refs #414.
Exposes the Rust core's batch write loop through `bindings/python`:
```python
wb = table.new_write_builder()
write = wb.new_write()
write.write_arrow(batch) # PyArrow RecordBatch, may be called repeatedly
messages = write.prepare_commit()
wb.new_commit().commit(messages) # persists a snapshot; readable via SQL
```
`CommitMessage` (from `prepare_commit()`) is an opaque object passed back to `commit()` — same-process only for now; cross-process (pickle, for Ray worker→driver) is a later PR.
Out of scope (later PRs): serializable `CommitMessage`, overwrite, pypaimon-side wiring.
### Notes
- `write_arrow` validates the incoming batch schema against the table schema and raises `ValueError` on mismatch (names + types; binary/fixed_size_binary tolerated), mirroring pypaimon's `_validate_pyarrow_schema`. No implicit casting — callers supply correctly-typed batches (e.g. `INT` → Arrow `int32`).
- A single `commit_user` is fixed per `WriteBuilder` and shared by `new_write()` and `new_commit()`, so the writer and committer agree (Paimon uses it for duplicate-commit detection).
- `commit([])` (empty messages) is a no-op.
### Tests
`bindings/python/tests/test_write.py` — 6 tests: write→commit→SQL read-back roundtrip, multi-batch, empty-commit no-op, schema-mismatch ValueError, non-message TypeError.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Expose DataFrame-style batch write + commit loop [paimon-rust]
Posted by "JunRuiLee (via GitHub)" <gi...@apache.org>.
JunRuiLee commented on code in PR #420:
URL: https://github.com/apache/paimon-rust/pull/420#discussion_r3492785739
##########
bindings/python/src/write.rs:
##########
@@ -0,0 +1,184 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::datatypes::{DataType as ArrowDataType, Schema as ArrowSchema};
+use arrow::pyarrow::FromPyArrow;
+use arrow::record_batch::RecordBatch;
+use paimon::table::{CommitMessage, Table, TableCommit, TableWrite};
+use paimon_datafusion::runtime::runtime;
+use pyo3::exceptions::{PyTypeError, PyValueError};
+use pyo3::prelude::*;
+
+use crate::error::to_py_err;
+
+/// Validate an incoming batch schema against the table's target Arrow schema,
+/// replicating pypaimon's `_validate_pyarrow_schema` semantics (minus the
+/// projection-write branch PR1 lacks): field count, order, and names must match;
+/// types must match field-by-field, tolerating binary-family interchange
+/// (Binary / LargeBinary / FixedSizeBinary). The nullable flag is intentionally
+/// NOT compared, since `build_target_arrow_schema` derives nullability from the
+/// Paimon field while pyarrow-constructed batches infer nullable=true. No cast.
+fn validate_batch_schema(input: &ArrowSchema, target: &ArrowSchema) -> PyResult<()> {
+ let mismatch = || {
+ PyValueError::new_err(format!(
+ "Input schema is not consistent with the table schema. \
+ input: {input:?}, table: {target:?}"
+ ))
+ };
+ if input.fields().len() != target.fields().len() {
+ return Err(mismatch());
+ }
+ for (i, t) in input.fields().iter().zip(target.fields().iter()) {
+ if i.name() != t.name() {
+ return Err(mismatch());
+ }
+ if i.data_type() != t.data_type()
+ && !(is_binary_family(i.data_type()) && is_binary_family(t.data_type()))
Review Comment:
Good catch — confirmed the lower write path downcasts binary fields to arrow_array::BinaryArray only, so tolerating LargeBinary/FixedSizeBinary here was a false positive. Made type matching strict (reject rather than cast, consistent with write_arrow not casting) and added a regression test. Fixed in 2fe6b39.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Expose DataFrame-style batch write + commit loop [paimon-rust]
Posted by "JunRuiLee (via GitHub)" <gi...@apache.org>.
JunRuiLee commented on code in PR #420:
URL: https://github.com/apache/paimon-rust/pull/420#discussion_r3497487141
##########
bindings/python/tests/test_write.py:
##########
@@ -0,0 +1,165 @@
+# 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.
+
+import tempfile
+
+import pyarrow as pa
+import pytest
+
+from pypaimon_rust.datafusion import PaimonCatalog, SQLContext
+
+# The table created by _make_empty_table is (id INT, name STRING). Paimon INT maps
+# to Arrow int32, so batches must use int32 for id — pyarrow infers Python ints as
+# int64, which write_arrow now (correctly, matching pypaimon) rejects as a type
+# mismatch. Build batches against this explicit schema to match the table.
+_TABLE_SCHEMA = pa.schema([("id", pa.int32()), ("name", pa.string())])
+
+
+def _batch(ids, names):
+ return pa.record_batch([ids, names], schema=_TABLE_SCHEMA)
+
+
+def _make_empty_table(warehouse):
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.wdb")
+ ctx.sql("CREATE TABLE paimon.wdb.t (id INT, name STRING)")
+ return ctx
+
+
+def _get_table(warehouse):
+ return PaimonCatalog({"warehouse": warehouse}).get_table("wdb.t")
+
+
+def test_write_commit_read_roundtrip():
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = _make_empty_table(warehouse)
+ table = _get_table(warehouse)
+ batch = _batch([1, 2, 3], ["a", "b", "c"])
+ wb = table.new_write_builder()
+ write = wb.new_write()
+ write.write_arrow(batch)
+ messages = write.prepare_commit()
+ assert len(messages) >= 1 # cover API shape in the first test
+ wb.new_commit().commit(messages) # same wb → shared commit_user
+ result = pa.Table.from_batches(
+ ctx.sql("SELECT id, name FROM paimon.wdb.t")
+ ).sort_by("id").to_pydict()
+ assert result == {"id": [1, 2, 3], "name": ["a", "b", "c"]}
+
+
+def test_write_multiple_batches():
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = _make_empty_table(warehouse)
+ table = _get_table(warehouse)
+ wb = table.new_write_builder()
+ write = wb.new_write()
+ write.write_arrow(_batch([1], ["a"]))
+ write.write_arrow(_batch([2], ["b"]))
+ messages = write.prepare_commit()
+ wb.new_commit().commit(messages)
+ result = pa.Table.from_batches(
+ ctx.sql("SELECT id, name FROM paimon.wdb.t")
+ ).sort_by("id").to_pydict()
+ assert result == {"id": [1, 2], "name": ["a", "b"]}
+
+
+def test_prepare_commit_returns_messages():
+ with tempfile.TemporaryDirectory() as warehouse:
+ _make_empty_table(warehouse)
+ table = _get_table(warehouse)
+ write = table.new_write_builder().new_write()
+ write.write_arrow(_batch([1], ["a"]))
+ messages = write.prepare_commit()
+ assert len(messages) >= 1
+ assert all(type(m).__name__ == "CommitMessage" for m in messages)
+
+
+def test_commit_empty_messages_noop():
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = _make_empty_table(warehouse)
+ table = _get_table(warehouse)
+ wb = table.new_write_builder()
+ messages = wb.new_write().prepare_commit() # no write
+ assert messages == []
+ wb.new_commit().commit(messages) # no-op success
+ batches = ctx.sql("SELECT COUNT(*) AS cnt FROM paimon.wdb.t")
+ assert batches[0].column(0).to_pylist() == [0]
+
+
+def test_write_arrow_type_mismatch_raises():
+ with tempfile.TemporaryDirectory() as warehouse:
+ _make_empty_table(warehouse) # table (id INT, name STRING)
+ table = _get_table(warehouse)
+ write = table.new_write_builder().new_write()
+ bad = pa.record_batch([["x", "y"], ["a", "b"]], names=["id", "name"]) # id as STRING
+ with pytest.raises(ValueError):
+ write.write_arrow(bad)
+
+
+def test_write_arrow_binary_family_mismatch_raises():
+ # A BINARY column requires Arrow `binary`; a near-equivalent `large_binary`
+ # must be rejected at validation (it would otherwise fail deeper, since the
+ # write path downcasts binary fields to arrow_array::BinaryArray only).
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.wdb")
+ ctx.sql("CREATE TABLE paimon.wdb.bt (id INT, data BINARY)")
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("wdb.bt")
+ write = table.new_write_builder().new_write()
+ schema = pa.schema([("id", pa.int32()), ("data", pa.large_binary())])
+ bad = pa.record_batch([[1], [b"x"]], schema=schema)
+ with pytest.raises(ValueError):
+ write.write_arrow(bad)
+
+
+def test_commit_non_message_raises_typeerror():
+ with tempfile.TemporaryDirectory() as warehouse:
+ _make_empty_table(warehouse)
+ table = _get_table(warehouse)
+ with pytest.raises(TypeError):
+ table.new_write_builder().new_commit().commit([object()])
+ # A non-iterable argument also raises TypeError (not a raw PyO3 error).
+ with pytest.raises(TypeError):
+ table.new_write_builder().new_commit().commit(42)
+
+
+def test_commit_cross_table_messages_raises():
+ # Messages prepared for one table must not be committed by another table's
+ # committer (would persist a snapshot referencing data files written
+ # elsewhere). The wrapper stamps each message with its source table location
+ # and the committer rejects mismatches.
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.wdb")
+ ctx.sql("CREATE TABLE paimon.wdb.t1 (id INT, name STRING)")
+ ctx.sql("CREATE TABLE paimon.wdb.t2 (id INT, name STRING)")
+ catalog = PaimonCatalog({"warehouse": warehouse})
+ t1 = catalog.get_table("wdb.t1")
+ t2 = catalog.get_table("wdb.t2")
+ batch = pa.record_batch(
+ [pa.array([1], pa.int32()), pa.array(["a"], pa.string())],
+ names=["id", "name"],
+ )
+ w1 = t1.new_write_builder().new_write()
+ w1.write_arrow(batch)
+ messages = w1.prepare_commit()
+ with pytest.raises(ValueError):
+ t2.new_write_builder().new_commit().commit(messages)
+
Review Comment:
Removed, thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Expose DataFrame-style batch write + commit loop [paimon-rust]
Posted by "JunRuiLee (via GitHub)" <gi...@apache.org>.
JunRuiLee commented on code in PR #420:
URL: https://github.com/apache/paimon-rust/pull/420#discussion_r3497387110
##########
bindings/python/src/write.rs:
##########
@@ -0,0 +1,180 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::datatypes::Schema as ArrowSchema;
+use arrow::pyarrow::FromPyArrow;
+use arrow::record_batch::RecordBatch;
+use paimon::table::{CommitMessage, Table, TableCommit, TableWrite};
+use paimon_datafusion::runtime::runtime;
+use pyo3::exceptions::{PyTypeError, PyValueError};
+use pyo3::prelude::*;
+
+use crate::error::to_py_err;
+
+/// Validate an incoming batch schema against the table's target Arrow schema:
+/// field count, order, and names must match, and types must match exactly. The
+/// nullable flag is intentionally NOT compared, since `build_target_arrow_schema`
+/// derives nullability from the Paimon field while pyarrow-constructed batches
+/// infer nullable=true. No cast — callers supply correctly-typed batches.
+///
+/// Type matching is strict (no binary-family interchange): the lower write path
+/// downcasts to the exact Arrow array for each Paimon type (e.g. a `Binary` /
+/// `VarBinary` field requires `arrow_array::BinaryArray`, not `LargeBinary` /
+/// `FixedSizeBinary`). Accepting a near-equivalent type here would pass
+/// validation but then fail deeper with a type-mismatch (or write files whose
+/// Arrow schema differs from the table), so it is rejected up front.
+fn validate_batch_schema(input: &ArrowSchema, target: &ArrowSchema) -> PyResult<()> {
+ let mismatch = || {
+ PyValueError::new_err(format!(
+ "Input schema is not consistent with the table schema. \
+ input: {input:?}, table: {target:?}"
+ ))
+ };
+ if input.fields().len() != target.fields().len() {
+ return Err(mismatch());
+ }
+ for (i, t) in input.fields().iter().zip(target.fields().iter()) {
+ if i.name() != t.name() {
+ return Err(mismatch());
+ }
+ if i.data_type() != t.data_type() {
+ return Err(mismatch());
+ }
+ }
+ Ok(())
+}
+
+/// Builder for the batch write loop, created via [`crate::table::PyTable::new_write_builder`].
+///
+/// Holds the owning table plus a single fixed `commit_user`, generated once and
+/// shared by both `new_write()` and `new_commit()` so that writers and the
+/// committer agree on the commit user (Paimon uses it for duplicate-commit
+/// detection). Creating a fresh `WriteBuilder` per call would otherwise mint a
+/// new random UUID each time.
+#[pyclass(name = "WriteBuilder", module = "pypaimon_rust.datafusion")]
+pub struct PyWriteBuilder {
+ table: Arc<Table>,
+ commit_user: String,
+}
+
+impl PyWriteBuilder {
+ pub fn new(table: Arc<Table>) -> Self {
+ let commit_user = table.new_write_builder().commit_user().to_string();
+ Self { table, commit_user }
+ }
+}
+
+#[pymethods]
+impl PyWriteBuilder {
+ /// Create a writer for accumulating Arrow batches.
+ fn new_write(&self) -> PyResult<PyTableWrite> {
+ let builder = self
+ .table
+ .new_write_builder()
+ .with_commit_user(self.commit_user.clone())
+ .map_err(to_py_err)?;
+ let target_schema = paimon::arrow::build_target_arrow_schema(self.table.schema().fields())
+ .map_err(to_py_err)?;
+ Ok(PyTableWrite {
+ inner: builder.new_write().map_err(to_py_err)?,
+ target_schema,
+ })
+ }
+
+ /// Create a committer for persisting prepared commit messages.
+ fn new_commit(&self) -> PyResult<PyTableCommit> {
+ let builder = self
+ .table
+ .new_write_builder()
+ .with_commit_user(self.commit_user.clone())
+ .map_err(to_py_err)?;
+ Ok(PyTableCommit {
+ inner: builder.new_commit(),
+ })
+ }
+}
+
+/// A stateful writer that accumulates Arrow batches until `prepare_commit`.
+///
+/// Marked `unsendable`: the underlying `TableWrite` holds file writers that are
+/// not `Sync`, so the object enforces single-thread access at runtime.
+#[pyclass(name = "TableWrite", module = "pypaimon_rust.datafusion", unsendable)]
+pub struct PyTableWrite {
+ inner: TableWrite,
+ /// The table's target Arrow schema, used to validate incoming batches.
+ target_schema: Arc<ArrowSchema>,
+}
+
+#[pymethods]
+impl PyTableWrite {
+ /// Write a single PyArrow RecordBatch into the table's writers.
+ fn write_arrow(&mut self, py: Python<'_>, batch: &Bound<'_, PyAny>) -> PyResult<()> {
+ let batch = RecordBatch::from_pyarrow_bound(batch)?;
+ validate_batch_schema(&batch.schema(), &self.target_schema)?;
+ let rt = runtime();
+ py.detach(|| rt.block_on(async { self.inner.write_arrow_batch(&batch).await }))
+ .map_err(to_py_err)
+ }
+
+ /// Close writers and return the commit messages (opaque; pass to commit()).
+ fn prepare_commit(&mut self, py: Python<'_>) -> PyResult<Vec<PyCommitMessage>> {
+ let rt = runtime();
+ let messages = py
+ .detach(|| rt.block_on(async { self.inner.prepare_commit().await }))
+ .map_err(to_py_err)?;
+ Ok(messages
+ .into_iter()
+ .map(|inner| PyCommitMessage { inner })
+ .collect())
+ }
+}
+
+/// A committer that persists prepared commit messages as a snapshot.
+#[pyclass(name = "TableCommit", module = "pypaimon_rust.datafusion")]
+pub struct PyTableCommit {
+ inner: TableCommit,
+}
+
+#[pymethods]
+impl PyTableCommit {
+ /// Commit the given commit messages. Empty input is a no-op success.
+ fn commit(&self, py: Python<'_>, messages: &Bound<'_, PyAny>) -> PyResult<()> {
+ let mut inner_messages = Vec::new();
+ let iter = messages.try_iter().map_err(|_| {
+ PyTypeError::new_err("commit() expects a sequence of CommitMessage objects")
+ })?;
+ for item in iter {
+ let item = item?;
+ let msg: PyRef<PyCommitMessage> = item.extract().map_err(|_| {
+ PyTypeError::new_err("commit() expects a sequence of CommitMessage objects")
+ })?;
+ inner_messages.push(msg.inner.clone());
Review Comment:
Confirmed and reproduced — commit() accepted any CommitMessage with no table check, so t1's messages committed via t2 persist a snapshot referencing t1's files and t2 then fails to read.
Guarded at the wrapper boundary: PyCommitMessage now carries its source table location (stamped in prepare_commit), and commit() rejects any message whose location differs from the committer's table before cloning the inner message. Added a regression test for the cross-table flow. Pushed in ea4cbf0.
Note: once CommitMessage pickling lands (a follow-up PR for Ray distributed commit), the serialized payload will include table_location so this guard survives cross-process round-trips.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Expose DataFrame-style batch write + commit loop [paimon-rust]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #420:
URL: https://github.com/apache/paimon-rust/pull/420#discussion_r3491241866
##########
bindings/python/src/write.rs:
##########
@@ -0,0 +1,184 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::datatypes::{DataType as ArrowDataType, Schema as ArrowSchema};
+use arrow::pyarrow::FromPyArrow;
+use arrow::record_batch::RecordBatch;
+use paimon::table::{CommitMessage, Table, TableCommit, TableWrite};
+use paimon_datafusion::runtime::runtime;
+use pyo3::exceptions::{PyTypeError, PyValueError};
+use pyo3::prelude::*;
+
+use crate::error::to_py_err;
+
+/// Validate an incoming batch schema against the table's target Arrow schema,
+/// replicating pypaimon's `_validate_pyarrow_schema` semantics (minus the
+/// projection-write branch PR1 lacks): field count, order, and names must match;
+/// types must match field-by-field, tolerating binary-family interchange
+/// (Binary / LargeBinary / FixedSizeBinary). The nullable flag is intentionally
+/// NOT compared, since `build_target_arrow_schema` derives nullability from the
+/// Paimon field while pyarrow-constructed batches infer nullable=true. No cast.
+fn validate_batch_schema(input: &ArrowSchema, target: &ArrowSchema) -> PyResult<()> {
+ let mismatch = || {
+ PyValueError::new_err(format!(
+ "Input schema is not consistent with the table schema. \
+ input: {input:?}, table: {target:?}"
+ ))
+ };
+ if input.fields().len() != target.fields().len() {
+ return Err(mismatch());
+ }
+ for (i, t) in input.fields().iter().zip(target.fields().iter()) {
+ if i.name() != t.name() {
+ return Err(mismatch());
+ }
+ if i.data_type() != t.data_type()
+ && !(is_binary_family(i.data_type()) && is_binary_family(t.data_type()))
Review Comment:
This accepts LargeBinary/FixedSizeBinary as equivalent to Binary, but the batch is then passed through unchanged. The lower write path still uses the original Arrow arrays/schema: for example `extract_datum_from_arrow` only downcasts binary Paimon fields to `arrow_array::BinaryArray`, and append writes would also see the incoming schema rather than the table schema. So a batch that this check accepts can later fail with a type-mismatch (or write files whose Arrow schema differs from the table schema). Please either reject non-exact binary-family types here or normalize/cast the batch to the target Arrow schema before calling `write_arrow_batch`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Expose DataFrame-style batch write + commit loop [paimon-rust]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #420:
URL: https://github.com/apache/paimon-rust/pull/420#discussion_r3497401925
##########
bindings/python/tests/test_write.py:
##########
@@ -0,0 +1,165 @@
+# 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.
+
+import tempfile
+
+import pyarrow as pa
+import pytest
+
+from pypaimon_rust.datafusion import PaimonCatalog, SQLContext
+
+# The table created by _make_empty_table is (id INT, name STRING). Paimon INT maps
+# to Arrow int32, so batches must use int32 for id — pyarrow infers Python ints as
+# int64, which write_arrow now (correctly, matching pypaimon) rejects as a type
+# mismatch. Build batches against this explicit schema to match the table.
+_TABLE_SCHEMA = pa.schema([("id", pa.int32()), ("name", pa.string())])
+
+
+def _batch(ids, names):
+ return pa.record_batch([ids, names], schema=_TABLE_SCHEMA)
+
+
+def _make_empty_table(warehouse):
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.wdb")
+ ctx.sql("CREATE TABLE paimon.wdb.t (id INT, name STRING)")
+ return ctx
+
+
+def _get_table(warehouse):
+ return PaimonCatalog({"warehouse": warehouse}).get_table("wdb.t")
+
+
+def test_write_commit_read_roundtrip():
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = _make_empty_table(warehouse)
+ table = _get_table(warehouse)
+ batch = _batch([1, 2, 3], ["a", "b", "c"])
+ wb = table.new_write_builder()
+ write = wb.new_write()
+ write.write_arrow(batch)
+ messages = write.prepare_commit()
+ assert len(messages) >= 1 # cover API shape in the first test
+ wb.new_commit().commit(messages) # same wb → shared commit_user
+ result = pa.Table.from_batches(
+ ctx.sql("SELECT id, name FROM paimon.wdb.t")
+ ).sort_by("id").to_pydict()
+ assert result == {"id": [1, 2, 3], "name": ["a", "b", "c"]}
+
+
+def test_write_multiple_batches():
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = _make_empty_table(warehouse)
+ table = _get_table(warehouse)
+ wb = table.new_write_builder()
+ write = wb.new_write()
+ write.write_arrow(_batch([1], ["a"]))
+ write.write_arrow(_batch([2], ["b"]))
+ messages = write.prepare_commit()
+ wb.new_commit().commit(messages)
+ result = pa.Table.from_batches(
+ ctx.sql("SELECT id, name FROM paimon.wdb.t")
+ ).sort_by("id").to_pydict()
+ assert result == {"id": [1, 2], "name": ["a", "b"]}
+
+
+def test_prepare_commit_returns_messages():
+ with tempfile.TemporaryDirectory() as warehouse:
+ _make_empty_table(warehouse)
+ table = _get_table(warehouse)
+ write = table.new_write_builder().new_write()
+ write.write_arrow(_batch([1], ["a"]))
+ messages = write.prepare_commit()
+ assert len(messages) >= 1
+ assert all(type(m).__name__ == "CommitMessage" for m in messages)
+
+
+def test_commit_empty_messages_noop():
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = _make_empty_table(warehouse)
+ table = _get_table(warehouse)
+ wb = table.new_write_builder()
+ messages = wb.new_write().prepare_commit() # no write
+ assert messages == []
+ wb.new_commit().commit(messages) # no-op success
+ batches = ctx.sql("SELECT COUNT(*) AS cnt FROM paimon.wdb.t")
+ assert batches[0].column(0).to_pylist() == [0]
+
+
+def test_write_arrow_type_mismatch_raises():
+ with tempfile.TemporaryDirectory() as warehouse:
+ _make_empty_table(warehouse) # table (id INT, name STRING)
+ table = _get_table(warehouse)
+ write = table.new_write_builder().new_write()
+ bad = pa.record_batch([["x", "y"], ["a", "b"]], names=["id", "name"]) # id as STRING
+ with pytest.raises(ValueError):
+ write.write_arrow(bad)
+
+
+def test_write_arrow_binary_family_mismatch_raises():
+ # A BINARY column requires Arrow `binary`; a near-equivalent `large_binary`
+ # must be rejected at validation (it would otherwise fail deeper, since the
+ # write path downcasts binary fields to arrow_array::BinaryArray only).
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.wdb")
+ ctx.sql("CREATE TABLE paimon.wdb.bt (id INT, data BINARY)")
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("wdb.bt")
+ write = table.new_write_builder().new_write()
+ schema = pa.schema([("id", pa.int32()), ("data", pa.large_binary())])
+ bad = pa.record_batch([[1], [b"x"]], schema=schema)
+ with pytest.raises(ValueError):
+ write.write_arrow(bad)
+
+
+def test_commit_non_message_raises_typeerror():
+ with tempfile.TemporaryDirectory() as warehouse:
+ _make_empty_table(warehouse)
+ table = _get_table(warehouse)
+ with pytest.raises(TypeError):
+ table.new_write_builder().new_commit().commit([object()])
+ # A non-iterable argument also raises TypeError (not a raw PyO3 error).
+ with pytest.raises(TypeError):
+ table.new_write_builder().new_commit().commit(42)
+
+
+def test_commit_cross_table_messages_raises():
+ # Messages prepared for one table must not be committed by another table's
+ # committer (would persist a snapshot referencing data files written
+ # elsewhere). The wrapper stamps each message with its source table location
+ # and the committer rejects mismatches.
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.wdb")
+ ctx.sql("CREATE TABLE paimon.wdb.t1 (id INT, name STRING)")
+ ctx.sql("CREATE TABLE paimon.wdb.t2 (id INT, name STRING)")
+ catalog = PaimonCatalog({"warehouse": warehouse})
+ t1 = catalog.get_table("wdb.t1")
+ t2 = catalog.get_table("wdb.t2")
+ batch = pa.record_batch(
+ [pa.array([1], pa.int32()), pa.array(["a"], pa.string())],
+ names=["id", "name"],
+ )
+ w1 = t1.new_write_builder().new_write()
+ w1.write_arrow(batch)
+ messages = w1.prepare_commit()
+ with pytest.raises(ValueError):
+ t2.new_write_builder().new_commit().commit(messages)
+
Review Comment:
`git diff --check origin/main...HEAD` reports this as a new blank line at EOF. Please remove the extra trailing blank line so the diff-check stays clean.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Expose DataFrame-style batch write + commit loop [paimon-rust]
Posted by "JunRuiLee (via GitHub)" <gi...@apache.org>.
JunRuiLee commented on code in PR #420:
URL: https://github.com/apache/paimon-rust/pull/420#discussion_r3497489071
##########
bindings/python/src/write.rs:
##########
@@ -0,0 +1,203 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::datatypes::Schema as ArrowSchema;
+use arrow::pyarrow::FromPyArrow;
+use arrow::record_batch::RecordBatch;
+use paimon::table::{CommitMessage, Table, TableCommit, TableWrite};
+use paimon_datafusion::runtime::runtime;
+use pyo3::exceptions::{PyTypeError, PyValueError};
+use pyo3::prelude::*;
+
+use crate::error::to_py_err;
+
+/// Validate an incoming batch schema against the table's target Arrow schema:
+/// field count, order, and names must match, and types must match exactly. The
+/// nullable flag is intentionally NOT compared, since `build_target_arrow_schema`
+/// derives nullability from the Paimon field while pyarrow-constructed batches
+/// infer nullable=true. No cast — callers supply correctly-typed batches.
+///
+/// Type matching is strict (no binary-family interchange): the lower write path
+/// downcasts to the exact Arrow array for each Paimon type (e.g. a `Binary` /
+/// `VarBinary` field requires `arrow_array::BinaryArray`, not `LargeBinary` /
+/// `FixedSizeBinary`). Accepting a near-equivalent type here would pass
+/// validation but then fail deeper with a type-mismatch (or write files whose
+/// Arrow schema differs from the table), so it is rejected up front.
+fn validate_batch_schema(input: &ArrowSchema, target: &ArrowSchema) -> PyResult<()> {
+ let mismatch = || {
+ PyValueError::new_err(format!(
+ "Input schema is not consistent with the table schema. \
+ input: {input:?}, table: {target:?}"
+ ))
+ };
+ if input.fields().len() != target.fields().len() {
+ return Err(mismatch());
+ }
+ for (i, t) in input.fields().iter().zip(target.fields().iter()) {
+ if i.name() != t.name() {
+ return Err(mismatch());
+ }
+ if i.data_type() != t.data_type() {
+ return Err(mismatch());
+ }
+ }
+ Ok(())
+}
+
+/// Builder for the batch write loop, created via [`crate::table::PyTable::new_write_builder`].
+///
+/// Holds the owning table plus a single fixed `commit_user`, generated once and
+/// shared by both `new_write()` and `new_commit()` so that writers and the
+/// committer agree on the commit user (Paimon uses it for duplicate-commit
+/// detection). Creating a fresh `WriteBuilder` per call would otherwise mint a
+/// new random UUID each time.
+#[pyclass(name = "WriteBuilder", module = "pypaimon_rust.datafusion")]
+pub struct PyWriteBuilder {
+ table: Arc<Table>,
+ commit_user: String,
+}
+
+impl PyWriteBuilder {
+ pub fn new(table: Arc<Table>) -> Self {
+ let commit_user = table.new_write_builder().commit_user().to_string();
+ Self { table, commit_user }
+ }
+}
+
+#[pymethods]
+impl PyWriteBuilder {
+ /// Create a writer for accumulating Arrow batches.
+ fn new_write(&self) -> PyResult<PyTableWrite> {
+ let builder = self
+ .table
+ .new_write_builder()
+ .with_commit_user(self.commit_user.clone())
+ .map_err(to_py_err)?;
+ let target_schema = paimon::arrow::build_target_arrow_schema(self.table.schema().fields())
+ .map_err(to_py_err)?;
+ Ok(PyTableWrite {
+ inner: builder.new_write().map_err(to_py_err)?,
+ target_schema,
+ table_location: self.table.location().to_string(),
+ })
+ }
+
+ /// Create a committer for persisting prepared commit messages.
+ fn new_commit(&self) -> PyResult<PyTableCommit> {
+ let builder = self
+ .table
+ .new_write_builder()
+ .with_commit_user(self.commit_user.clone())
+ .map_err(to_py_err)?;
+ Ok(PyTableCommit {
+ inner: builder.new_commit(),
+ table_location: self.table.location().to_string(),
+ })
+ }
+}
+
+/// A stateful writer that accumulates Arrow batches until `prepare_commit`.
+///
+/// Marked `unsendable`: the underlying `TableWrite` holds file writers that are
+/// not `Sync`, so the object enforces single-thread access at runtime.
+#[pyclass(name = "TableWrite", module = "pypaimon_rust.datafusion", unsendable)]
+pub struct PyTableWrite {
+ inner: TableWrite,
+ /// The table's target Arrow schema, used to validate incoming batches.
+ target_schema: Arc<ArrowSchema>,
+ /// The owning table's location, stamped onto produced commit messages so a
+ /// committer can reject messages prepared for a different table.
+ table_location: String,
+}
+
+#[pymethods]
+impl PyTableWrite {
+ /// Write a single PyArrow RecordBatch into the table's writers.
+ fn write_arrow(&mut self, py: Python<'_>, batch: &Bound<'_, PyAny>) -> PyResult<()> {
+ let batch = RecordBatch::from_pyarrow_bound(batch)?;
+ validate_batch_schema(&batch.schema(), &self.target_schema)?;
+ let rt = runtime();
+ py.detach(|| rt.block_on(async { self.inner.write_arrow_batch(&batch).await }))
+ .map_err(to_py_err)
+ }
+
+ /// Close writers and return the commit messages (opaque; pass to commit()).
+ fn prepare_commit(&mut self, py: Python<'_>) -> PyResult<Vec<PyCommitMessage>> {
+ let rt = runtime();
+ let messages = py
+ .detach(|| rt.block_on(async { self.inner.prepare_commit().await }))
+ .map_err(to_py_err)?;
+ Ok(messages
+ .into_iter()
+ .map(|inner| PyCommitMessage {
Review Comment:
You're right — the location check alone still let a different WriteBuilder (same table) through, and since each builder mints its own commit_user, the writer and committer would disagree on it. Now PyCommitMessage also carries the originating builder's commit_user, and commit() rejects a mismatch in addition to the table-location check. Added a same-table-different-builder regression test (the legit table.new_write_builder() write+commit flow still passes since it shares one builder). Fixed in [6c1337a](https://github.com/apache/paimon-rust/pull/420/commits/6c1337ac51e245839bfdab234bd1809d480b14b0).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Expose DataFrame-style batch write + commit loop [paimon-rust]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #420:
URL: https://github.com/apache/paimon-rust/pull/420#discussion_r3497399522
##########
bindings/python/src/write.rs:
##########
@@ -0,0 +1,203 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::datatypes::Schema as ArrowSchema;
+use arrow::pyarrow::FromPyArrow;
+use arrow::record_batch::RecordBatch;
+use paimon::table::{CommitMessage, Table, TableCommit, TableWrite};
+use paimon_datafusion::runtime::runtime;
+use pyo3::exceptions::{PyTypeError, PyValueError};
+use pyo3::prelude::*;
+
+use crate::error::to_py_err;
+
+/// Validate an incoming batch schema against the table's target Arrow schema:
+/// field count, order, and names must match, and types must match exactly. The
+/// nullable flag is intentionally NOT compared, since `build_target_arrow_schema`
+/// derives nullability from the Paimon field while pyarrow-constructed batches
+/// infer nullable=true. No cast — callers supply correctly-typed batches.
+///
+/// Type matching is strict (no binary-family interchange): the lower write path
+/// downcasts to the exact Arrow array for each Paimon type (e.g. a `Binary` /
+/// `VarBinary` field requires `arrow_array::BinaryArray`, not `LargeBinary` /
+/// `FixedSizeBinary`). Accepting a near-equivalent type here would pass
+/// validation but then fail deeper with a type-mismatch (or write files whose
+/// Arrow schema differs from the table), so it is rejected up front.
+fn validate_batch_schema(input: &ArrowSchema, target: &ArrowSchema) -> PyResult<()> {
+ let mismatch = || {
+ PyValueError::new_err(format!(
+ "Input schema is not consistent with the table schema. \
+ input: {input:?}, table: {target:?}"
+ ))
+ };
+ if input.fields().len() != target.fields().len() {
+ return Err(mismatch());
+ }
+ for (i, t) in input.fields().iter().zip(target.fields().iter()) {
+ if i.name() != t.name() {
+ return Err(mismatch());
+ }
+ if i.data_type() != t.data_type() {
+ return Err(mismatch());
+ }
+ }
+ Ok(())
+}
+
+/// Builder for the batch write loop, created via [`crate::table::PyTable::new_write_builder`].
+///
+/// Holds the owning table plus a single fixed `commit_user`, generated once and
+/// shared by both `new_write()` and `new_commit()` so that writers and the
+/// committer agree on the commit user (Paimon uses it for duplicate-commit
+/// detection). Creating a fresh `WriteBuilder` per call would otherwise mint a
+/// new random UUID each time.
+#[pyclass(name = "WriteBuilder", module = "pypaimon_rust.datafusion")]
+pub struct PyWriteBuilder {
+ table: Arc<Table>,
+ commit_user: String,
+}
+
+impl PyWriteBuilder {
+ pub fn new(table: Arc<Table>) -> Self {
+ let commit_user = table.new_write_builder().commit_user().to_string();
+ Self { table, commit_user }
+ }
+}
+
+#[pymethods]
+impl PyWriteBuilder {
+ /// Create a writer for accumulating Arrow batches.
+ fn new_write(&self) -> PyResult<PyTableWrite> {
+ let builder = self
+ .table
+ .new_write_builder()
+ .with_commit_user(self.commit_user.clone())
+ .map_err(to_py_err)?;
+ let target_schema = paimon::arrow::build_target_arrow_schema(self.table.schema().fields())
+ .map_err(to_py_err)?;
+ Ok(PyTableWrite {
+ inner: builder.new_write().map_err(to_py_err)?,
+ target_schema,
+ table_location: self.table.location().to_string(),
+ })
+ }
+
+ /// Create a committer for persisting prepared commit messages.
+ fn new_commit(&self) -> PyResult<PyTableCommit> {
+ let builder = self
+ .table
+ .new_write_builder()
+ .with_commit_user(self.commit_user.clone())
+ .map_err(to_py_err)?;
+ Ok(PyTableCommit {
+ inner: builder.new_commit(),
+ table_location: self.table.location().to_string(),
+ })
+ }
+}
+
+/// A stateful writer that accumulates Arrow batches until `prepare_commit`.
+///
+/// Marked `unsendable`: the underlying `TableWrite` holds file writers that are
+/// not `Sync`, so the object enforces single-thread access at runtime.
+#[pyclass(name = "TableWrite", module = "pypaimon_rust.datafusion", unsendable)]
+pub struct PyTableWrite {
+ inner: TableWrite,
+ /// The table's target Arrow schema, used to validate incoming batches.
+ target_schema: Arc<ArrowSchema>,
+ /// The owning table's location, stamped onto produced commit messages so a
+ /// committer can reject messages prepared for a different table.
+ table_location: String,
+}
+
+#[pymethods]
+impl PyTableWrite {
+ /// Write a single PyArrow RecordBatch into the table's writers.
+ fn write_arrow(&mut self, py: Python<'_>, batch: &Bound<'_, PyAny>) -> PyResult<()> {
+ let batch = RecordBatch::from_pyarrow_bound(batch)?;
+ validate_batch_schema(&batch.schema(), &self.target_schema)?;
+ let rt = runtime();
+ py.detach(|| rt.block_on(async { self.inner.write_arrow_batch(&batch).await }))
+ .map_err(to_py_err)
+ }
+
+ /// Close writers and return the commit messages (opaque; pass to commit()).
+ fn prepare_commit(&mut self, py: Python<'_>) -> PyResult<Vec<PyCommitMessage>> {
+ let rt = runtime();
+ let messages = py
+ .detach(|| rt.block_on(async { self.inner.prepare_commit().await }))
+ .map_err(to_py_err)?;
+ Ok(messages
+ .into_iter()
+ .map(|inner| PyCommitMessage {
Review Comment:
This still lets commit messages from a different `WriteBuilder` for the same table pass the guard. Each `PyWriteBuilder` generates its own `commit_user`, and the Rust builder contract is that writers and committers created from that builder share it; it is used for snapshot duplicate detection and, for postpone-bucket tables, embedded in data file names. A natural Python flow like `table.new_write_builder().new_write()` followed by `table.new_write_builder().new_commit().commit(messages)` now succeeds but commits the files with a different committer `commit_user` than the writer used. Please stamp the originating `commit_user` onto `PyCommitMessage` as well and reject messages whose `commit_user` does not match the committer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Expose DataFrame-style batch write + commit loop [paimon-rust]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #420:
URL: https://github.com/apache/paimon-rust/pull/420
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org