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 85096ccc5d [python] data-evolution: speed up split planning from
O(n^2) to O(n log n) (#8477)
85096ccc5d is described below
commit 85096ccc5d962ce65b77f79cc70a48c151747ab0
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Jul 6 17:51:11 2026 +0800
[python] data-evolution: speed up split planning from O(n^2) to O(n log n)
(#8477)
DataEvolutionSplitGenerator._split_by_row_id (split planning) grouped a
DE table's files by scanning every merged range per file -> O(n^2), the
only quadratic step in planning; planning is single-threaded, so every
DE read that plans splits pays it.
---
.../pypaimon/manifest/schema/data_file_meta.py | 6 ++
.../read/scanner/chunk_shuffle_split_generator.py | 26 ++---
.../read/scanner/data_evolution_split_generator.py | 24 +----
.../tests/data_evolution_split_generator_test.py | 116 +++++++++++++++++++++
.../scanner/chunk_shuffle_split_generator_test.py | 21 ++++
5 files changed, 156 insertions(+), 37 deletions(-)
diff --git a/paimon-python/pypaimon/manifest/schema/data_file_meta.py
b/paimon-python/pypaimon/manifest/schema/data_file_meta.py
index e01e097ce0..4cc9a3a935 100644
--- a/paimon-python/pypaimon/manifest/schema/data_file_meta.py
+++ b/paimon-python/pypaimon/manifest/schema/data_file_meta.py
@@ -60,6 +60,12 @@ class DataFileMeta:
return None
return Range(self.first_row_id, self.first_row_id + self.row_count - 1)
+ def non_null_row_id_range(self) -> Range:
+ """Row-id range, failing fast if first_row_id is null (mirrors Java
nonNullRowIdRange)."""
+ if self.first_row_id is None:
+ raise ValueError(f"First row id of '{self.file_name}' should not
be null.")
+ return Range(self.first_row_id, self.first_row_id + self.row_count - 1)
+
def get_creation_time(self) -> Optional[Timestamp]:
return self.creation_time
diff --git
a/paimon-python/pypaimon/read/scanner/chunk_shuffle_split_generator.py
b/paimon-python/pypaimon/read/scanner/chunk_shuffle_split_generator.py
index 504236e500..6493b2e6fb 100644
--- a/paimon-python/pypaimon/read/scanner/chunk_shuffle_split_generator.py
+++ b/paimon-python/pypaimon/read/scanner/chunk_shuffle_split_generator.py
@@ -29,6 +29,7 @@ from pypaimon.read.sliced_split import SlicedSplit
from pypaimon.read.split import DataSplit, Split
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.utils.range import Range
+from pypaimon.utils.range_helper import RangeHelper
def _null_safe_partition_key(partition_values) -> tuple:
@@ -353,25 +354,16 @@ class
DataEvolutionChunkShuffleSplitGenerator(ChunkShuffleSplitGeneratorBase):
also returns the merged row_id range per group, which the chunk
slicer needs to drive row-count accumulation.
"""
- list_ranges = []
for f in files:
- file_range = f.row_id_range()
- if file_range is None:
+ if f.row_id_range() is None:
raise ValueError(
"chunk_shuffle for data evolution tables requires row
tracking; "
f"file {f.file_name} is missing first_row_id"
)
- list_ranges.append(file_range)
- if not list_ranges:
- return []
- sorted_ranges = Range.sort_and_merge_overlap(list_ranges, True, False)
-
- range_to_files: "dict[Range, List[DataFileMeta]]" = {}
- for f in files:
- file_range = f.row_id_range()
- for r in sorted_ranges:
- if r.overlaps(file_range):
- range_to_files.setdefault(r, []).append(f)
- break
-
- return sorted(range_to_files.items(), key=lambda kv: kv[0].from_)
+ groups = RangeHelper(lambda f:
f.row_id_range()).merge_overlapping_ranges(files)
+ result = []
+ for group in groups:
+ ranges = [f.row_id_range() for f in group]
+ merged = Range(min(r.from_ for r in ranges), max(r.to for r in
ranges))
+ result.append((merged, group))
+ return sorted(result, key=lambda kv: kv[0].from_)
diff --git
a/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
b/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
index 7886e93234..60a8d9c700 100644
--- a/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
+++ b/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
@@ -20,6 +20,7 @@ from typing import List, Optional, Tuple
from pypaimon.globalindex.indexed_split import IndexedSplit
from pypaimon.utils.range import Range
+from pypaimon.utils.range_helper import RangeHelper
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
from pypaimon.manifest.schema.manifest_entry import ManifestEntry
from pypaimon.read.scanner.split_generator import AbstractSplitGenerator
@@ -296,26 +297,9 @@ class DataEvolutionSplitGenerator(AbstractSplitGenerator):
@staticmethod
def _split_by_row_id(files: List[DataFileMeta]) ->
List[List[DataFileMeta]]:
- """
- Split files by row ID for data evolution tables.
- Files are grouped by their overlapping row ID ranges.
- """
- list_ranges = [file.row_id_range() for file in files]
-
- if not list_ranges:
- return []
-
- sorted_ranges = Range.sort_and_merge_overlap(list_ranges, True, False)
-
- range_to_files = {}
- for file in files:
- file_range = file.row_id_range()
- for r in sorted_ranges:
- if r.overlaps(file_range):
- range_to_files.setdefault(r, []).append(file)
- break
-
- return list(range_to_files.values())
+ """Group data-evolution files with overlapping row-id ranges (an
original file and its
+ column deltas) via the shared RangeHelper; fails fast on a file
missing first_row_id."""
+ return RangeHelper(lambda f:
f.non_null_row_id_range()).merge_overlapping_ranges(files)
def _wrap_to_indexed_splits(self, splits: List[Split], row_ranges:
List[Range]) -> List[Split]:
"""
diff --git
a/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py
b/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py
new file mode 100644
index 0000000000..dbb0a756cc
--- /dev/null
+++ b/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py
@@ -0,0 +1,116 @@
+# 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 random
+import unittest
+
+from pypaimon.read.scanner.data_evolution_split_generator import
DataEvolutionSplitGenerator
+from pypaimon.utils.range import Range
+
+
+class _F:
+ def __init__(self, tag: int, from_: int = None, to: int = None):
+ self.tag = tag
+ self.file_name = f"f{tag}"
+ self._range = Range(from_, to) if from_ is not None else None
+
+ def row_id_range(self):
+ return self._range
+
+ def non_null_row_id_range(self) -> Range:
+ if self._range is None:
+ raise ValueError(f"First row id of '{self.file_name}' should not
be null.")
+ return self._range
+
+
+def _reference_split(files):
+ """The original O(n^2) linear scan, kept to lock equivalence."""
+ list_ranges = [f.row_id_range() for f in files]
+ if not list_ranges:
+ return []
+ sorted_ranges = Range.sort_and_merge_overlap(list_ranges, True, False)
+ range_to_files = {}
+ for f in files:
+ file_range = f.row_id_range()
+ for r in sorted_ranges:
+ if r.overlaps(file_range):
+ range_to_files.setdefault(r, []).append(f)
+ break
+ return list(range_to_files.values())
+
+
+def _shape(groups):
+ """Group structure by file tag: locks grouping + order."""
+ return [[f.tag for f in g] for g in groups]
+
+
+def _grouping(groups):
+ """Files grouped together, ignoring order -- the functional invariant."""
+ return {frozenset(f.tag for f in g) for g in groups}
+
+
+class SplitByRowIdEquivalenceTest(unittest.TestCase):
+ def test_empty(self):
+ self.assertEqual(DataEvolutionSplitGenerator._split_by_row_id([]), [])
+
+ def test_raises_on_file_missing_first_row_id(self):
+ # A file without first_row_id must fail fast with a readable error.
+ with self.assertRaisesRegex(ValueError, "should not be null"):
+ DataEvolutionSplitGenerator._split_by_row_id([_F(0, 0, 4), _F(1)])
+
+ def test_disjoint_files_each_its_own_group(self):
+ files = [_F(0, 0, 4), _F(1, 5, 9), _F(2, 10, 14)]
+
self.assertEqual(_shape(DataEvolutionSplitGenerator._split_by_row_id(files)),
+ [[0], [1], [2]])
+
+ def test_evolution_delta_grouped_with_original(self):
+ original = _F(0, 0, 9)
+ delta = _F(1, 3, 5) # sub-range of original -> same merged range
+ groups = DataEvolutionSplitGenerator._split_by_row_id([original,
delta])
+ self.assertEqual(_shape(groups), [[0, 1]])
+
+ def test_groups_ordered_by_range_start(self):
+ files = [_F(0, 10, 14), _F(1, 0, 4), _F(2, 5, 9)] # unsorted input
+
self.assertEqual(_shape(DataEvolutionSplitGenerator._split_by_row_id(files)),
+ [[1], [2], [0]]) # groups come out ordered by range
start
+
+ def test_matches_reference_grouping_on_random_inputs(self):
+ rng = random.Random(1234)
+ for _ in range(1000):
+ n = rng.randint(0, 50)
+ files, cursor = [], 0
+ for tag in range(n):
+ roll = rng.random()
+ if roll < 0.6: # disjoint
+ from_ = cursor + rng.randint(1, 5)
+ to = from_ + rng.randint(0, 10)
+ elif roll < 0.85: # overlapping (evolution-like)
+ from_ = rng.randint(max(0, cursor - 8), max(0, cursor))
+ to = from_ + rng.randint(0, 6)
+ else: # duplicate / same start
+ from_ = rng.randint(0, cursor + 1)
+ to = from_ + rng.randint(0, 12)
+ cursor = max(cursor, to)
+ files.append(_F(tag, from_, to))
+ rng.shuffle(files)
+ self.assertEqual(
+ _grouping(DataEvolutionSplitGenerator._split_by_row_id(files)),
+ _grouping(_reference_split(files)))
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git
a/paimon-python/pypaimon/tests/scanner/chunk_shuffle_split_generator_test.py
b/paimon-python/pypaimon/tests/scanner/chunk_shuffle_split_generator_test.py
index be50cfa637..ce69096d7c 100644
--- a/paimon-python/pypaimon/tests/scanner/chunk_shuffle_split_generator_test.py
+++ b/paimon-python/pypaimon/tests/scanner/chunk_shuffle_split_generator_test.py
@@ -107,6 +107,15 @@ def _mock_de_entry(partition_values, bucket, file_name,
first_row_id, row_count,
return entry
+def _de_file(file_name, first_row_id, row_count):
+ """A DE file mock; row_id_range() is a real Range, or None when
first_row_id is None."""
+ file = Mock(spec=DataFileMeta)
+ file.file_name = file_name
+ rng = Range(first_row_id, first_row_id + row_count - 1) if first_row_id is
not None else None
+ file.row_id_range = lambda r=rng: r
+ return file
+
+
def _split_signature(split):
"""A stable, comparable identity for a split — what the worker would
actually read."""
if isinstance(split, SlicedSplit):
@@ -532,6 +541,18 @@ class DataEvolutionChunkShuffleAlgoTest(unittest.TestCase):
gen = _make_de_generator(seed=1, chunk_size=100)
self.assertEqual(gen.create_splits([]), [])
+ def test_split_by_row_id_with_range_groups_and_merges(self):
+ # a[0,4] & c[2,3] overlap -> one merged group [0,4]; b[10,14] ->
another. Input unsorted.
+ files = [_de_file("a", 0, 5), _de_file("b", 10, 5), _de_file("c", 2,
2)]
+ out =
DataEvolutionChunkShuffleSplitGenerator._split_by_row_id_with_range(files)
+ self.assertEqual([(r.from_, r.to) for r, _ in out], [(0, 4), (10, 14)])
+ self.assertEqual([[f.file_name for f in fs] for _, fs in out], [["a",
"c"], ["b"]])
+
+ def test_split_by_row_id_with_range_raises_on_missing_first_row_id(self):
+ files = [_de_file("a", 0, 5), _de_file("b", None, 5)]
+ with self.assertRaisesRegex(ValueError, "requires row tracking"):
+
DataEvolutionChunkShuffleSplitGenerator._split_by_row_id_with_range(files)
+
def test_full_aligned_groups_one_per_chunk(self):
# Three commits of 100 rows each → three aligned groups.
# chunk_size = 100 → 3 chunks, each holding one group whole.
