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 9956a067fc [python] optimize blob bunch construction (#8859)
9956a067fc is described below
commit 9956a067fc8f6959050383559d74aa61bfae8389
Author: Faiz <[email protected]>
AuthorDate: Mon Jul 27 15:51:31 2026 +0800
[python] optimize blob bunch construction (#8859)
---
paimon-python/pypaimon/read/reader/field_bunch.py | 28 +++--
paimon-python/pypaimon/read/split_read.py | 4 +-
paimon-python/pypaimon/tests/field_bunch_test.py | 130 ++++++++++++++++++++++
3 files changed, 151 insertions(+), 11 deletions(-)
diff --git a/paimon-python/pypaimon/read/reader/field_bunch.py
b/paimon-python/pypaimon/read/reader/field_bunch.py
index f2e0ae756c..5474bf855a 100644
--- a/paimon-python/pypaimon/read/reader/field_bunch.py
+++ b/paimon-python/pypaimon/read/reader/field_bunch.py
@@ -144,13 +144,24 @@ class _SpecialFieldBunch(FieldBunch):
class BlobBunch(_SpecialFieldBunch):
"""Files for partial field (blob files)."""
+ def __init__(self, expected_row_count: int, row_id_push_down: bool =
False):
+ super().__init__(expected_row_count, row_id_push_down)
+ self._finished = False
+
def add(self, file: DataFileMeta) -> None:
+ if self._finished:
+ raise ValueError("Cannot add a blob file to a finished blob
bunch.")
if not self._is_special_file(file.file_name):
raise ValueError("Only blob file can be added to a blob bunch.")
if self._files and file.write_cols != self._files[0].write_cols:
raise ValueError("All files in a blob bunch should have the same
write columns.")
self._files.append(file)
+
+ def finish(self) -> None:
+ if self._finished:
+ return
+
merged = Range.sort_and_merge_overlap(
[blob_file.row_id_range() for blob_file in self._files],
True,
@@ -162,22 +173,19 @@ class BlobBunch(_SpecialFieldBunch):
f"Blob files row count exceed the expect
{self.expected_row_count}"
)
- def row_count(self) -> int:
- merged = Range.sort_and_merge_overlap(
- [blob_file.row_id_range() for blob_file in self._files],
- True,
- True,
- )
- row_count = sum(row_range.count() for row_range in merged)
if not self.row_id_push_down:
if len(merged) != 1:
raise ValueError("Blob file bunch should always contain a
contiguous row range.")
- if self.expected_row_count >= 0 and row_count !=
self.expected_row_count:
+ if self.expected_row_count >= 0 and self._row_count !=
self.expected_row_count:
raise ValueError(
"The merged row count of blob file bunch should be aligned
"
- f"with normal files, expect {self.expected_row_count}, got
{row_count}."
+ f"with normal files, expect {self.expected_row_count}, got
{self._row_count}."
)
- return row_count
+ self._finished = True
+
+ def row_count(self) -> int:
+ self.finish()
+ return self._row_count
def sequential_read_optimize(self) -> bool:
if not self._files:
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index 71e6b623f4..77bbd44837 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -1386,7 +1386,9 @@ class DataEvolutionSplitRead(SplitRead):
fields_files.append(DataBunch(file))
row_count = file.row_count
- fields_files.extend(blob_bunch_map.values())
+ for bunch in blob_bunch_map.values():
+ bunch.finish()
+ fields_files.append(bunch)
fields_files.extend(vector_bunch_map.values())
return fields_files
diff --git a/paimon-python/pypaimon/tests/field_bunch_test.py
b/paimon-python/pypaimon/tests/field_bunch_test.py
new file mode 100644
index 0000000000..0b6772d284
--- /dev/null
+++ b/paimon-python/pypaimon/tests/field_bunch_test.py
@@ -0,0 +1,130 @@
+# 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 unittest
+from unittest import mock
+
+from pypaimon.read.reader.field_bunch import BlobBunch
+from pypaimon.utils.range import Range
+
+
+class _BlobFile:
+ def __init__(
+ self,
+ file_name: str,
+ first_row_id: int,
+ row_count: int,
+ write_cols=None):
+ self.file_name = file_name
+ self.first_row_id = first_row_id
+ self.row_count = row_count
+ self.write_cols = write_cols or ["blob_col"]
+
+ def row_id_range(self) -> Range:
+ return Range(
+ self.first_row_id,
+ self.first_row_id + self.row_count - 1,
+ )
+
+
+class BlobBunchTest(unittest.TestCase):
+
+ def test_finish_merges_ranges_once(self):
+ bunch = BlobBunch(expected_row_count=1000)
+ merge_ranges = Range.sort_and_merge_overlap
+
+ with mock.patch.object(
+ Range,
+ "sort_and_merge_overlap",
+ wraps=merge_ranges) as mocked_merge:
+ for index in range(1000):
+ bunch.add(_BlobFile(f"{index}.blob", index, 1))
+
+ self.assertEqual(0, mocked_merge.call_count)
+ bunch.finish()
+ self.assertEqual(1, mocked_merge.call_count)
+ self.assertEqual(1000, bunch.row_count())
+ self.assertEqual(1, mocked_merge.call_count)
+
+ def test_finish_preserves_overlapping_versions(self):
+ bunch = BlobBunch(expected_row_count=20)
+ files = [
+ _BlobFile("old.blob", 0, 10),
+ _BlobFile("new.blob", 0, 10),
+ _BlobFile("tail.blob", 10, 10),
+ ]
+ for file in files:
+ bunch.add(file)
+
+ bunch.finish()
+
+ self.assertEqual(files, bunch.files())
+ self.assertEqual(20, bunch.row_count())
+
+ def test_add_rejects_file_after_finish(self):
+ bunch = BlobBunch(expected_row_count=1)
+ bunch.add(_BlobFile("first.blob", 0, 1))
+ bunch.finish()
+
+ with self.assertRaisesRegex(ValueError, "finished blob bunch"):
+ bunch.add(_BlobFile("second.blob", 1, 1))
+
+ def test_finish_can_recover_after_validation_failure(self):
+ bunch = BlobBunch(expected_row_count=10)
+ bunch.add(_BlobFile("first.blob", 0, 5))
+ with self.assertRaisesRegex(ValueError, "aligned with normal files"):
+ bunch.finish()
+
+ bunch.add(_BlobFile("second.blob", 5, 5))
+
+ bunch.finish()
+ self.assertEqual(10, bunch.row_count())
+
+ def test_finish_rejects_gap_without_row_id_push_down(self):
+ bunch = BlobBunch(expected_row_count=20)
+ bunch.add(_BlobFile("left.blob", 0, 5))
+ bunch.add(_BlobFile("right.blob", 10, 10))
+
+ with self.assertRaisesRegex(ValueError, "contiguous row range"):
+ bunch.finish()
+
+ def test_finish_rejects_unaligned_count_without_row_id_push_down(self):
+ bunch = BlobBunch(expected_row_count=20)
+ bunch.add(_BlobFile("short.blob", 0, 10))
+
+ with self.assertRaisesRegex(ValueError, "aligned with normal files"):
+ bunch.finish()
+
+ def test_finish_allows_gap_with_row_id_push_down(self):
+ bunch = BlobBunch(expected_row_count=20, row_id_push_down=True)
+ bunch.add(_BlobFile("left.blob", 0, 5))
+ bunch.add(_BlobFile("right.blob", 10, 10))
+
+ bunch.finish()
+
+ self.assertEqual(15, bunch.row_count())
+
+ def test_finish_rejects_row_count_exceeding_expected(self):
+ bunch = BlobBunch(expected_row_count=10, row_id_push_down=True)
+ bunch.add(_BlobFile("too-large.blob", 0, 11))
+
+ with self.assertRaisesRegex(ValueError, "row count exceed"):
+ bunch.finish()
+
+
+if __name__ == "__main__":
+ unittest.main()
