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 aebc9846fe [python][ray] Add row-id range processing API (#9112)
aebc9846fe is described below
commit aebc9846fe0d574b48cbb62ec3161a5d47dfca72
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Aug 8 19:09:32 2026 +0800
[python][ray] Add row-id range processing API (#9112)
---
docs/docs/pypaimon/ray-data.md | 113 +++++++++++++
paimon-python/pypaimon/ray/__init__.py | 2 +
.../pypaimon/ray/process_row_id_ranges.py | 101 +++++++++++
.../tests/ray_process_row_id_ranges_test.py | 188 +++++++++++++++++++++
paimon-python/pypaimon/tests/vector_table_test.py | 95 +++++++++++
.../pypaimon/write/table_update_by_row_id.py | 13 +-
6 files changed, 510 insertions(+), 2 deletions(-)
diff --git a/docs/docs/pypaimon/ray-data.md b/docs/docs/pypaimon/ray-data.md
index 082abfe3ef..3f90068336 100644
--- a/docs/docs/pypaimon/ray-data.md
+++ b/docs/docs/pypaimon/ray-data.md
@@ -637,3 +637,116 @@ ds = read_by_row_id(
- For a non-empty target, the `row_ids` source is consumed lazily by the
downstream
action, not read here. A lazy source missing `row_id_col` raises when the
read runs
(a materialized source raises up front).
+
+## Process Row Id Ranges
+
+`process_row_id_ranges` plans the latest snapshot into logical file groups and
+calls a user-supplied processor synchronously for each target-sized batch. The
+processor receives a `List[Range]` and owns the read, distributed computation,
+commit, and retry policy. Base files and overlapping data-evolution, BLOB, or
+VECTOR files remain in one indivisible group.
+
+`rows_per_commit` is therefore a target rather than a hard limit: the function
+never splits a file group, so a batch can contain more rows. The range plan is
+captured once at the start of a run, callbacks execute in row-id order, and an
+exception stops later callbacks.
+
+### Resumable embedding backfill from a BLOB column
+
+The following pattern reads an `image` BLOB, computes a nullable `embedding`
+VECTOR with Ray, and commits about one million row ids at a time. It pushes
+`embedding IS NULL` into each range scan, so a completed row is filtered before
+its image payload is materialized. Rerun the whole function after a failure;
+already committed ranges are skipped automatically.
+
+```python
+import pyarrow as pa
+
+from my_embedding_model import load_model
+from pypaimon import CatalogFactory
+from pypaimon.ray import process_row_id_ranges, update_by_row_id
+
+TARGET = "database_name.images"
+CATALOG_OPTIONS = {"warehouse": "/path/to/warehouse"}
+EMBEDDING_DIM = 768
+
+
+class EmbedImages:
+ def __init__(self):
+ # Constructed once in every Ray actor, not once per Arrow batch.
+ self.model = load_model()
+
+ def __call__(self, batch: pa.Table) -> pa.Table:
+ vectors = self.model.encode(batch["image"].to_pylist())
+ return pa.table({
+ "_ROW_ID": batch["_ROW_ID"],
+ "embedding": pa.array(
+ vectors.tolist(),
+ type=pa.list_(pa.float32(), EMBEDDING_DIM),
+ ),
+ })
+
+
+def process_ranges(ranges):
+ # Resolve a fresh table for every batch so this scan sees embeddings
+ # committed by earlier callbacks. Force BLOB payloads rather than
descriptors.
+ table = (
+ CatalogFactory.create(CATALOG_OPTIONS)
+ .get_table(TARGET)
+ .copy({"blob-as-descriptor": "false"})
+ )
+ read_builder = table.new_read_builder().with_projection(
+ ["image", "embedding", "_ROW_ID"]
+ )
+ read_builder.with_filter(
+ read_builder.new_predicate_builder().is_null("embedding")
+ )
+ splits = (
+ read_builder.new_scan()
+ .with_row_ranges(ranges)
+ .plan()
+ .splits()
+ )
+ pending = read_builder.new_read().to_ray(
+ splits,
+ concurrency=64,
+ ray_remote_args={"num_cpus": 1},
+ )
+ if pending.limit(1).count() == 0:
+ return
+
+ updates = pending.map_batches(
+ EmbedImages,
+ batch_format="pyarrow",
+ batch_size=128,
+ concurrency=8, # required for a callable-class Ray actor pool
+ num_gpus=1,
+ )
+
+ # update_by_row_id executes the Ray pipeline and makes one Paimon commit.
+ # It is valid for VECTOR/ARRAY embedding columns; BLOB columns themselves
+ # cannot be updated through update_by_row_id.
+ update_by_row_id(
+ target=TARGET,
+ source=updates,
+ catalog_options=CATALOG_OPTIONS,
+ update_cols=["embedding"],
+ num_partitions=128,
+ )
+
+
+process_row_id_ranges(
+ TARGET,
+ CATALOG_OPTIONS,
+ rows_per_commit=1_000_000,
+ processor=process_ranges,
+)
+```
+
+The target must enable `row-tracking.enabled` and
+`data-evolution.enabled`; `embedding` must be nullable and the table must not
+enable deletion vectors. If the source BLOB or embedding model can change,
+use an additional source/model-version column instead of treating every
+non-null embedding as permanently complete. `process_row_id_ranges` does not
+retry a failed processor itself—the resumability in this example comes from
+rerunning it and selecting only rows whose embedding is still null.
diff --git a/paimon-python/pypaimon/ray/__init__.py
b/paimon-python/pypaimon/ray/__init__.py
index 03de068249..63141ecd41 100644
--- a/paimon-python/pypaimon/ray/__init__.py
+++ b/paimon-python/pypaimon/ray/__init__.py
@@ -30,6 +30,7 @@ from pypaimon.ray.data_evolution_merge_transform import (
)
from pypaimon.ray.update_by_row_id import update_by_row_id
from pypaimon.ray.read_by_row_id import read_by_row_id
+from pypaimon.ray.process_row_id_ranges import process_row_id_ranges
__all__ = [
"read_paimon",
@@ -40,6 +41,7 @@ __all__ = [
"merge_into",
"update_by_row_id",
"read_by_row_id",
+ "process_row_id_ranges",
"WhenMatched",
"WhenNotMatched",
"source_col",
diff --git a/paimon-python/pypaimon/ray/process_row_id_ranges.py
b/paimon-python/pypaimon/ray/process_row_id_ranges.py
new file mode 100644
index 0000000000..4af31d5ef0
--- /dev/null
+++ b/paimon-python/pypaimon/ray/process_row_id_ranges.py
@@ -0,0 +1,101 @@
+# 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.
+
+"""Sequential processing of row-id ranges at file-group boundaries."""
+
+from typing import Callable, Dict, List
+
+from pypaimon.utils.range import Range
+from pypaimon.utils.range_helper import RangeHelper
+
+__all__ = ["process_row_id_ranges"]
+
+
+def process_row_id_ranges(
+ target: str,
+ catalog_options: Dict[str, str],
+ *,
+ rows_per_commit: int,
+ processor: Callable[[List[Range]], None]
+) -> None:
+ """Process the target's row-id file groups in sequential batches.
+
+ The latest snapshot is planned once before processing starts. Files with
+ overlapping row-id ranges (for example, a base file and its data-evolution
+ files) form one indivisible file group. Adjacent groups are accumulated
+ until their row count reaches ``rows_per_commit``, then ``processor`` is
+ called synchronously with their inclusive :class:`Range` objects.
+
+ A batch may exceed ``rows_per_commit`` because file groups are never split.
+ The processor owns reading, distributed execution, committing, retries, and
+ cleanup. Its exception is propagated immediately and later batches are not
+ processed.
+ """
+ _validate_arguments(rows_per_commit, processor)
+
+ from pypaimon.catalog.catalog_factory import CatalogFactory
+
+ table = CatalogFactory.create(catalog_options).get_table(target)
+ if not table.options.row_tracking_enabled():
+ raise ValueError(
+ "process_row_id_ranges requires 'row-tracking.enabled'='true' "
+ "on '{}'.".format(target)
+ )
+
+ pending_ranges = []
+ pending_rows = 0
+ for row_id_range in _file_group_ranges(table):
+ pending_ranges.append(row_id_range)
+ pending_rows += row_id_range.count()
+ if pending_rows >= rows_per_commit:
+ processor(pending_ranges)
+ pending_ranges = []
+ pending_rows = 0
+
+ if pending_ranges:
+ processor(pending_ranges)
+
+
+def _validate_arguments(rows_per_commit, processor) -> None:
+ if (
+ isinstance(rows_per_commit, bool)
+ or not isinstance(rows_per_commit, int)
+ or rows_per_commit <= 0
+ ):
+ raise ValueError("rows_per_commit must be a positive integer.")
+ if not callable(processor):
+ raise ValueError("processor must be callable.")
+
+
+def _file_group_ranges(table) -> List[Range]:
+ plan = table.new_read_builder().new_scan().plan_for_write()
+ files = [data_file for split in plan.splits() for data_file in split.files]
+ file_groups = RangeHelper(
+ lambda data_file: data_file.non_null_row_id_range()
+ ).merge_overlapping_ranges(files)
+
+ ranges = []
+ for file_group in file_groups:
+ group_ranges = [
+ data_file.non_null_row_id_range() for data_file in file_group
+ ]
+ ranges.append(
+ Range(
+ min(row_range.from_ for row_range in group_ranges),
+ max(row_range.to for row_range in group_ranges),
+ )
+ )
+ return sorted(ranges, key=lambda row_range: row_range.from_)
diff --git a/paimon-python/pypaimon/tests/ray_process_row_id_ranges_test.py
b/paimon-python/pypaimon/tests/ray_process_row_id_ranges_test.py
new file mode 100644
index 0000000000..1978e733ec
--- /dev/null
+++ b/paimon-python/pypaimon/tests/ray_process_row_id_ranges_test.py
@@ -0,0 +1,188 @@
+# 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 os
+import shutil
+import tempfile
+import unittest
+import uuid
+from unittest import mock
+
+import pyarrow as pa
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.ray import process_row_id_ranges
+from pypaimon.utils.range import Range
+
+
+class ProcessRowIdRangesTest(unittest.TestCase):
+
+ schema = pa.schema([
+ ("id", pa.int32()),
+ ("name", pa.string()),
+ ])
+
+ @classmethod
+ def setUpClass(cls):
+ cls.tempdir = tempfile.mkdtemp()
+ cls.catalog_options = {
+ "warehouse": os.path.join(cls.tempdir, "warehouse")
+ }
+ cls.catalog = CatalogFactory.create(cls.catalog_options)
+ cls.catalog.create_database("default", True)
+
+ @classmethod
+ def tearDownClass(cls):
+ shutil.rmtree(cls.tempdir, ignore_errors=True)
+
+ def _create(self, options=None):
+ target = "default.ranges_{}".format(uuid.uuid4().hex[:8])
+ self.catalog.create_table(
+ target,
+ Schema.from_pyarrow_schema(self.schema, options=options or {}),
+ False,
+ )
+ return target
+
+ def _write(self, target, row_count):
+ table = self.catalog.get_table(target)
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(pa.Table.from_pydict({
+ "id": list(range(row_count)),
+ "name": ["n{}".format(i) for i in range(row_count)],
+ }, schema=self.schema))
+ commit = builder.new_commit()
+ commit.commit(writer.prepare_commit())
+ writer.close()
+ commit.close()
+
+ def test_processes_file_groups_in_target_sized_sequential_batches(self):
+ target = self._create({
+ "row-tracking.enabled": "true",
+ "data-evolution.enabled": "true",
+ "target-file-row-num": "3",
+ })
+ self._write(target, 10)
+
+ # Add an overlapping data-evolution file. It belongs to the first
+ # logical file group and must not be counted as another range.
+ table = self.catalog.get_table(target)
+ builder = table.new_batch_write_builder()
+ messages = (
+ builder.new_update()
+ .with_update_type(["name"])
+ .update_by_arrow_with_row_id(pa.table({
+ "_ROW_ID": pa.array([1], type=pa.int64()),
+ "name": pa.array(["updated"], type=pa.string()),
+ }))
+ )
+ commit = builder.new_commit()
+ commit.commit(messages)
+ commit.close()
+
+ batches = []
+ process_row_id_ranges(
+ target,
+ self.catalog_options,
+ rows_per_commit=7,
+ processor=batches.append,
+ )
+
+ self.assertEqual([
+ [Range(0, 2), Range(3, 5), Range(6, 8)],
+ [Range(9, 9)],
+ ], batches)
+
+ def test_processor_failure_stops_later_batches(self):
+ target = self._create({
+ "row-tracking.enabled": "true",
+ "data-evolution.enabled": "true",
+ "target-file-row-num": "2",
+ })
+ self._write(target, 6)
+ calls = []
+
+ def processor(ranges):
+ calls.append(ranges)
+ if len(calls) == 2:
+ raise RuntimeError("processor failed")
+
+ with self.assertRaisesRegex(RuntimeError, "processor failed"):
+ process_row_id_ranges(
+ target,
+ self.catalog_options,
+ rows_per_commit=2,
+ processor=processor,
+ )
+
+ self.assertEqual([[Range(0, 1)], [Range(2, 3)]], calls)
+
+ def test_empty_table_does_not_call_processor(self):
+ target = self._create({"row-tracking.enabled": "true"})
+ processor = mock.Mock()
+
+ process_row_id_ranges(
+ target,
+ self.catalog_options,
+ rows_per_commit=10,
+ processor=processor,
+ )
+
+ processor.assert_not_called()
+
+ def test_requires_row_tracking(self):
+ target = self._create()
+ with self.assertRaisesRegex(ValueError, "row-tracking.enabled"):
+ process_row_id_ranges(
+ target,
+ self.catalog_options,
+ rows_per_commit=10,
+ processor=lambda ranges: None,
+ )
+
+ def test_validates_arguments_before_loading_table(self):
+ invalid_values = [True, False, 0, -1, 1.5, "1", None]
+ for value in invalid_values:
+ with self.subTest(rows_per_commit=value), mock.patch(
+ "pypaimon.catalog.catalog_factory.CatalogFactory.create"
+ ) as create:
+ with self.assertRaisesRegex(ValueError, "positive integer"):
+ process_row_id_ranges(
+ "default.missing",
+ {},
+ rows_per_commit=value,
+ processor=lambda ranges: None,
+ )
+ create.assert_not_called()
+
+ with mock.patch(
+ "pypaimon.catalog.catalog_factory.CatalogFactory.create"
+ ) as create:
+ with self.assertRaisesRegex(
+ ValueError, "processor must be callable"
+ ):
+ process_row_id_ranges(
+ "default.missing",
+ {},
+ rows_per_commit=1,
+ processor=None,
+ )
+ create.assert_not_called()
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/vector_table_test.py
b/paimon-python/pypaimon/tests/vector_table_test.py
index 0f96455bbd..6a4515475e 100644
--- a/paimon-python/pypaimon/tests/vector_table_test.py
+++ b/paimon-python/pypaimon/tests/vector_table_test.py
@@ -394,6 +394,101 @@ class VectorTableWriteReadTest(unittest.TestCase):
result = rb.new_read().to_arrow(splits).sort_by('id').to_pydict()
self.assertEqual(result['name'], ['updated', 'updated'])
+ def test_backfill_vector_column_added_after_existing_rows(self):
+ from pypaimon.schema.data_types import AtomicType, VectorType
+ from pypaimon.schema.schema_change import SchemaChange
+ from pypaimon.utils.range import Range
+
+ table_name = 'test_db.vector_backfill_added_column'
+ base_schema = pa.schema([
+ ('id', pa.int64()),
+ ('image', pa.large_binary()),
+ ])
+ opts = {
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'vector.file.format': 'parquet',
+ }
+ self.catalog.create_table(
+ table_name,
+ Schema.from_pyarrow_schema(base_schema, options=opts),
+ False,
+ )
+ table = self.catalog.get_table(table_name)
+ wb = table.new_batch_write_builder()
+ writer = wb.new_write()
+ writer.write_arrow(pa.table({
+ 'id': pa.array([1, 2, 3], type=pa.int64()),
+ 'image': pa.array([b'a', b'b', b'c'], type=pa.large_binary()),
+ }))
+ commit = wb.new_commit()
+ commit.commit(writer.prepare_commit())
+ writer.close()
+ commit.close()
+
+ self.catalog.alter_table(
+ table_name,
+ [SchemaChange.add_column(
+ 'embedding', VectorType(True, AtomicType('FLOAT'), 2))],
+ False,
+ )
+ table = self.catalog.get_table(table_name)
+ read_builder = table.new_read_builder().with_projection(
+ ['image', 'embedding', '_ROW_ID'])
+ read_builder.with_filter(
+ read_builder.new_predicate_builder().is_null('embedding'))
+ splits = (
+ read_builder.new_scan()
+ .with_row_ranges([Range(0, 2)])
+ .plan()
+ .splits()
+ )
+ pending = read_builder.new_read().to_arrow(splits).sort_by('_ROW_ID')
+ self.assertEqual([0, 1, 2], pending['_ROW_ID'].to_pylist())
+ self.assertEqual([b'a', b'b', b'c'], pending['image'].to_pylist())
+
+ updates = pa.table({
+ '_ROW_ID': pending['_ROW_ID'],
+ 'embedding': pa.array(
+ [[1.0, 0.0], [0.5, 0.5], [0.0, 1.0]],
+ type=pa.list_(pa.float32(), 2),
+ ),
+ })
+ wb = table.new_batch_write_builder()
+ messages = (
+ wb.new_update()
+ .with_update_type(['embedding'])
+ .update_by_arrow_with_row_id(updates)
+ )
+ commit = wb.new_commit()
+ commit.commit(messages)
+ commit.close()
+
+ table = self.catalog.get_table(table_name)
+ read_builder = table.new_read_builder().with_projection(
+ ['id', 'embedding'])
+ result = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits()).sort_by('id')
+ self.assertEqual(
+ [[1.0, 0.0], [0.5, 0.5], [0.0, 1.0]],
+ result['embedding'].to_pylist(),
+ )
+
+ retry_builder = table.new_read_builder().with_projection(
+ ['embedding', '_ROW_ID'])
+ retry_builder.with_filter(
+ retry_builder.new_predicate_builder().is_null('embedding'))
+ retry_splits = (
+ retry_builder.new_scan()
+ .with_row_ranges([Range(0, 2)])
+ .plan()
+ .splits()
+ )
+ self.assertEqual(
+ 0,
+ retry_builder.new_read().to_arrow(retry_splits).num_rows,
+ )
+
def
test_vector_table_partial_update_non_vector_column_with_rolling_files(self):
from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
diff --git a/paimon-python/pypaimon/write/table_update_by_row_id.py
b/paimon-python/pypaimon/write/table_update_by_row_id.py
index b0caa6ad03..c74fbc54fa 100644
--- a/paimon-python/pypaimon/write/table_update_by_row_id.py
+++ b/paimon-python/pypaimon/write/table_update_by_row_id.py
@@ -410,8 +410,17 @@ class TableUpdateByRowId:
bucket=owning_split.bucket,
raw_convertible=True,
)
- table_read = TableRead(self.table, predicate=None,
read_type=read_fields)
- return table_read.to_arrow([origin_split])
+ # Keep _ROW_ID as a row-count anchor. If every requested column was
+ # added after the original file was written, reading only those
+ # missing columns can otherwise produce a zero-row table instead of
+ # one null value per original row.
+ table_read = TableRead(
+ self.table,
+ predicate=None,
+ read_type=read_fields + [SpecialFields.ROW_ID],
+ )
+ original = table_read.to_arrow([origin_split])
+ return original.select([field.name for field in read_fields])
def _merge_update_with_original(
self,
