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 6944e9ef9b [python][ray] Flatten large self-merge predicates (#9361)
6944e9ef9b is described below
commit 6944e9ef9b9100069f7319efd199e6c981348e17
Author: XiaoHongbo <[email protected]>
AuthorDate: Sun Aug 23 16:04:04 2026 +0800
[python][ray] Flatten large self-merge predicates (#9361)
---
paimon-python/pypaimon/common/predicate.py | 27 +++++++++---
paimon-python/pypaimon/ray/merge_condition.py | 25 +++++++----
paimon-python/pypaimon/read/push_down_utils.py | 24 ++++++++---
paimon-python/pypaimon/read/split_read.py | 13 +++++-
paimon-python/pypaimon/tests/predicates_test.py | 20 +++++++++
.../tests/projection_predicate_index_test.py | 12 ++++++
.../tests/ray_data_evolution_merge_into_test.py | 49 ++++++++++++++++++++++
.../pypaimon/tests/reader_append_only_test.py | 20 +++++++++
8 files changed, 168 insertions(+), 22 deletions(-)
diff --git a/paimon-python/pypaimon/common/predicate.py
b/paimon-python/pypaimon/common/predicate.py
index 0bd1edd774..d93bbedfea 100644
--- a/paimon-python/pypaimon/common/predicate.py
+++ b/paimon-python/pypaimon/common/predicate.py
@@ -18,7 +18,6 @@
import re
from abc import ABC, ABCMeta, abstractmethod
from dataclasses import dataclass
-from functools import reduce
from typing import Any, Dict, List, Optional
from typing import ClassVar
@@ -30,6 +29,20 @@ from pypaimon.manifest.schema.simple_stats import SimpleStats
from pypaimon.table.row.internal_row import InternalRow
+def _combine_arrow_expressions(expressions, combine):
+ while len(expressions) > 1:
+ next_level = []
+ for index in range(0, len(expressions), 2):
+ if index + 1 == len(expressions):
+ next_level.append(expressions[index])
+ else:
+ next_level.append(combine(
+ expressions[index], expressions[index + 1],
+ ))
+ expressions = next_level
+ return expressions[0]
+
+
@dataclass
class Predicate:
method: str
@@ -113,11 +126,15 @@ class Predicate:
def to_arrow(self) -> Any:
if self.method == 'and':
- return reduce(lambda x, y: x & y,
- [p.to_arrow() for p in self.literals])
+ return _combine_arrow_expressions(
+ [p.to_arrow() for p in self.literals],
+ lambda left, right: left & right,
+ )
if self.method == 'or':
- return reduce(lambda x, y: x | y,
- [p.to_arrow() for p in self.literals])
+ return _combine_arrow_expressions(
+ [p.to_arrow() for p in self.literals],
+ lambda left, right: left | right,
+ )
if self.method == 'startsWith':
pattern = self.literals[0]
diff --git a/paimon-python/pypaimon/ray/merge_condition.py
b/paimon-python/pypaimon/ray/merge_condition.py
index 291c9f20d5..cfbb2ada80 100644
--- a/paimon-python/pypaimon/ray/merge_condition.py
+++ b/paimon-python/pypaimon/ray/merge_condition.py
@@ -162,15 +162,22 @@ def _to_paimon_predicate(expression, builder,
fields_by_name):
if kind == 'BinaryExpr':
op = node.op().upper()
if op in ('AND', 'OR'):
- left = _to_paimon_predicate(
- node.left(), builder, fields_by_name,
- )
- right = _to_paimon_predicate(
- node.right(), builder, fields_by_name,
- )
- if left is None or right is None:
- return None
- predicates = [left, right]
+ predicates = []
+ pending = [expression]
+ while pending:
+ current = pending.pop()
+ if current.variant_name() == 'BinaryExpr':
+ current_node = current.to_variant()
+ if current_node.op().upper() == op:
+ pending.append(current_node.right())
+ pending.append(current_node.left())
+ continue
+ predicate = _to_paimon_predicate(
+ current, builder, fields_by_name,
+ )
+ if predicate is None:
+ return None
+ predicates.append(predicate)
if op == 'AND':
return PredicateBuilder.and_predicates(predicates)
return PredicateBuilder.or_predicates(predicates)
diff --git a/paimon-python/pypaimon/read/push_down_utils.py
b/paimon-python/pypaimon/read/push_down_utils.py
index ebf4da3be5..1b096053e1 100644
--- a/paimon-python/pypaimon/read/push_down_utils.py
+++ b/paimon-python/pypaimon/read/push_down_utils.py
@@ -28,6 +28,10 @@ _UNSAFE_ARROW_FILTER_METHODS = frozenset([
'like',
])
+# Large boolean trees can overflow or crash native Dataset scanners even when
+# balanced. Keep complex predicates on Paimon's exact row-filter path instead.
+_MAX_ARROW_FILTER_LEAVES = 256
+
def extract_partition_spec_from_predicate(
predicate: Predicate, partition_keys: List[str]
@@ -149,12 +153,20 @@ def predicate_supports_arrow_filter(predicate:
Optional[Predicate]) -> bool:
"""
if predicate is None:
return True
- if predicate.method == 'and' or predicate.method == 'or':
- return all(
- predicate_supports_arrow_filter(p)
- for p in (predicate.literals or [])
- )
- return predicate.method not in _UNSAFE_ARROW_FILTER_METHODS
+
+ leaves = 0
+ pending = [predicate]
+ while pending:
+ current = pending.pop()
+ if current.method == 'and' or current.method == 'or':
+ pending.extend(current.literals or [])
+ continue
+ if current.method in _UNSAFE_ARROW_FILTER_METHODS:
+ return False
+ leaves += 1
+ if leaves > _MAX_ARROW_FILTER_LEAVES:
+ return False
+ return True
def remove_row_id_filter(predicate: Predicate) -> Optional[Predicate]:
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index 1085802dd5..b647e713c4 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -34,6 +34,7 @@ from pypaimon.read.interval_partition import
IntervalPartition, SortedRun
from pypaimon.read.partition_info import PartitionInfo
from pypaimon.read.push_down_utils import (
predicate_field_names,
+ predicate_supports_arrow_filter,
rewrite_predicate_indices,
trim_predicate_by_fields,
)
@@ -143,6 +144,8 @@ class SplitRead(ABC):
self.table: FileStoreTable = table
self.predicate = predicate
self.push_down_predicate = self._push_down_predicate()
+ self._arrow_filter_pushdown_enabled = predicate_supports_arrow_filter(
+ self.push_down_predicate)
self.split = split
self.row_tracking_enabled = row_tracking_enabled
self.value_arity = len(read_type)
@@ -558,7 +561,11 @@ class SplitRead(ABC):
if _is_reachable(read_field)
]
read_predicate =
trim_predicate_by_fields(self.push_down_predicate, read_file_fields)
- read_arrow_predicate = read_predicate.to_arrow() if read_predicate
else None
+ read_arrow_predicate = (
+ read_predicate.to_arrow()
+ if read_predicate and self._arrow_filter_pushdown_enabled
+ else None
+ )
self.schema_id_2_fields[key] = (
read_file_fields,
read_arrow_predicate,
@@ -855,7 +862,9 @@ class RawFileSplitRead(SplitRead):
blob_field_indices=blob_field_indices(self.read_fields),
vector_field_indices=vector_field_indices(self.read_fields))
reader = concat_reader
- if self.table.is_primary_key_table and self.predicate_for_reader:
+ if (self.predicate_for_reader
+ and (self.table.is_primary_key_table
+ or not self._arrow_filter_pushdown_enabled)):
reader = FilterRecordBatchReader(
reader,
self.predicate_for_reader,
diff --git a/paimon-python/pypaimon/tests/predicates_test.py
b/paimon-python/pypaimon/tests/predicates_test.py
index ff98be7937..918c95c578 100644
--- a/paimon-python/pypaimon/tests/predicates_test.py
+++ b/paimon-python/pypaimon/tests/predicates_test.py
@@ -49,6 +49,26 @@ def _random_format():
class PredicateTest(unittest.TestCase):
+ def test_large_boolean_arrow_expression_is_balanced(self):
+ class Expression:
+ def __init__(self, depth=1):
+ self.depth = depth
+
+ def __or__(self, other):
+ return Expression(max(self.depth, other.depth) + 1)
+
+ predicates = []
+ for _ in range(2000):
+ predicate = Predicate('equal', 0, 'id', [1])
+ predicate.to_arrow = lambda: Expression()
+ predicates.append(predicate)
+
+ expression = Predicate(
+ 'or', None, None, predicates,
+ ).to_arrow()
+
+ self.assertLessEqual(expression.depth, 12)
+
@classmethod
def setUpClass(cls):
cls.tempdir = tempfile.mkdtemp()
diff --git a/paimon-python/pypaimon/tests/projection_predicate_index_test.py
b/paimon-python/pypaimon/tests/projection_predicate_index_test.py
index 2953456223..8016a1b6da 100644
--- a/paimon-python/pypaimon/tests/projection_predicate_index_test.py
+++ b/paimon-python/pypaimon/tests/projection_predicate_index_test.py
@@ -334,6 +334,18 @@ class RewritePredicateIndicesUnitTest(unittest.TestCase):
self.assertFalse(predicate_supports_arrow_filter(unsafe))
self.assertFalse(predicate_supports_arrow_filter(mixed))
+ def test_arrow_filter_support_rejects_large_boolean_expression(self):
+ from pypaimon.read.push_down_utils import
predicate_supports_arrow_filter
+
+ pb = self._build_predicate()
+ large = pb.or_predicates([
+ pb.equal('c', value) for value in range(257)
+ ])
+ large_in = pb.is_in('c', list(range(2000)))
+
+ self.assertFalse(predicate_supports_arrow_filter(large))
+ self.assertTrue(predicate_supports_arrow_filter(large_in))
+
def test_missing_first_row_id_materializes_null_row_ids(self):
from pypaimon.read.reader.data_file_batch_reader import
DataFileBatchReader
from pypaimon.table.special_fields import SpecialFields
diff --git a/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
b/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
index d75dbf4a39..e0a35e1740 100644
--- a/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
+++ b/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
@@ -3245,6 +3245,55 @@ class RayDataEvolutionMergeIntoTest(unittest.TestCase):
plan.table.table_schema.id,
)
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_large_or_condition_pushes_down_predicate(self):
+ target = self._create_table()
+ self._write(
+ target,
+ pa.Table.from_pydict(
+ {
+ 'id': pa.array([0, 1, 2], type=pa.int32()),
+ 'name': ['Alice', 'Alice', 'Alice'],
+ 'age': pa.array([10, 11, 12], type=pa.int32()),
+ },
+ schema=self.pa_schema,
+ ),
+ )
+ condition = "t.name = 'Alice' AND ({})".format(
+ ' OR '.join(
+ '(t.id = {} AND t.age = {})'.format(i, i + 10)
+ for i in range(2000)
+ )
+ )
+
+ result, plan = self._merge_and_capture_self_merge_plan(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[WhenMatched.update(
+ {'age': lit(99)}, condition=condition,
+ )],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(result['num_matched'], 3)
+ self.assertEqual(self._read_sorted(target)['age'], [99, 99, 99])
+ predicate = plan.predicate
+ self.assertEqual(predicate.method, 'and')
+ self.assertEqual(
+ (predicate.literals[0].field,
+ predicate.literals[0].literals),
+ ('name', ['Alice']),
+ )
+ large_or = predicate.literals[1]
+ self.assertEqual(large_or.method, 'or')
+ self.assertEqual(len(large_or.literals), 2000)
+ self.assertTrue(all(
+ child.method == 'and' and len(child.literals) == 2
+ for child in large_or.literals
+ ))
+
@unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
def test_self_merge_pushdown_handles_evolved_file_groups(self):
from pypaimon.schema.data_types import AtomicType
diff --git a/paimon-python/pypaimon/tests/reader_append_only_test.py
b/paimon-python/pypaimon/tests/reader_append_only_test.py
index 3988359ad9..480cd4ebbe 100644
--- a/paimon-python/pypaimon/tests/reader_append_only_test.py
+++ b/paimon-python/pypaimon/tests/reader_append_only_test.py
@@ -809,6 +809,26 @@ class AoReaderTest(unittest.TestCase):
])
self.assertEqual(actual.sort_by('user_id'), expected)
+ def test_ao_reader_with_large_filter(self):
+ schema = Schema.from_pyarrow_schema(self.pa_schema,
partition_keys=['dt'])
+ self.catalog.create_table('default.test_append_only_large_filter',
schema, False)
+ table = self.catalog.get_table('default.test_append_only_large_filter')
+ self._write_test_table(table)
+
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ predicate = predicate_builder.or_predicates([
+ predicate_builder.equal('user_id', value)
+ for value in list(range(100, 355)) + [2, 6]
+ ])
+ read_builder = table.new_read_builder().with_filter(predicate)
+
+ actual = self._read_test_table(read_builder).sort_by('user_id')
+ expected = pa.concat_tables([
+ self.expected.slice(1, 1),
+ self.expected.slice(5, 1),
+ ])
+ self.assertEqual(actual, expected)
+
def test_ao_reader_with_projection(self):
schema = Schema.from_pyarrow_schema(self.pa_schema,
partition_keys=['dt'])
self.catalog.create_table('default.test_append_only_projection',
schema, False)
