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 18a19e1c05 [python] Avoid full-file blob rewrite on partial row
updates (#8234)
18a19e1c05 is described below
commit 18a19e1c0574f078ac83ad581f433606972f4cc6
Author: XiaoHongbo <[email protected]>
AuthorDate: Sun Jun 14 19:05:46 2026 +0800
[python] Avoid full-file blob rewrite on partial row updates (#8234)
Updating one blob row via `_ROW_ID` produces a new blob delta file whose
`row_count` and `row_id_range` cover the **entire** original file's row
range, fully overlapping it. For a 1-of-100 update, the new file shows
`row_count=100`, `row_range=[0,99]` — identical to the original at the
metadata level.
After repeated partial updates, manifest entries grow with the table
size (not the update size), and every read has to walk a chain of
overlapping delta files.
---
paimon-python/pypaimon/tests/blob_table_test.py | 390 ++++++++++++++++++++-
.../pypaimon/write/table_update_by_row_id.py | 77 ++--
2 files changed, 427 insertions(+), 40 deletions(-)
diff --git a/paimon-python/pypaimon/tests/blob_table_test.py
b/paimon-python/pypaimon/tests/blob_table_test.py
index 9d87716a08..24b527c49b 100755
--- a/paimon-python/pypaimon/tests/blob_table_test.py
+++ b/paimon-python/pypaimon/tests/blob_table_test.py
@@ -1103,24 +1103,24 @@ class DedicatedFormatWriterTest(unittest.TestCase):
update_files = [f for msg in update_messages for f in msg.new_files]
update_blob_files = [f for f in update_files if
f.file_name.endswith('.blob')]
- self.assertGreater(len(update_blob_files), 0)
+ self.assertEqual(len(update_blob_files), 1)
+ self.assertEqual(update_blob_files[0].row_count, 2)
+ self.assertEqual(update_blob_files[0].first_row_id, 0)
self.assertTrue(all(f.write_cols == ['blob_data'] for f in
update_files))
- update_blob_lengths = []
blob_fields = [field for field in table.fields if field.name ==
'blob_data']
- for blob_file in update_blob_files:
- blob_reader = FormatBlobReader(
- file_io=table.file_io,
- file_path=blob_file.file_path,
- read_fields=['blob_data'],
- full_fields=blob_fields,
- push_down_predicate=None,
- blob_as_descriptor=False,
- )
- update_blob_lengths.extend(blob_reader.blob_lengths)
- blob_reader.close()
+ blob_reader = FormatBlobReader(
+ file_io=table.file_io,
+ file_path=update_blob_files[0].file_path,
+ read_fields=['blob_data'],
+ full_fields=blob_fields,
+ push_down_predicate=None,
+ blob_as_descriptor=False,
+ )
+ update_blob_lengths = list(blob_reader.blob_lengths)
+ blob_reader.close()
self.assertEqual(
update_blob_lengths.count(BlobFormatWriter.PLACE_HOLDER_LENGTH),
- 2,
+ 1,
)
read_builder = table.new_read_builder()
@@ -1135,6 +1135,366 @@ class DedicatedFormatWriterTest(unittest.TestCase):
3: b'blob-3',
})
+ def test_blob_update_single_row_at_first_position(self):
+ from pypaimon import Schema
+
+ pa_schema = pa.schema([
+ ('id', pa.int32()),
+ ('blob_data', pa.large_binary()),
+ ])
+ schema = Schema.from_pyarrow_schema(pa_schema, options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ })
+ self.catalog.create_table(
+ 'test_db.blob_update_first_row', schema, False)
+ table = self.catalog.get_table('test_db.blob_update_first_row')
+
+ initial = pa.Table.from_pydict({
+ 'id': list(range(10)),
+ 'blob_data': [f'blob-{i}'.encode() for i in range(10)],
+ }, schema=pa_schema)
+ write_builder = table.new_batch_write_builder()
+ writer = write_builder.new_write()
+ writer.write_arrow(initial)
+ write_builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+
+ table = self.catalog.get_table('test_db.blob_update_first_row')
+ update_builder = table.new_batch_write_builder()
+ table_update = update_builder.new_update().with_update_type(
+ ['blob_data'])
+ update_data = pa.Table.from_pydict({
+ '_ROW_ID': pa.array([0], type=pa.int64()),
+ 'blob_data': pa.array([b'updated-0'], type=pa.large_binary()),
+ })
+ update_messages = table_update.update_by_arrow_with_row_id(update_data)
+ update_builder.new_commit().commit(update_messages)
+
+ update_blob_files = [
+ f for msg in update_messages for f in msg.new_files
+ if f.file_name.endswith('.blob')
+ ]
+ self.assertEqual(len(update_blob_files), 1)
+ self.assertEqual(update_blob_files[0].row_count, 1)
+ self.assertEqual(update_blob_files[0].first_row_id, 0)
+
+ read_builder = table.new_read_builder()
+ result = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits())
+ by_id = {
+ row['id']: row['blob_data']
+ for row in result.select(['id', 'blob_data']).to_pylist()
+ }
+ self.assertEqual(by_id[0], b'updated-0')
+ self.assertEqual(by_id[9], b'blob-9')
+
+ def test_blob_update_all_rows_full_span(self):
+ from pypaimon import Schema
+
+ pa_schema = pa.schema([
+ ('id', pa.int32()),
+ ('blob_data', pa.large_binary()),
+ ])
+ schema = Schema.from_pyarrow_schema(pa_schema, options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ })
+ self.catalog.create_table(
+ 'test_db.blob_update_full', schema, False)
+ table = self.catalog.get_table('test_db.blob_update_full')
+
+ num_rows = 5
+ initial = pa.Table.from_pydict({
+ 'id': list(range(num_rows)),
+ 'blob_data': [f'blob-{i}'.encode() for i in range(num_rows)],
+ }, schema=pa_schema)
+ write_builder = table.new_batch_write_builder()
+ writer = write_builder.new_write()
+ writer.write_arrow(initial)
+ write_builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+
+ table = self.catalog.get_table('test_db.blob_update_full')
+ update_builder = table.new_batch_write_builder()
+ table_update = update_builder.new_update().with_update_type(
+ ['blob_data'])
+ update_data = pa.Table.from_pydict({
+ '_ROW_ID': pa.array(list(range(num_rows)), type=pa.int64()),
+ 'blob_data': pa.array(
+ [f'updated-{i}'.encode() for i in range(num_rows)],
+ type=pa.large_binary()),
+ })
+ update_messages = table_update.update_by_arrow_with_row_id(update_data)
+ update_builder.new_commit().commit(update_messages)
+
+ update_blob_files = [
+ f for msg in update_messages for f in msg.new_files
+ if f.file_name.endswith('.blob')
+ ]
+ self.assertEqual(len(update_blob_files), 1)
+ self.assertEqual(update_blob_files[0].row_count, num_rows)
+ self.assertEqual(update_blob_files[0].first_row_id, 0)
+
+ read_builder = table.new_read_builder()
+ result = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits())
+ by_id = {
+ row['id']: row['blob_data']
+ for row in result.select(['id', 'blob_data']).to_pylist()
+ }
+ for i in range(num_rows):
+ self.assertEqual(by_id[i], f'updated-{i}'.encode())
+
+ def test_blob_update_multiple_blob_columns(self):
+ from pypaimon import Schema
+
+ pa_schema = pa.schema([
+ ('id', pa.int32()),
+ ('blob_a', pa.large_binary()),
+ ('blob_b', pa.large_binary()),
+ ])
+ schema = Schema.from_pyarrow_schema(pa_schema, options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ })
+ self.catalog.create_table(
+ 'test_db.blob_update_two_cols', schema, False)
+ table = self.catalog.get_table('test_db.blob_update_two_cols')
+
+ initial = pa.Table.from_pydict({
+ 'id': list(range(5)),
+ 'blob_a': [f'a-{i}'.encode() for i in range(5)],
+ 'blob_b': [f'b-{i}'.encode() for i in range(5)],
+ }, schema=pa_schema)
+ write_builder = table.new_batch_write_builder()
+ writer = write_builder.new_write()
+ writer.write_arrow(initial)
+ write_builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+
+ table = self.catalog.get_table('test_db.blob_update_two_cols')
+ update_builder = table.new_batch_write_builder()
+ table_update = update_builder.new_update().with_update_type(
+ ['blob_a', 'blob_b'])
+ update_data = pa.Table.from_pydict({
+ '_ROW_ID': pa.array([2], type=pa.int64()),
+ 'blob_a': pa.array([b'a-updated'], type=pa.large_binary()),
+ 'blob_b': pa.array([b'b-updated'], type=pa.large_binary()),
+ })
+ update_messages = table_update.update_by_arrow_with_row_id(update_data)
+ update_builder.new_commit().commit(update_messages)
+
+ update_blob_files = [
+ f for msg in update_messages for f in msg.new_files
+ if f.file_name.endswith('.blob')
+ ]
+ self.assertEqual(len(update_blob_files), 2)
+ for f in update_blob_files:
+ self.assertEqual(f.row_count, 3)
+ self.assertEqual(f.first_row_id, 0)
+ cols_written = sorted(f.write_cols[0] for f in update_blob_files)
+ self.assertEqual(cols_written, ['blob_a', 'blob_b'])
+
+ read_builder = table.new_read_builder()
+ result = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits())
+ rows = result.select(['id', 'blob_a', 'blob_b']).to_pylist()
+ by_id = {r['id']: r for r in rows}
+ self.assertEqual(by_id[2]['blob_a'], b'a-updated')
+ self.assertEqual(by_id[2]['blob_b'], b'b-updated')
+ self.assertEqual(by_id[0]['blob_a'], b'a-0')
+ self.assertEqual(by_id[4]['blob_b'], b'b-4')
+
+ def test_blob_update_across_multiple_file_groups(self):
+ from pypaimon import Schema
+
+ pa_schema = pa.schema([
+ ('id', pa.int32()),
+ ('blob_data', pa.large_binary()),
+ ])
+ schema = Schema.from_pyarrow_schema(pa_schema, options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ })
+ self.catalog.create_table(
+ 'test_db.blob_update_multi_group', schema, False)
+ table = self.catalog.get_table('test_db.blob_update_multi_group')
+
+ # Two separate commits -> two data files with distinct first_row_id.
+ first = pa.Table.from_pydict({
+ 'id': [0, 1, 2, 3, 4],
+ 'blob_data': [f'a-{i}'.encode() for i in range(5)],
+ }, schema=pa_schema)
+ wb = table.new_batch_write_builder()
+ w = wb.new_write()
+ w.write_arrow(first)
+ wb.new_commit().commit(w.prepare_commit())
+ w.close()
+
+ table = self.catalog.get_table('test_db.blob_update_multi_group')
+ second = pa.Table.from_pydict({
+ 'id': [5, 6, 7, 8, 9],
+ 'blob_data': [f'b-{i}'.encode() for i in range(5)],
+ }, schema=pa_schema)
+ wb = table.new_batch_write_builder()
+ w = wb.new_write()
+ w.write_arrow(second)
+ wb.new_commit().commit(w.prepare_commit())
+ w.close()
+
+ # Touch one row in each group: _ROW_ID 0 (group A, pos 0)
+ # and _ROW_ID 7 (group B, pos 2).
+ table = self.catalog.get_table('test_db.blob_update_multi_group')
+ update_builder = table.new_batch_write_builder()
+ table_update = update_builder.new_update().with_update_type(
+ ['blob_data'])
+ update_data = pa.Table.from_pydict({
+ '_ROW_ID': pa.array([0, 7], type=pa.int64()),
+ 'blob_data': pa.array(
+ [b'updated-0', b'updated-7'],
+ type=pa.large_binary()),
+ })
+ update_messages = table_update.update_by_arrow_with_row_id(update_data)
+ update_builder.new_commit().commit(update_messages)
+
+ update_blob_files = [
+ f for msg in update_messages for f in msg.new_files
+ if f.file_name.endswith('.blob')
+ ]
+ self.assertEqual(len(update_blob_files), 2)
+ by_first = {f.first_row_id: f for f in update_blob_files}
+ self.assertEqual(set(by_first), {0, 5})
+ self.assertEqual(by_first[0].row_count, 1)
+ self.assertEqual(by_first[5].row_count, 3)
+
+ read_builder = table.new_read_builder()
+ result = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits())
+ by_id = {
+ row['id']: row['blob_data']
+ for row in result.select(['id', 'blob_data']).to_pylist()
+ }
+ self.assertEqual(by_id[0], b'updated-0')
+ self.assertEqual(by_id[7], b'updated-7')
+ self.assertEqual(by_id[3], b'a-3')
+ self.assertEqual(by_id[5], b'b-0')
+
+ def test_blob_update_scattered_endpoints_spans_full_range(self):
+ from pypaimon import Schema
+
+ pa_schema = pa.schema([
+ ('id', pa.int32()),
+ ('blob_data', pa.large_binary()),
+ ])
+ schema = Schema.from_pyarrow_schema(pa_schema, options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ })
+ self.catalog.create_table(
+ 'test_db.blob_update_scattered', schema, False)
+ table = self.catalog.get_table('test_db.blob_update_scattered')
+
+ num_rows = 100
+ initial = pa.Table.from_pydict({
+ 'id': list(range(num_rows)),
+ 'blob_data': [f'blob-{i}'.encode() for i in range(num_rows)],
+ }, schema=pa_schema)
+ write_builder = table.new_batch_write_builder()
+ writer = write_builder.new_write()
+ writer.write_arrow(initial)
+ write_builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+
+ table = self.catalog.get_table('test_db.blob_update_scattered')
+ update_builder = table.new_batch_write_builder()
+ table_update = update_builder.new_update().with_update_type(
+ ['blob_data'])
+ update_data = pa.Table.from_pydict({
+ '_ROW_ID': pa.array([0, 99], type=pa.int64()),
+ 'blob_data': pa.array(
+ [b'updated-blob-0', b'updated-blob-99'],
+ type=pa.large_binary()),
+ })
+ update_messages = table_update.update_by_arrow_with_row_id(update_data)
+ update_builder.new_commit().commit(update_messages)
+
+ update_blob_files = [
+ f for msg in update_messages for f in msg.new_files
+ if f.file_name.endswith('.blob')
+ ]
+ self.assertEqual(len(update_blob_files), 1)
+ self.assertEqual(update_blob_files[0].row_count, num_rows)
+ self.assertEqual(update_blob_files[0].first_row_id, 0)
+
+ read_builder = table.new_read_builder()
+ result = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits())
+ by_id = {
+ row['id']: row['blob_data']
+ for row in result.select(['id', 'blob_data']).to_pylist()
+ }
+ self.assertEqual(by_id[0], b'updated-blob-0')
+ self.assertEqual(by_id[99], b'updated-blob-99')
+ self.assertEqual(by_id[50], b'blob-50')
+
+ def test_blob_update_shrinks_delta_to_max_updated_row(self):
+ from pypaimon import Schema
+
+ pa_schema = pa.schema([
+ ('id', pa.int32()),
+ ('blob_data', pa.large_binary()),
+ ])
+ schema = Schema.from_pyarrow_schema(pa_schema, options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ })
+ self.catalog.create_table(
+ 'test_db.blob_update_max_span', schema, False)
+ table = self.catalog.get_table('test_db.blob_update_max_span')
+
+ num_rows = 100
+ initial = pa.Table.from_pydict({
+ 'id': list(range(num_rows)),
+ 'blob_data': [f'blob-{i}'.encode() for i in range(num_rows)],
+ }, schema=pa_schema)
+ write_builder = table.new_batch_write_builder()
+ writer = write_builder.new_write()
+ writer.write_arrow(initial)
+ write_builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+
+ table = self.catalog.get_table('test_db.blob_update_max_span')
+ update_builder = table.new_batch_write_builder()
+ table_update = update_builder.new_update().with_update_type(
+ ['blob_data'])
+ update_data = pa.Table.from_pydict({
+ '_ROW_ID': pa.array([50], type=pa.int64()),
+ 'blob_data': pa.array(
+ [b'updated-blob-50'], type=pa.large_binary()),
+ })
+ update_messages = table_update.update_by_arrow_with_row_id(update_data)
+ update_builder.new_commit().commit(update_messages)
+
+ update_blob_files = [
+ f for msg in update_messages for f in msg.new_files
+ if f.file_name.endswith('.blob')
+ ]
+ self.assertEqual(len(update_blob_files), 1)
+ self.assertEqual(update_blob_files[0].row_count, 51)
+ self.assertEqual(update_blob_files[0].first_row_id, 0)
+
+ read_builder = table.new_read_builder()
+ result = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits())
+ by_id = {
+ row['id']: row['blob_data']
+ for row in result.select(['id', 'blob_data']).to_pylist()
+ }
+ self.assertEqual(by_id[50], b'updated-blob-50')
+ self.assertEqual(by_id[0], b'blob-0')
+ self.assertEqual(by_id[99], b'blob-99')
+
def test_update_blob_column_with_rolling_files(self):
from pypaimon import Schema
@@ -1286,7 +1646,7 @@ class DedicatedFormatWriterTest(unittest.TestCase):
blob_reader.close()
self.assertEqual(
update_blob_lengths.count(BlobFormatWriter.PLACE_HOLDER_LENGTH),
- 3,
+ 2,
)
update_builder.new_commit().commit(update_messages)
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 130d2763a2..d5653cce98 100644
--- a/paimon-python/pypaimon/write/table_update_by_row_id.py
+++ b/paimon-python/pypaimon/write/table_update_by_row_id.py
@@ -57,6 +57,12 @@ class TableUpdateByRowId:
This update is designed for adding/updating specific columns in existing
tables.
Input data should contain _ROW_ID column.
+
+ Python is the writer-side source of truth for update-by-row-id. The Java
+ side ships the read path (``BlobFallbackRecordReader``) and pins the
+ on-disk blob delta layout via ``BlobUpdateTest`` only; there is no Java
+ writer. Changes to the blob delta layout here must stay compatible with
+ that reader.
"""
FIRST_ROW_ID_COLUMN = '_FIRST_ROW_ID'
@@ -270,12 +276,22 @@ class TableUpdateByRowId:
original_data: Optional[pa.Table],
update_data: pa.Table,
column_names: List[str],
- first_row_id: int) -> Tuple[Optional[pa.Table], Dict[str,
List[object]]]:
+ first_row_id: int,
+ ) -> Tuple[Optional[pa.Table], Dict[str, List[object]]]:
"""Merge update data with original data, preserving row order.
For rows that have updates, use the update values.
For rows without updates, use the original values (if available).
+ Blob delta files cover ``[first_row_id, first_row_id +
max_updated_pos]``
+ — anchored at the original file's first_row_id, spanning up to and
+ including the last updated row. The span is NOT shrunk at the head:
+ ``BlobFallbackRecordReader`` resolves placeholders by relative offset
+ from the delta file's ``first_row_id``, so anchoring anywhere other
+ than the original ``first_row_id`` would misalign unchanged rows
+ before ``min_updated_pos`` with the older blob file. This anchor is
+ the same in every blob column being updated.
+
Args:
original_data: Original data from the file (may be None if no
columns need to be read)
update_data: Update data (may contain only partial rows)
@@ -283,7 +299,8 @@ class TableUpdateByRowId:
first_row_id: The first_row_id of this file group
Returns:
- Normal merged PyArrow Table and blob values to write row-by-row.
+ Normal merged PyArrow Table, and per-blob-column values list. All
+ blob value lists have the same length (= ``max_updated_pos + 1``).
"""
# Get the _ROW_ID values from update_data to determine which rows are
being updated
@@ -298,7 +315,7 @@ class TableUpdateByRowId:
# Build the merged table column by column
merged_columns = {}
- blob_columns = {}
+ blob_columns: Dict[str, List[object]] = {}
update_by_col = {
col_name: update_data[col_name].combine_chunks()
for col_name in column_names
@@ -307,33 +324,36 @@ class TableUpdateByRowId:
int(relative_index.as_py()): idx
for idx, relative_index in enumerate(relative_indices)
}
+ # Caller (_write_by_first_row_id) only enters this method with a
+ # non-empty group, so update_positions is non-empty here.
+ blob_row_count = max(update_positions) + 1
for col_name in column_names:
update_col = update_by_col[col_name]
- original_col = original_data[col_name].combine_chunks()
- if update_col.type != original_col.type:
- update_col = self._coerce_column(
- update_col, original_col.type)
if self._is_blob_column(col_name):
blob_columns[col_name] = [
update_col[update_positions[i]].as_py()
if i in update_positions
else Blob.PLACE_HOLDER
- for i in range(original_data.num_rows)
+ for i in range(blob_row_count)
]
- else:
- try:
- merged_columns[col_name] = pc.replace_with_mask(
- original_col, mask, update_col)
- except pa.lib.ArrowNotImplementedError:
- n = original_data.num_rows
- combined = pa.concat_arrays(
- [original_col, update_col])
- offset = len(original_col)
- indices = np.arange(n, dtype=np.int64)
- for orig_pos, upd_idx in update_positions.items():
- indices[orig_pos] = offset + upd_idx
- merged_columns[col_name] = combined.take(
- pa.array(indices))
+ continue
+ original_col = original_data[col_name].combine_chunks()
+ if update_col.type != original_col.type:
+ update_col = self._coerce_column(
+ update_col, original_col.type)
+ try:
+ merged_columns[col_name] = pc.replace_with_mask(
+ original_col, mask, update_col)
+ except pa.lib.ArrowNotImplementedError:
+ n = original_data.num_rows
+ combined = pa.concat_arrays(
+ [original_col, update_col])
+ offset = len(original_col)
+ indices = np.arange(n, dtype=np.int64)
+ for orig_pos, upd_idx in update_positions.items():
+ indices[orig_pos] = offset + upd_idx
+ merged_columns[col_name] = combined.take(
+ pa.array(indices))
merged_table = pa.table(merged_columns) if merged_columns else None
@@ -419,7 +439,7 @@ class TableUpdateByRowId:
if new_files:
self._assign_update_file_metadata(
- new_files, first_row_id, column_names,
original_data.num_rows)
+ new_files, first_row_id, column_names, blob_columns)
self.commit_messages.append(
CommitMessage(
partition=partition_tuple,
@@ -436,8 +456,15 @@ class TableUpdateByRowId:
@staticmethod
def _assign_update_file_metadata(new_files: List[DataFileMeta],
first_row_id: int,
- column_names: List[str],
expected_row_count: int):
- blob_end = first_row_id + expected_row_count
+ column_names: List[str],
+ blob_columns: Dict[str, List[object]]):
+ # All blob columns share the same anchored span (see
+ # _merge_update_with_original docstring), so any column's length is
+ # the per-blob delta-file row count.
+ blob_row_count = (
+ len(next(iter(blob_columns.values()))) if blob_columns else 0
+ )
+ blob_end = first_row_id + blob_row_count
blob_starts = {}
# BlobWriter.prepare_commit preserves write/rolling order, which is
required
# for assigning continuous row-id ranges to rolled blob files.
