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 fc99df74c1 [python] Add predicate update for data evolution (#8413)
fc99df74c1 is described below
commit fc99df74c1c34d77362bf19b19302c6e53f9af80
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jul 1 17:36:07 2026 +0800
[python] Add predicate update for data evolution (#8413)
Adds a PyPaimon predicate-based data evolution update API for SQL-like
`UPDATE ... SET ... WHERE ...` flows. The row-id discovery scan is
aligned with #8411 by resolving point-in-time scan options through
`TimeTravelUtil`, pinning the resolved snapshot, and forcing
`global-index.search-mode=full` so matched updates include rows appended
after a global index was created.
---
docs/docs/multimodal-table/data-evolution.md | 172 ----------
docs/docs/multimodal-table/data-evolution.mdx | 307 +++++++++++++++++
docs/docs/pypaimon/data-evolution.md | 168 ++++++++--
paimon-python/pypaimon/tests/table_update_test.py | 386 ++++++++++++++++++++++
paimon-python/pypaimon/write/table_update.py | 166 +++++++++-
5 files changed, 1004 insertions(+), 195 deletions(-)
diff --git a/docs/docs/multimodal-table/data-evolution.md
b/docs/docs/multimodal-table/data-evolution.md
deleted file mode 100644
index fe71e78415..0000000000
--- a/docs/docs/multimodal-table/data-evolution.md
+++ /dev/null
@@ -1,172 +0,0 @@
----
-title: "Data Evolution"
-sidebar_position: 6
----
-
-<!--
-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.
--->
-
-# Data Evolution
-
-## Overview
-
-Paimon supports complete Schema Evolution, allowing you to freely add, modify,
or delete column schema. But how to
-backfill newly added columns or update column data.
-
-Data Evolution Mode is a new feature for Append tables that revolutionizes how
you handle data evolution,
-particularly when adding new columns. This mode allows you to update partial
columns without rewriting entire data
-files. Instead, it writes new column data to separate files and intelligently
merges them with the original data
-during read operations.
-
-The data evolution mode offers significant advantages for your data lake
architecture:
-
-* Efficient Partial Column Updates: With this mode, you can use Spark's MERGE
INTO statement to update a subset of columns. This avoids the high I/O cost of
rewriting the whole file, as only the updated columns are written.
-
-* Reduced File Rewrites: In scenarios with frequent schema changes, such as
adding new columns, the traditional method requires constant file rewriting.
Data evolution mode eliminates this overhead by appending new column data to
dedicated files. This approach is much more efficient and reduces the burden on
your storage system.
-
-* Optimized Read Performance: The new mode is designed for seamless data
retrieval. During query execution, Paimon's engine efficiently combines the
original data with the new column data, ensuring that read performance remains
uncompromised. The merge process is highly optimized, so your queries run just
as fast as they would on a single, consolidated file.
-
-To enable data evolution, you must enable row-tracking and set the
`row-tracking.enabled` and `data-evolution.enabled` property to `true` when
creating an append table. This ensures that the table is ready for efficient
schema evolution operations.
-
-Use Spark Sql as an example:
-
-```sql
-CREATE TABLE target_table (id INT, b INT, c INT) TBLPROPERTIES (
- 'row-tracking.enabled' = 'true',
- 'data-evolution.enabled' = 'true'
-);
-
-INSERT INTO target_table VALUES (1, 1, 1), (2, 2, 2);
-```
-
-Now we could update partial columns by spark 'MERGE INTO' statement or flink
'data_evolution_merge_into' procedure:
-
-### Spark
-
-```sql
-CREATE TABLE source_table (id INT, b INT);
-INSERT INTO source_table VALUES (1, 11), (2, 22), (3, 33);
-
-MERGE INTO target_table AS t
-USING source_table AS s
-ON t.id = s.id
-WHEN MATCHED THEN UPDATE SET t.b = s.b
-WHEN NOT MATCHED THEN INSERT (id, b, c) VALUES (id, b, 0);
-
-SELECT * FROM target_table;
-+----+----+----+
-| id | b | c |
-+----+----+----+
-| 1 | 11 | 1 |
-| 2 | 22 | 2 |
-| 3 | 33 | 0 |
-```
-
-This statement updates only the `b` column in the target table `target_table`
based on the matching records from the source table
-`source_table`. The `id` column and `c` column remain unchanged, and new
records are inserted with the specified values. The difference between this and
table those are not enabled with data evolution is that only the `b` column
data is written to new files.
-
-Note that:
-* Data Evolution Table does not support 'Delete' and 'Update' statement yet.
-* Merge Into for Data Evolution Table does not support 'WHEN NOT MATCHED BY
SOURCE' clause.
-
-### Flink
-Since Flink does not currently support the MERGE INTO syntax, we simulate the
merge-into process using the data_evolution_merge_into procedure, as shown
below:
-
-```sql
-CREATE TABLE source_table (id INT, b INT);
-INSERT INTO source_table VALUES (1, 11), (2, 22), (3, 33);
-
-CALL sys.data_evolution_merge_into(
- 'my_db.target_table',
- '', /* Optional target alias */
- '', /* Optional source sqls */
- 'source_table',
- 'source_table.id=target_table.id',
- 'b=source_table.b',
- 2 /* Specify sink parallelism */
-);
-
-SELECT * FROM source_table
-+----+----+----+
-| id | b | c |
-+----+----+----+
-| 1 | 11 | 1 |
-| 2 | 22 | 2 |
-```
-Note that:
-* Compared to Spark implementation, Flink data_evolution_merge_into procedure
only supports updating/inserting new columns now. Inserting new rows is not
supported yet.
-
-#### Self Merge
-
-Self-merge refers to the case where the source and target of the merge
operation are the **same table**. This is useful
-when you want to transform existing column values in place — for example,
applying a UDF to rewrite a column.
-
-Since the source table cannot be the same as the target table directly, you
need to create a **temporary view** based on
-the system table `T$row_tracking` (which exposes the hidden `_ROW_ID` column)
and use `_ROW_ID` as the merge condition.
-
-```sql
--- 1. Register a UDF
-CREATE TEMPORARY FUNCTION concat_string AS 'com.example.StringConcatUdf';
-
--- 2. Create a view from the row-tracking system table
-CREATE TEMPORARY VIEW source_view AS
-SELECT _ROW_ID, concat_string(name) AS name
-FROM my_db.target_table$row_tracking;
-
--- 3. Self-merge: update the name column using the UDF result
-CALL sys.data_evolution_merge_into(
- 'my_db.target_table',
- 'TempT',
- -- alternatively, you could also pass the create sqls in procedure directly
- -- like: 'CREATE TEMPORARY FUNCTION concat_string AS
''com.example.StringConcatUdf''; CREATE TEMPORARY VIEW XXX'
- '',
- 'source_view',
- 'TempT._ROW_ID=source_view._ROW_ID',
- 'name=source_view.name',
- 2
-);
-```
-
-Note that:
-* The source and target table name cannot be the same. You must create a
temporary view as the source.
-* use `view._ROW_ID` = `source._ROW_ID` to identify the self-merge pattern.
-* `_ROW_ID` is only available via the `$row_tracking` system table.
-* Self-merge only supports `WHEN MATCHED THEN UPDATE` semantics.
-
-## File Group Spec
-
-Through the RowId metadata, files are organized into a file group.
-
-When writing: MERGE INTO clause for Data Evolution Table only updates the
specified columns, and writes the updated column data to new files. The
original data files remain unchanged.
-
-When reading: Paimon reads both the original data files and the new files
containing the updated column data. It then merges the data from these two
sources to present a unified view of the table. This merging process is
optimized to ensure that read performance is not significantly impacted.
-
-After writing, the files in `target_table` like below:
-
-
-
-When reading, the files with the same `first row id` will merge fields.
-
-
-
-The advantage to the mode is:
-
-* Avoid rewriting the whole file when updating partial columns, reducing I/O
cost.
-* The read performance is not significantly impacted, as the merge process is
optimized.
-* The disk space is used more efficiently, as only the updated columns are
written to new files.
diff --git a/docs/docs/multimodal-table/data-evolution.mdx
b/docs/docs/multimodal-table/data-evolution.mdx
new file mode 100644
index 0000000000..abafa67240
--- /dev/null
+++ b/docs/docs/multimodal-table/data-evolution.mdx
@@ -0,0 +1,307 @@
+---
+title: "Data Evolution"
+sidebar_position: 6
+---
+
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+<!--
+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.
+-->
+
+# Data Evolution
+
+## Overview
+
+Paimon supports schema evolution, allowing you to add, modify, or delete column
+schema. Data Evolution mode extends this for append tables by allowing partial
+column updates without rewriting entire data files. Updated column data is
+written to new files and merged with the original data during reads.
+
+Data Evolution mode offers the following advantages:
+
+- **Efficient partial-column updates**: update a subset of columns and avoid
the
+ I/O cost of rewriting untouched columns.
+- **Reduced file rewrites**: append new column data to dedicated files when
+ backfilling or evolving data.
+- **Optimized reads**: merge original and updated column files at read time.
+
+To enable Data Evolution, create an append table with both
+`row-tracking.enabled` and `data-evolution.enabled` set to `true`.
+
+<Tabs groupId="data-evolution-create-table">
+
+<TabItem value="spark-sql" label="Spark SQL">
+
+```sql
+CREATE TABLE target_table (id INT, b INT, c INT) TBLPROPERTIES (
+ 'row-tracking.enabled' = 'true',
+ 'data-evolution.enabled' = 'true'
+);
+
+INSERT INTO target_table VALUES (1, 1, 1), (2, 2, 2);
+```
+
+</TabItem>
+
+<TabItem value="flink-sql" label="Flink SQL">
+
+```sql
+CREATE TABLE target_table (id INT, b INT, c INT) WITH (
+ 'row-tracking.enabled' = 'true',
+ 'data-evolution.enabled' = 'true'
+);
+
+INSERT INTO target_table VALUES (1, 1, 1), (2, 2, 2);
+```
+
+</TabItem>
+
+<TabItem value="python-api" label="Python API">
+
+```python
+import pyarrow as pa
+
+from pypaimon import CatalogFactory, Schema
+
+catalog_options = {"warehouse": "/path/to/warehouse"}
+catalog = CatalogFactory.create(catalog_options)
+catalog.create_database("default", True)
+
+pa_schema = pa.schema([
+ ("id", pa.int32()),
+ ("b", pa.int32()),
+ ("c", pa.int32()),
+])
+schema = Schema.from_pyarrow_schema(
+ pa_schema,
+ options={
+ "row-tracking.enabled": "true",
+ "data-evolution.enabled": "true",
+ },
+)
+catalog.create_table("default.target_table", schema, False)
+
+table = catalog.get_table("default.target_table")
+builder = table.new_batch_write_builder()
+write = builder.new_write()
+commit = builder.new_commit()
+write.write_arrow(pa.Table.from_pydict(
+ {"id": [1, 2], "b": [1, 2], "c": [1, 2]},
+ schema=pa_schema,
+))
+commit.commit(write.prepare_commit())
+write.close()
+commit.close()
+```
+
+</TabItem>
+
+</Tabs>
+
+## Partial Updates
+
+You can update selected columns with Spark `MERGE INTO`, the Flink
+`data_evolution_merge_into` procedure, or the PyPaimon table update API. Only
+the updated column files are written; untouched columns remain in their
original
+files.
+
+<Tabs groupId="data-evolution-partial-update">
+
+<TabItem value="spark-sql" label="Spark SQL">
+
+```sql
+CREATE TABLE source_table (id INT, b INT);
+INSERT INTO source_table VALUES (1, 11), (2, 22);
+
+MERGE INTO target_table AS t
+USING source_table AS s
+ON t.id = s.id
+WHEN MATCHED THEN UPDATE SET t.b = s.b;
+
+SELECT * FROM target_table;
++----+----+----+
+| id | b | c |
++----+----+----+
+| 1 | 11 | 1 |
+| 2 | 22 | 2 |
++----+----+----+
+```
+
+</TabItem>
+
+<TabItem value="flink-sql" label="Flink SQL">
+
+Flink does not currently support `MERGE INTO` syntax. Use the
+`data_evolution_merge_into` procedure instead:
+
+```sql
+CREATE TABLE source_table (id INT, b INT);
+INSERT INTO source_table VALUES (1, 11), (2, 22);
+
+CALL sys.data_evolution_merge_into(
+ 'default.target_table',
+ '',
+ '',
+ 'source_table',
+ 'source_table.id=target_table.id',
+ 'b=source_table.b',
+ 2
+);
+
+SELECT * FROM target_table;
++----+----+----+
+| id | b | c |
++----+----+----+
+| 1 | 11 | 1 |
+| 2 | 22 | 2 |
++----+----+----+
+```
+
+</TabItem>
+
+<TabItem value="python-api" label="Python API">
+
+PyPaimon exposes an UPDATE-like table update API. It accepts a `Predicate`
+for the `WHERE` condition and literal assignments for the `SET` clause.
+
+```python
+table = catalog.get_table("default.target_table")
+
+builder = table.new_batch_write_builder()
+table_update = builder.new_update()
+predicate = table_update.new_predicate_builder().is_in("id", [1, 2])
+
+messages = table_update.update_by_predicate(predicate, {"b": 100})
+
+commit = builder.new_commit()
+commit.commit(messages)
+commit.close()
+```
+
+</TabItem>
+
+</Tabs>
+
+Notes:
+
+- SQL `DELETE` and standalone SQL `UPDATE` statements are not supported for
+ Data Evolution tables yet. Use Spark `MERGE INTO`, the Flink procedure, or
+ PyPaimon APIs.
+- `MERGE INTO` for Data Evolution tables does not support the
+ `WHEN NOT MATCHED BY SOURCE` clause.
+- The Flink `data_evolution_merge_into` procedure currently supports updating
+ or inserting columns, but not inserting new rows.
+
+## Self Updates
+
+Self updates transform existing column values in place. In Flink SQL, create a
+temporary source view from the `$row_tracking` system table and join by
+`_ROW_ID`. In PyPaimon, use the shard scan + rewrite workflow to read existing
+values and write the derived columns back.
+
+<Tabs groupId="data-evolution-self-update">
+
+<TabItem value="flink-sql" label="Flink SQL">
+
+```sql
+CREATE TEMPORARY VIEW source_view AS
+SELECT _ROW_ID, b + c AS b
+FROM default.target_table$row_tracking;
+
+CALL sys.data_evolution_merge_into(
+ 'default.target_table',
+ 'TempT',
+ '',
+ 'source_view',
+ 'TempT._ROW_ID=source_view._ROW_ID',
+ 'b=source_view.b',
+ 2
+);
+```
+
+</TabItem>
+
+<TabItem value="python-api" label="Python API">
+
+```python
+import pyarrow as pa
+
+table = catalog.get_table("default.target_table")
+
+builder = table.new_batch_write_builder()
+table_update = builder.new_update()
+table_update.with_read_projection(["b", "c"])
+table_update.with_update_type(["b"])
+
+updater = table_update.new_shard_updator(0, 1)
+reader = updater.arrow_reader()
+for batch in iter(reader.read_next_batch, None):
+ b = batch.column("b").to_pylist()
+ c = batch.column("c").to_pylist()
+ updater.update_by_arrow_batch(pa.RecordBatch.from_pydict(
+ {"b": [bi + ci for bi, ci in zip(b, c)]},
+ schema=pa.schema([("b", pa.int32())]),
+ ))
+
+messages = updater.prepare_commit()
+commit = builder.new_commit()
+commit.commit(messages)
+commit.close()
+```
+
+</TabItem>
+
+</Tabs>
+
+Self-update notes:
+
+- The source and target table name cannot be the same in the Flink procedure.
+ Create a temporary view as the source.
+- Use `view._ROW_ID = source._ROW_ID` to identify the self-merge pattern in
+ Flink.
+- `_ROW_ID` is only available via the `$row_tracking` system table in SQL.
+- Self-merge only supports `WHEN MATCHED THEN UPDATE` semantics.
+
+## File Group Spec
+
+Through the row-id metadata, files are organized into file groups.
+
+When writing, the Data Evolution update path writes only the specified updated
+columns to new files. The original data files remain unchanged.
+
+When reading, Paimon reads both the original data files and the new files
+containing updated column data, then merges files with the same `first row id`
+to present a unified view of the table.
+
+After writing, files in `target_table` are organized as below:
+
+
+
+When reading, files with the same `first row id` are merged:
+
+
+
+The advantages of this mode are:
+
+- Avoid rewriting the whole file when updating partial columns, reducing I/O
+ cost.
+- Keep read performance efficient through optimized merge processing.
+- Use disk space more efficiently because only updated columns are written to
+ new files.
diff --git a/docs/docs/pypaimon/data-evolution.md
b/docs/docs/pypaimon/data-evolution.md
index 2ce89d2314..1ec5a3a0a2 100644
--- a/docs/docs/pypaimon/data-evolution.md
+++ b/docs/docs/pypaimon/data-evolution.md
@@ -3,6 +3,8 @@ title: "Data Evolution"
sidebar_position: 5
---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
<!--
Licensed to the Apache Software Foundation (ASF) under one
@@ -36,25 +38,43 @@ To use partial updates / data evolution, enable both
options when creating the t
## Batch vs Stream
-Data evolution supports both batch and stream modes. The API differs as
follows:
+Data evolution supports both batch and stream modes.
-| | Batch | Stream
|
-|---------------------|----------------------------------------------|------------------------------------------------|
-| Builder | `table.new_batch_write_builder()` |
`table.new_stream_write_builder()` |
-| Write | `BatchTableWrite` |
`StreamTableWrite` |
-| Update | `BatchTableUpdate` |
`StreamTableUpdate` |
-| Commit | `BatchTableCommit` |
`StreamTableCommit` |
-| `commit_identifier` | Not required |
Required (monotonically increasing integer) |
-| Lifecycle | One-shot: each instance can commit only once |
Reusable: same instance can commit many rounds |
+<Tabs groupId="pypaimon-data-evolution-mode">
-Method signatures that differ between modes:
+<TabItem value="batch" label="Batch">
-| Method | Batch
| Stream |
-|---------------------------------|------------------------------------------------|-------------------------------------------------------------------|
-| `prepare_commit()` | `write.prepare_commit()`
| `write.prepare_commit(commit_identifier)` |
-| `update_by_arrow_with_row_id()` |
`update.update_by_arrow_with_row_id(table)` |
`update.update_by_arrow_with_row_id(table, commit_identifier)` |
-| `upsert_by_arrow_with_key()` | `update.upsert_by_arrow_with_key(table,
keys)` | `update.upsert_by_arrow_with_key(table, keys, commit_identifier)` |
-| `commit()` | `commit.commit(messages)`
| `commit.commit(messages, commit_identifier)` |
+- Builder: `table.new_batch_write_builder()`
+- Write: `BatchTableWrite`
+- Update: `BatchTableUpdate`
+- Commit: `BatchTableCommit`
+- `commit_identifier`: not required
+- Lifecycle: one-shot; each instance can commit only once
+- `prepare_commit()`: `write.prepare_commit()`
+- `update_by_arrow_with_row_id()`: `update.update_by_arrow_with_row_id(table)`
+- `update_by_predicate()`: `update.update_by_predicate(predicate, assignments)`
+- `upsert_by_arrow_with_key()`: `update.upsert_by_arrow_with_key(table, keys)`
+- `commit()`: `commit.commit(messages)`
+
+</TabItem>
+
+<TabItem value="stream" label="Stream">
+
+- Builder: `table.new_stream_write_builder()`
+- Write: `StreamTableWrite`
+- Update: `StreamTableUpdate`
+- Commit: `StreamTableCommit`
+- `commit_identifier`: required; use a monotonically increasing integer
+- Lifecycle: reusable; the same instance can commit many rounds
+- `prepare_commit()`: `write.prepare_commit(commit_identifier)`
+- `update_by_arrow_with_row_id()`: `update.update_by_arrow_with_row_id(table,
commit_identifier)`
+- `update_by_predicate()`: `update.update_by_predicate(predicate, assignments,
commit_identifier)`
+- `upsert_by_arrow_with_key()`: `update.upsert_by_arrow_with_key(table, keys,
commit_identifier)`
+- `commit()`: `commit.commit(messages, commit_identifier)`
+
+</TabItem>
+
+</Tabs>
## Update Columns By Row ID
@@ -67,7 +87,9 @@ to its corresponding `first_row_id`, then group rows with the
same `first_row_id
- **Update columns only**: include `_ROW_ID` plus the columns you want to
update (partial schema is OK).
-### Batch Mode
+<Tabs groupId="pypaimon-data-evolution-mode">
+
+<TabItem value="batch" label="Batch">
```python
import pyarrow as pa
@@ -118,7 +140,9 @@ table_commit.close()
# 'f1': [-1001, 1002]
```
-### Stream Mode
+</TabItem>
+
+<TabItem value="stream" label="Stream">
```python
import pyarrow as pa
@@ -170,6 +194,90 @@ table_commit.commit(cmts2, commit_identifier=2)
table_commit.close()
```
+</TabItem>
+
+</Tabs>
+
+## Update Columns By Predicate
+
+You can use `update_by_predicate` for SQL-like `UPDATE ... SET ... WHERE ...`
+operations. The `Predicate` identifies rows to update, and the assignment map
+contains literal values for updated columns.
+When global indexes are available, `update_by_predicate` discovers matching
+`_ROW_ID` values with `global-index.search-mode=full` on the configured
+point-in-time scan snapshot or, if none is configured, the latest snapshot.
+
+<Tabs groupId="pypaimon-data-evolution-mode">
+
+<TabItem value="batch" label="Batch">
+
+```python
+import pyarrow as pa
+from pypaimon import CatalogFactory, Schema
+
+catalog = CatalogFactory.create({'warehouse': '/tmp/warehouse'})
+catalog.create_database('default', False)
+
+pa_schema = pa.schema([
+ ('id', pa.int32()),
+ ('name', pa.string()),
+ ('age', pa.int32()),
+])
+schema = Schema.from_pyarrow_schema(
+ pa_schema,
+ options={'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true'},
+)
+catalog.create_table('default.users_update', schema, False)
+table = catalog.get_table('default.users_update')
+
+# write initial data
+write_builder = table.new_batch_write_builder()
+write = write_builder.new_write()
+commit = write_builder.new_commit()
+write.write_arrow(pa.Table.from_pydict(
+ {'id': [1, 2, 3], 'name': ['Alice', 'Bob', 'Charlie'], 'age': [30, 25,
28]},
+ schema=pa_schema,
+))
+commit.commit(write.prepare_commit())
+write.close()
+commit.close()
+
+# UPDATE users_update SET age = 99 WHERE id IN (1, 3)
+write_builder = table.new_batch_write_builder()
+table_update = write_builder.new_update()
+predicate = table_update.new_predicate_builder().is_in('id', [1, 3])
+messages = table_update.update_by_predicate(predicate, {'age': 99})
+
+commit = write_builder.new_commit()
+commit.commit(messages)
+commit.close()
+```
+
+</TabItem>
+
+<TabItem value="stream" label="Stream">
+
+Pass the `commit_identifier` to both `update_by_predicate` and `commit`:
+
+```python
+stream_builder = table.new_stream_write_builder()
+table_update = stream_builder.new_update()
+table_commit = stream_builder.new_commit()
+
+predicate = table_update.new_predicate_builder().equal('id', 2)
+messages = table_update.update_by_predicate(
+ predicate,
+ {'name': 'Bob_v2'},
+ commit_identifier=1,
+)
+table_commit.commit(messages, commit_identifier=1)
+table_commit.close()
+```
+
+</TabItem>
+
+</Tabs>
+
## Filter by _ROW_ID
Requires the same [Prerequisites](#prerequisites) (row-tracking and
data-evolution enabled). On such tables you can filter by `_ROW_ID` to prune
files at scan time. Supported: `equal('_ROW_ID', id)`, `is_in('_ROW_ID', [id1,
...])`, `between('_ROW_ID', low, high)`.
@@ -196,7 +304,9 @@ If you want to **upsert** (update-or-insert) rows by one or
more business key co
**automatically stripped** from `upsert_keys` during matching (since each
partition is processed independently),
so you do **not** need to include them in `upsert_keys`.
-### Batch Mode
+<Tabs groupId="pypaimon-data-evolution-mode">
+
+<TabItem value="batch" label="Batch">
**Example: basic upsert**
@@ -310,7 +420,9 @@ table_commit.close()
- Duplicate keys in the input data are automatically deduplicated — the **last
occurrence** is kept.
- The upsert is atomic per commit — all matched updates and new appends are
included in the same commit.
-### Stream Mode
+</TabItem>
+
+<TabItem value="stream" label="Stream">
```python
import pyarrow as pa
@@ -367,6 +479,10 @@ table_commit.close()
The `with_update_type` and partitioned table patterns shown in Batch Mode also
work in Stream Mode — just add `commit_identifier` to the
`upsert_by_arrow_with_key` and `commit` calls.
+</TabItem>
+
+</Tabs>
+
## Update Columns By Shards
If you want to **compute a derived column** (or **update an existing column
based on other columns**) without providing
@@ -379,7 +495,9 @@ If you want to **compute a derived column** (or **update an
existing column base
This is useful for backfilling a newly added column, or recomputing a column
from other columns.
-### Batch Mode
+<Tabs groupId="pypaimon-data-evolution-mode">
+
+<TabItem value="batch" label="Batch">
**Example: compute `d = c + b - a`**
@@ -462,7 +580,9 @@ commit.commit(commit_messages)
commit.close()
```
-### Stream Mode
+</TabItem>
+
+<TabItem value="stream" label="Stream">
```python
import pyarrow as pa
@@ -518,6 +638,10 @@ commit_messages = upd.prepare_commit()
table_commit.commit(commit_messages, commit_identifier=1)
```
+</TabItem>
+
+</Tabs>
+
**Notes**
- **Row order matters**: the batches you write must have the **same number of
rows** as the batches you read, in the
diff --git a/paimon-python/pypaimon/tests/table_update_test.py
b/paimon-python/pypaimon/tests/table_update_test.py
index 97f8a4f676..500224324e 100644
--- a/paimon-python/pypaimon/tests/table_update_test.py
+++ b/paimon-python/pypaimon/tests/table_update_test.py
@@ -50,6 +50,10 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
def _apply_update(self, table_update, data, cid):
raise NotImplementedError
+ def _apply_update_by_predicate(
+ self, table_update, predicate, assignments, cid):
+ raise NotImplementedError
+
# ------------------------------------------------------------------
# Helpers built on the primitives
# ------------------------------------------------------------------
@@ -79,6 +83,36 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
}, schema=self.pa_schema))
return table
+ def _create_global_indexed_table_for_predicate_update(self):
+ options = dict(self.table_options)
+ options.update({
+ 'global-index.enabled': 'true',
+ 'bucket': '-1',
+ 'file.format': 'parquet',
+ })
+ table = self._create_table(options=options)
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [1, 2],
+ 'name': ['old', 'indexed'],
+ 'age': [10, 15],
+ 'city': ['NYC', 'LA'],
+ }, schema=self.pa_schema))
+
+ self.assertEqual(
+ 1,
+ table.create_global_index(
+ 'name',
+ options={'sorted-index.records-per-range': '1000'},
+ ),
+ )
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [3, 4],
+ 'name': ['new', 'other'],
+ 'age': [20, 30],
+ 'city': ['LA', 'SF'],
+ }, schema=self.pa_schema))
+ return table
+
def _do_update(self, table, data, columns):
"""End-to-end ``update_by_arrow_with_row_id`` + commit. Returns the
commit messages so callers can inspect produced files."""
@@ -91,6 +125,21 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
tc.close()
return msgs
+ def _do_update_by_predicate(self, table, predicate, assignments):
+ wb = self._make_write_builder(table)
+ tu = wb.new_update()
+ cid = self._next_commit_id()
+ msgs = self._apply_update_by_predicate(
+ tu,
+ predicate,
+ assignments,
+ cid,
+ )
+ tc = wb.new_commit()
+ self._apply_commit(tc, msgs, cid)
+ tc.close()
+ return msgs
+
# ==================================================================
# Shared tests (run under both batch and stream modes)
# ==================================================================
@@ -107,6 +156,331 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
self._read_all(table)['age'].to_pylist(),
)
+ def test_update_by_predicate(self):
+ table = self._create_seeded_table()
+ wb = self._make_write_builder(table)
+ tu = wb.new_update()
+ pb = tu.new_predicate_builder()
+ predicate = pb.greater_or_equal('age', 35)
+
+ cid = self._next_commit_id()
+ msgs = self._apply_update_by_predicate(
+ tu,
+ predicate,
+ {'age': 99, 'city': 'Updated'},
+ cid,
+ )
+ tc = wb.new_commit()
+ self._apply_commit(tc, msgs, cid)
+ tc.close()
+
+ result = self._read_all(table)
+ self.assertEqual(
+ [25, 30, 99, 99, 99],
+ result['age'].to_pylist(),
+ )
+ self.assertEqual(
+ ['NYC', 'LA', 'Updated', 'Updated', 'Updated'],
+ result['city'].to_pylist(),
+ )
+ self.assertEqual(
+ ['Alice', 'Bob', 'Charlie', 'David', 'Eve'],
+ result['name'].to_pylist(),
+ )
+
+ def test_update_by_predicate_no_match_is_noop(self):
+ table = self._create_seeded_table()
+ pb = table.new_read_builder().new_predicate_builder()
+ msgs = self._do_update_by_predicate(
+ table,
+ pb.greater_than('age', 100),
+ {'age': 1},
+ )
+
+ self.assertEqual([], msgs)
+ self.assertEqual(
+ [25, 30, 35, 40, 45],
+ self._read_all(table)['age'].to_pylist(),
+ )
+
+ def test_update_by_predicate_accepts_array_chunked_and_scalar_values(self):
+ table = self._create_seeded_table()
+ pb = table.new_read_builder().new_predicate_builder()
+ self._do_update_by_predicate(
+ table,
+ pb.greater_or_equal('age', 35),
+ {
+ 'age': pa.array([101, 102, 103], type=pa.int64()),
+ 'city': pa.chunked_array([
+ pa.array(['Chicago_v2']),
+ pa.array(['Houston_v2', 'Phoenix_v2']),
+ ]),
+ 'name': pa.scalar('patched'),
+ },
+ )
+
+ result = self._read_all(table)
+ rows = {
+ row_id: (name, age, city)
+ for row_id, name, age, city in zip(
+ result['id'].to_pylist(),
+ result['name'].to_pylist(),
+ result['age'].to_pylist(),
+ result['city'].to_pylist(),
+ )
+ }
+ self.assertEqual(
+ {
+ 1: ('Alice', 25, 'NYC'),
+ 2: ('Bob', 30, 'LA'),
+ 3: ('patched', 101, 'Chicago_v2'),
+ 4: ('patched', 102, 'Houston_v2'),
+ 5: ('patched', 103, 'Phoenix_v2'),
+ },
+ rows,
+ )
+
+ def test_update_by_predicate_updates_all_rows_when_predicate_is_none(self):
+ table = self._create_seeded_table()
+ self._do_update_by_predicate(
+ table,
+ None,
+ {'age': pa.scalar(7, type=pa.int64()), 'city': None},
+ )
+
+ result = self._read_all(table)
+ self.assertEqual([7, 7, 7, 7, 7], result['age'].to_pylist())
+ self.assertEqual([None, None, None, None, None],
+ result['city'].to_pylist())
+ self.assertEqual(
+ ['Alice', 'Bob', 'Charlie', 'David', 'Eve'],
+ result['name'].to_pylist(),
+ )
+
+ def
test_update_by_predicate_rejects_assignment_array_length_mismatch(self):
+ table = self._create_seeded_table()
+ pb = table.new_read_builder().new_predicate_builder()
+ with self.assertRaises(ValueError) as ctx:
+ self._do_update_by_predicate(
+ table,
+ pb.greater_or_equal('age', 35),
+ {'age': pa.array([1, 2], type=pa.int32())},
+ )
+
+ self.assertIn('Assignment array length', str(ctx.exception))
+ self.assertEqual(
+ [25, 30, 35, 40, 45],
+ self._read_all(table)['age'].to_pylist(),
+ )
+
+ def test_update_by_predicate_rejects_uncastable_assignment(self):
+ table = self._create_seeded_table()
+ pb = table.new_read_builder().new_predicate_builder()
+ with self.assertRaises((pa.ArrowInvalid, pa.ArrowTypeError,
ValueError)):
+ self._do_update_by_predicate(
+ table,
+ pb.equal('id', 1),
+ {'age': 'not-an-int'},
+ )
+
+ self.assertEqual(
+ [25, 30, 35, 40, 45],
+ self._read_all(table)['age'].to_pylist(),
+ )
+
+ def
test_update_by_predicate_with_global_index_updates_unindexed_rows(self):
+ table = self._create_global_indexed_table_for_predicate_update()
+
+ pb = table.new_read_builder().new_predicate_builder()
+ self._do_update_by_predicate(
+ table,
+ pb.equal('name', 'new'),
+ {'age': 21},
+ )
+
+ result = self._read_all(table)
+ ages_by_id = dict(zip(
+ result['id'].to_pylist(),
+ result['age'].to_pylist(),
+ ))
+ self.assertEqual({1: 10, 2: 15, 3: 21, 4: 30}, ages_by_id)
+
+ def
test_update_by_predicate_with_global_index_falls_back_to_full_scan(self):
+ table = self._create_global_indexed_table_for_predicate_update()
+
+ pb = table.new_read_builder().new_predicate_builder()
+ self._do_update_by_predicate(
+ table,
+ pb.equal('city', 'LA'),
+ {'age': 88},
+ )
+
+ result = self._read_all(table)
+ ages_by_id = dict(zip(
+ result['id'].to_pylist(),
+ result['age'].to_pylist(),
+ ))
+ self.assertEqual({1: 10, 2: 88, 3: 88, 4: 30}, ages_by_id)
+
+ def
test_update_by_predicate_with_global_index_handles_compound_predicate(self):
+ table = self._create_global_indexed_table_for_predicate_update()
+
+ pb = table.new_read_builder().new_predicate_builder()
+ predicate = pb.or_predicates([
+ pb.equal('name', 'old'),
+ pb.equal('city', 'SF'),
+ ])
+ self._do_update_by_predicate(table, predicate, {'age': 77})
+
+ result = self._read_all(table)
+ ages_by_id = dict(zip(
+ result['id'].to_pylist(),
+ result['age'].to_pylist(),
+ ))
+ self.assertEqual({1: 77, 2: 15, 3: 20, 4: 77}, ages_by_id)
+
+ def test_update_by_predicate_resolves_time_travel_scan_snapshot(self):
+ table = self._create_table()
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [1],
+ 'name': ['old'],
+ 'age': [10],
+ 'city': ['NYC'],
+ }, schema=self.pa_schema))
+ table.create_tag('before_new')
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [2],
+ 'name': ['new'],
+ 'age': [20],
+ 'city': ['LA'],
+ }, schema=self.pa_schema))
+
+ travel_table = table.copy({'scan.tag-name': 'before_new'})
+ pb = travel_table.new_read_builder().new_predicate_builder()
+ msgs = self._do_update_by_predicate(
+ travel_table,
+ pb.equal('name', 'new'),
+ {'age': 21},
+ )
+
+ self.assertEqual([], msgs)
+ result = self._read_all(table)
+ ages_by_id = dict(zip(
+ result['id'].to_pylist(),
+ result['age'].to_pylist(),
+ ))
+ self.assertEqual({1: 10, 2: 20}, ages_by_id)
+
+ def test_update_by_predicate_resolves_scan_snapshot_id(self):
+ table = self._create_table()
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [1],
+ 'name': ['old'],
+ 'age': [10],
+ 'city': ['NYC'],
+ }, schema=self.pa_schema))
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [2],
+ 'name': ['new'],
+ 'age': [20],
+ 'city': ['LA'],
+ }, schema=self.pa_schema))
+
+ travel_table = table.copy({'scan.snapshot-id': str(snapshot.id)})
+ pb = travel_table.new_read_builder().new_predicate_builder()
+ msgs = self._do_update_by_predicate(
+ travel_table,
+ pb.equal('name', 'new'),
+ {'age': 21},
+ )
+
+ self.assertEqual([], msgs)
+ result = self._read_all(table)
+ ages_by_id = dict(zip(
+ result['id'].to_pylist(),
+ result['age'].to_pylist(),
+ ))
+ self.assertEqual({1: 10, 2: 20}, ages_by_id)
+
+ def test_update_by_predicate_resets_explicit_scan_mode_after_travel(self):
+ table = self._create_table()
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [1],
+ 'name': ['old'],
+ 'age': [10],
+ 'city': ['NYC'],
+ }, schema=self.pa_schema))
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [2],
+ 'name': ['new'],
+ 'age': [20],
+ 'city': ['LA'],
+ }, schema=self.pa_schema))
+
+ travel_table = table.copy({
+ 'scan.mode': 'from-timestamp',
+ 'scan.timestamp-millis': str(snapshot.time_millis),
+ })
+ pb = travel_table.new_read_builder().new_predicate_builder()
+ msgs = self._do_update_by_predicate(
+ travel_table,
+ pb.equal('name', 'new'),
+ {'age': 21},
+ )
+
+ self.assertEqual([], msgs)
+ result = self._read_all(table)
+ ages_by_id = dict(zip(
+ result['id'].to_pylist(),
+ result['age'].to_pylist(),
+ ))
+ self.assertEqual({1: 10, 2: 20}, ages_by_id)
+
+ def test_update_by_predicate_rejects_empty_assignments(self):
+ table = self._create_seeded_table()
+ pb = table.new_read_builder().new_predicate_builder()
+
+ with self.assertRaises(ValueError) as ctx:
+ self._do_update_by_predicate(
+ table,
+ pb.equal('id', 1),
+ {},
+ )
+ self.assertIn('assignments must not be empty', str(ctx.exception))
+
+ def test_update_by_predicate_rejects_unknown_column(self):
+ table = self._create_seeded_table()
+ pb = table.new_read_builder().new_predicate_builder()
+
+ with self.assertRaises(ValueError) as ctx:
+ self._do_update_by_predicate(
+ table,
+ pb.equal('id', 1),
+ {'unknown': 1},
+ )
+ self.assertIn('Column unknown is not in table schema',
+ str(ctx.exception))
+
+ def test_update_by_predicate_rejects_partition_column(self):
+ table = self._create_table(partition_keys=['city'])
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [1],
+ 'name': ['Alice'],
+ 'age': [25],
+ 'city': ['NYC'],
+ }, schema=self.pa_schema))
+ pb = table.new_read_builder().new_predicate_builder()
+
+ with self.assertRaises(ValueError) as ctx:
+ self._do_update_by_predicate(
+ table,
+ pb.equal('id', 1),
+ {'city': 'LA'},
+ )
+ self.assertIn('partition column', str(ctx.exception))
+
def test_update_multiple_columns(self):
"""Update ``age`` + ``city`` together; other columns are untouched."""
table = self._create_seeded_table()
@@ -659,11 +1033,23 @@ class _BatchModeMixin(BatchModeMixin):
def _apply_update(self, table_update, data, cid):
return table_update.update_by_arrow_with_row_id(data)
+ def _apply_update_by_predicate(
+ self, table_update, predicate, assignments, cid):
+ return table_update.update_by_predicate(predicate, assignments)
+
class _StreamModeMixin(StreamModeMixin):
def _apply_update(self, table_update, data, cid):
return table_update.update_by_arrow_with_row_id(data, cid)
+ def _apply_update_by_predicate(
+ self, table_update, predicate, assignments, cid):
+ return table_update.update_by_predicate(
+ predicate,
+ assignments,
+ cid,
+ )
+
# ======================================================================
# Concrete test classes
diff --git a/paimon-python/pypaimon/write/table_update.py
b/paimon-python/pypaimon/write/table_update.py
index 8271b19f37..fe1bbee83a 100644
--- a/paimon-python/pypaimon/write/table_update.py
+++ b/paimon-python/pypaimon/write/table_update.py
@@ -16,16 +16,25 @@
# under the License.
from collections import defaultdict
-from typing import List, Optional, Tuple
+from typing import Any, List, Mapping, Optional, Tuple
import pyarrow
import pyarrow as pa
from pypaimon.common.memory_size import MemorySize
+from pypaimon.common.options.core_options import (
+ CoreOptions,
+ GlobalIndexSearchMode,
+ StartupMode,
+)
+from pypaimon.common.predicate import Predicate
+from pypaimon.common.predicate_builder import PredicateBuilder
from pypaimon.globalindex import Range
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
from pypaimon.read.split import DataSplit
+from pypaimon.schema.data_types import PyarrowFieldParser
from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
+from pypaimon.snapshot.time_travel_util import SCAN_KEYS, TimeTravelUtil
from pypaimon.table.special_fields import SpecialFields
from pypaimon.write.commit_message import CommitMessage
from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
@@ -118,6 +127,9 @@ class TableUpdate:
def with_read_projection(self, projection: List[str]):
self.projection = projection
+ def new_predicate_builder(self) -> PredicateBuilder:
+ return self.table.new_read_builder().new_predicate_builder()
+
def new_shard_updator(self, shard_num: int, total_shard_count: int):
"""Create a shard updater for scan+rewrite style updates.
@@ -173,6 +185,136 @@ class TableUpdate:
self.table, self.commit_user, commit_identifier
).upsert(table, upsert_keys, self.update_cols)
+ def _update_by_predicate(
+ self,
+ predicate: Optional[Predicate],
+ assignments: Mapping[str, Any],
+ commit_identifier: int,
+ ) -> List[CommitMessage]:
+ """Shared implementation for SQL-like ``UPDATE ... WHERE ...``.
+
+ ``predicate`` identifies the target rows. ``assignments`` maps target
+ column names to literal values. The method reads matching ``_ROW_ID``
+ values, builds an Arrow update table, then delegates to the existing
+ row-id update path.
+ """
+ self._validate_predicate_update(assignments)
+
+ scan_table = self._matched_update_scan_table()
+ read_builder = scan_table.new_read_builder()
+ if predicate is not None:
+ read_builder.with_filter(predicate)
+ read_builder.with_projection(
+ list(scan_table.field_names) + [SpecialFields.ROW_ID.name]
+ )
+ else:
+ read_builder.with_projection([SpecialFields.ROW_ID.name])
+
+ splits = read_builder.new_scan().plan().splits()
+ matched = read_builder.new_read().to_arrow(splits)
+ if matched.num_rows == 0:
+ return []
+
+ update_table = self._build_predicate_update_table(
+ matched[SpecialFields.ROW_ID.name],
+ assignments,
+ matched.num_rows,
+ )
+ return TableUpdateByRowId(
+ self.table, self.commit_user, commit_identifier,
+ ).update_columns(update_table, list(assignments.keys()))
+
+ def _matched_update_scan_table(self):
+ snapshot_manager = self.table.snapshot_manager()
+ snapshot = TimeTravelUtil.try_travel_to_snapshot(
+ self.table.options.options,
+ self.table.tag_manager(),
+ snapshot_manager,
+ )
+ if snapshot is None:
+ snapshot = snapshot_manager.get_latest_snapshot()
+ if snapshot is None:
+ return self.table
+
+ dynamic_options = {
+ CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key():
+ GlobalIndexSearchMode.FULL.value,
+ CoreOptions.SCAN_MODE.key(): StartupMode.DEFAULT.value,
+ CoreOptions.SCAN_SNAPSHOT_ID.key(): str(snapshot.id),
+ }
+ for scan_key in SCAN_KEYS:
+ if (
+ scan_key != CoreOptions.SCAN_SNAPSHOT_ID.key()
+ and self.table.options.options.contains_key(scan_key)
+ ):
+ dynamic_options[scan_key] = None
+
+ return self.table.copy(dynamic_options)
+
+ def _validate_predicate_update(self, assignments: Mapping[str, Any]):
+ if not self.table.options.data_evolution_enabled():
+ raise ValueError(
+ "update_by_predicate requires "
+ "'data-evolution.enabled' = 'true'."
+ )
+ if not self.table.options.row_tracking_enabled():
+ raise ValueError(
+ "update_by_predicate requires "
+ "'row-tracking.enabled' = 'true'."
+ )
+ if not assignments:
+ raise ValueError("assignments must not be empty.")
+
+ partition_keys = set(self.table.partition_keys)
+ for col in assignments:
+ if col not in self.table.field_names:
+ raise ValueError(f"Column {col} is not in table schema.")
+ if col in partition_keys:
+ raise ValueError(
+ "update_by_predicate does not support updating "
+ f"partition column '{col}'."
+ )
+
+ def _build_predicate_update_table(
+ self,
+ row_ids,
+ assignments: Mapping[str, Any],
+ row_count: int,
+ ) -> pa.Table:
+ table_schema = PyarrowFieldParser.from_paimon_schema(
+ self.table.table_schema.fields
+ )
+ arrays = [row_ids]
+ fields = [pa.field(SpecialFields.ROW_ID.name, pa.int64())]
+ for col, value in assignments.items():
+ target_field = table_schema.field(col)
+ arrays.append(
+ self._assignment_to_array(value, target_field.type, row_count)
+ )
+ fields.append(target_field)
+ return pa.Table.from_arrays(arrays, schema=pa.schema(fields))
+
+ @staticmethod
+ def _assignment_to_array(
+ value: Any, data_type: pa.DataType, row_count: int):
+ if isinstance(value, pa.ChunkedArray):
+ array = value.combine_chunks()
+ elif isinstance(value, pa.Array):
+ array = value
+ else:
+ if isinstance(value, pa.Scalar):
+ value = value.as_py()
+ return pa.array([value] * row_count, type=data_type)
+
+ if len(array) != row_count:
+ raise ValueError(
+ "Assignment array length must match matched row count: "
+ f"{len(array)} != {row_count}."
+ )
+ if array.type != data_type:
+ array = array.cast(data_type)
+ return array
+
class BatchTableUpdate(TableUpdate):
"""Batch-mode table update; commit messages always use
@@ -190,6 +332,16 @@ class BatchTableUpdate(TableUpdate):
table, upsert_keys, BATCH_COMMIT_IDENTIFIER
)
+ def update_by_predicate(
+ self,
+ predicate: Optional[Predicate],
+ assignments: Mapping[str, Any],
+ ) -> List[CommitMessage]:
+ """Update rows matching ``predicate`` with literal assignments."""
+ return self._update_by_predicate(
+ predicate, assignments, BATCH_COMMIT_IDENTIFIER
+ )
+
class StreamTableUpdate(TableUpdate):
"""Stream-mode table update; the same instance may drive many rounds,
@@ -214,6 +366,18 @@ class StreamTableUpdate(TableUpdate):
table, upsert_keys, commit_identifier
)
+ def update_by_predicate(
+ self,
+ predicate: Optional[Predicate],
+ assignments: Mapping[str, Any],
+ commit_identifier: int,
+ ) -> List[CommitMessage]:
+ """Update rows matching ``predicate`` with literal assignments,
+ tagging the produced commit messages with ``commit_identifier``."""
+ return self._update_by_predicate(
+ predicate, assignments, commit_identifier
+ )
+
class ShardTableUpdator:
