This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 97dbf29d8f [python] Add overwrite API to multimodal table (#8441)
97dbf29d8f is described below
commit 97dbf29d8f56b903e90b3e52e6d85f6fc2994752
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Jul 3 10:55:26 2026 +0800
[python] Add overwrite API to multimodal table (#8441)
Add a high-level `MultimodalTable.overwrite` API so multimodal users can
replace table data without dropping down to the lower-level write
builder. The API reuses Paimon batch overwrite semantics and documents
how whole-table, dynamic partition, and static partition overwrite
behave.
---
docs/docs/pypaimon/multimodal-api.mdx | 36 +++++++
paimon-python/pypaimon/multimodal/table.py | 17 ++++
.../pypaimon/tests/multimodal_table_test.py | 105 +++++++++++++++++++++
3 files changed, 158 insertions(+)
diff --git a/docs/docs/pypaimon/multimodal-api.mdx
b/docs/docs/pypaimon/multimodal-api.mdx
index b061aa36c0..83d13ae345 100644
--- a/docs/docs/pypaimon/multimodal-api.mdx
+++ b/docs/docs/pypaimon/multimodal-api.mdx
@@ -190,6 +190,42 @@ docs.add([
])
```
+## Overwrite
+
+`overwrite` accepts the same input formats as `add` and replaces existing data
+using Paimon's batch overwrite semantics. On an unpartitioned table it replaces
+the whole table. On a partitioned table it follows the table's
+`dynamic-partition-overwrite` option; with the default dynamic mode, only
+partitions present in the input data are replaced.
+
+```python
+docs.overwrite([
+ {
+ "id": 3,
+ "content": "Fresh replacement text",
+ "embedding": [0.7, 0.8, 0.9],
+ "category": "docs",
+ }
+])
+```
+
+For static partition overwrite, disable dynamic partition overwrite on the
+table and pass the target partition:
+
+```python
+docs.overwrite(
+ [
+ {
+ "id": 4,
+ "content": "Only this day is replaced.",
+ "category": "docs",
+ "dt": "2024-01-01",
+ }
+ ],
+ partition={"dt": "2024-01-01"},
+)
+```
+
## Update
`update` modifies rows matched by a SQL-like predicate.
diff --git a/paimon-python/pypaimon/multimodal/table.py
b/paimon-python/pypaimon/multimodal/table.py
index 916938137c..12de790060 100644
--- a/paimon-python/pypaimon/multimodal/table.py
+++ b/paimon-python/pypaimon/multimodal/table.py
@@ -109,6 +109,23 @@ class MultimodalTable:
table_commit.close()
return self
+ def overwrite(self, data, partition: Optional[Mapping[str, object]] =
None):
+ arrow_table = _to_arrow_table(data, _target_schema(self.raw_table))
+ overwrite_partition = dict(partition) if partition is not None else
None
+ write_builder = (
+ self.raw_table.new_batch_write_builder()
+ .overwrite(overwrite_partition)
+ )
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ try:
+ table_write.write_arrow(arrow_table)
+ table_commit.commit(table_write.prepare_commit())
+ finally:
+ table_write.close()
+ table_commit.close()
+ return self
+
def update(self, where, values):
query = self.scan().where(where)
predicate = query._predicate
diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py
b/paimon-python/pypaimon/tests/multimodal_table_test.py
index e745932c21..14c11e0040 100644
--- a/paimon-python/pypaimon/tests/multimodal_table_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_table_test.py
@@ -513,6 +513,111 @@ class MultimodalTableTest(unittest.TestCase):
self.assertEqual(1, result.num_rows)
self.assertEqual([1], result["id"].to_pylist())
+ def test_overwrite_replaces_unpartitioned_table(self):
+ users = self.conn.create_table(
+ "users",
+ data=[
+ {"id": 1, "name": "Alice", "age": 30},
+ {"id": 2, "name": "Bob", "age": 25},
+ ],
+ schema=_schema({
+ "id": pa.int32(),
+ "name": pa.string(),
+ "age": pa.int32(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+
+ result = users.overwrite([
+ {"id": 3, "name": "Carol", "age": 40},
+ ])
+
+ self.assertIs(users, result)
+ self.assertEqual(
+ [
+ {"id": 3, "name": "Carol", "age": 40},
+ ],
+ users.scan().to_list(),
+ )
+
+ def test_empty_overwrite_clears_unpartitioned_table(self):
+ users = self.conn.create_table(
+ "users",
+ data=[
+ {"id": 1, "name": "Alice"},
+ {"id": 2, "name": "Bob"},
+ ],
+ schema=_schema({
+ "id": pa.int32(),
+ "name": pa.string(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+
+ users.overwrite([])
+
+ self.assertEqual([], users.scan().to_list())
+
+ def test_overwrite_replaces_dynamic_partitions(self):
+ users = self.conn.create_table(
+ "users",
+ data=[
+ {"id": 1, "name": "Alice", "dt": "2024-01-01"},
+ {"id": 2, "name": "Bob", "dt": "2024-01-02"},
+ ],
+ schema=_schema({
+ "id": pa.int32(),
+ "name": pa.string(),
+ "dt": pa.string(),
+ }),
+ options=_PARQUET_OPTIONS,
+ partitioned=["dt"],
+ )
+
+ users.overwrite([
+ {"id": 3, "name": "Carol", "dt": "2024-01-01"},
+ ])
+
+ rows = sorted(users.scan().to_list(), key=lambda r: r["id"])
+ self.assertEqual(
+ [
+ {"id": 2, "name": "Bob", "dt": "2024-01-02"},
+ {"id": 3, "name": "Carol", "dt": "2024-01-01"},
+ ],
+ rows,
+ )
+
+ def test_overwrite_replaces_static_partition(self):
+ users = self.conn.create_table(
+ "users",
+ data=[
+ {"id": 1, "name": "Alice", "dt": "2024-01-01"},
+ {"id": 2, "name": "Bob", "dt": "2024-01-02"},
+ ],
+ schema=_schema({
+ "id": pa.int32(),
+ "name": pa.string(),
+ "dt": pa.string(),
+ }),
+ options=dict(_PARQUET_OPTIONS, **{
+ "dynamic-partition-overwrite": "false",
+ }),
+ partitioned=["dt"],
+ )
+
+ users.overwrite([
+ {"id": 3, "name": "Carol", "dt": "2024-01-01"},
+ ], partition={"dt": "2024-01-01"})
+
+ rows = sorted(users.scan().to_list(), key=lambda r: r["id"])
+ self.assertEqual(
+ [
+ {"id": 2, "name": "Bob", "dt": "2024-01-02"},
+ {"id": 3, "name": "Carol", "dt": "2024-01-01"},
+ ],
+ rows,
+ )
+
def test_scan_does_not_expose_pre_filter(self):
users = self.conn.create_table(
"users",
