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 fe51b14070 [python][ray] Push down self-merge matched conditions
(#9314)
fe51b14070 is described below
commit fe51b14070a06ae7836f7eb60910ea8ac6302c24
Author: XiaoHongbo <[email protected]>
AuthorDate: Fri Aug 21 08:52:26 2026 +0800
[python][ray] Push down self-merge matched conditions (#9314)
---
.../pypaimon/ray/data_evolution_merge_into.py | 22 +
.../pypaimon/ray/data_evolution_merge_join.py | 26 +
paimon-python/pypaimon/ray/merge_condition.py | 162 +++++-
paimon-python/pypaimon/ray/ray_paimon.py | 2 +
.../pypaimon/read/datasource/split_provider.py | 7 +-
.../tests/ray_data_evolution_merge_into_test.py | 647 +++++++++++++++++++++
6 files changed, 864 insertions(+), 2 deletions(-)
diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_into.py
b/paimon-python/pypaimon/ray/data_evolution_merge_into.py
index b7ba555822..8695c07d77 100644
--- a/paimon-python/pypaimon/ray/data_evolution_merge_into.py
+++ b/paimon-python/pypaimon/ray/data_evolution_merge_into.py
@@ -24,6 +24,7 @@ from typing import Any, Dict, List, Mapping, Optional,
Sequence, Tuple
import pyarrow as pa
+from pypaimon.common.predicate import Predicate
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
from pypaimon.ray.data_evolution_merge_join import (
_resolve_source_projection,
@@ -63,6 +64,7 @@ class _PrepareCtx:
full_pa_schema: pa.Schema
catalog_options: Dict[str, str]
is_self_merge: bool = False
+ self_merge_scan_predicate: Optional[Predicate] = None
def merge_into(
@@ -251,6 +253,23 @@ def _prepare(target, source, catalog_options,
when_matched, when_not_matched, on
update_pa_schema = pa.schema(
[full_pa_schema.field(c) for c in settable_field_names]
)
+ self_merge_scan_predicate = None
+ if (is_self_merge and matched_specs
+ and all(c.condition is not None for c in matched_specs)):
+ from pypaimon.common.predicate_builder import PredicateBuilder
+ from pypaimon.ray.merge_condition import (
+ try_parse_self_merge_predicate,
+ )
+ predicates = [
+ try_parse_self_merge_predicate(
+ c.condition, table.table_schema.fields,
+ )
+ for c in matched_specs
+ ]
+ if all(predicate is not None for predicate in predicates):
+ self_merge_scan_predicate = PredicateBuilder.or_predicates(
+ predicates
+ )
ctx = _PrepareCtx(
target_on_cols=target_on_cols,
source_on_cols=source_on_cols,
@@ -260,6 +279,7 @@ def _prepare(target, source, catalog_options, when_matched,
when_not_matched, on
full_pa_schema=full_pa_schema,
catalog_options=catalog_options,
is_self_merge=is_self_merge,
+ self_merge_scan_predicate=self_merge_scan_predicate,
)
return table, source_ds, matched_specs, not_matched_specs, ctx
@@ -300,6 +320,7 @@ def _build_datasets(
catalog_options=ctx.catalog_options,
resolve_target_projection=_resolve_target_projection,
snapshot_id=base_snapshot_id,
+ scan_predicate=ctx.self_merge_scan_predicate,
ray_remote_args=ray_remote_args,
)
if any(c.delete for c in matched_specs):
@@ -310,6 +331,7 @@ def _build_datasets(
catalog_options=ctx.catalog_options,
resolve_target_projection=_resolve_target_projection,
snapshot_id=base_snapshot_id,
+ scan_predicate=ctx.self_merge_scan_predicate,
ray_remote_args=ray_remote_args,
)
return update_ds, delete_ds, insert_ds, update_cols_union
diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
index 4a0dd74b35..863dfd5984 100644
--- a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
+++ b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
@@ -188,6 +188,7 @@ def build_self_merge_update_ds(
catalog_options: Dict[str, str],
resolve_target_projection,
snapshot_id: Optional[int] = None,
+ scan_predicate=None,
ray_remote_args: Optional[Dict[str, Any]] = None,
) -> Tuple:
from pypaimon.ray.ray_paimon import read_paimon
@@ -213,9 +214,21 @@ def build_self_merge_update_ds(
c for c in target_field_names if c in needed_cols
]
+ read_kwargs = {}
+ if scan_predicate is not None:
+ from pypaimon.common.options.core_options import (
+ CoreOptions, GlobalIndexSearchMode,
+ )
+ read_kwargs["filter"] = scan_predicate
+ read_kwargs["dynamic_options"] = {
+ CoreOptions.SCALAR_INDEX_SEARCH_MODE.key():
+ GlobalIndexSearchMode.FULL.value,
+ }
target_ds = read_paimon(
target_identifier, catalog_options,
projection=projection, snapshot_id=snapshot_id,
+ _preserve_current_schema=True,
+ **read_kwargs,
)
update_schema = build_update_schema(target_pa_schema, update_cols,
row_id_name)
@@ -260,6 +273,7 @@ def build_self_merge_delete_ds(
catalog_options: Dict[str, str],
resolve_target_projection,
snapshot_id: Optional[int] = None,
+ scan_predicate=None,
ray_remote_args: Optional[Dict[str, Any]] = None,
) -> Tuple:
from pypaimon.ray.ray_paimon import read_paimon
@@ -281,9 +295,21 @@ def build_self_merge_delete_ds(
c for c in target_field_names if c in needed_cols
]
+ read_kwargs = {}
+ if scan_predicate is not None:
+ from pypaimon.common.options.core_options import (
+ CoreOptions, GlobalIndexSearchMode,
+ )
+ read_kwargs["filter"] = scan_predicate
+ read_kwargs["dynamic_options"] = {
+ CoreOptions.SCALAR_INDEX_SEARCH_MODE.key():
+ GlobalIndexSearchMode.FULL.value,
+ }
target_ds = read_paimon(
target_identifier, catalog_options,
projection=projection, snapshot_id=snapshot_id,
+ _preserve_current_schema=True,
+ **read_kwargs,
)
delete_schema = build_delete_schema(row_id_name)
diff --git a/paimon-python/pypaimon/ray/merge_condition.py
b/paimon-python/pypaimon/ray/merge_condition.py
index 12710898d7..291c9f20d5 100644
--- a/paimon-python/pypaimon/ray/merge_condition.py
+++ b/paimon-python/pypaimon/ray/merge_condition.py
@@ -16,13 +16,19 @@
# limitations under the License.
################################################################################
+import logging
import re
-from typing import Mapping, Set
+from typing import Mapping, Optional, Set
import pyarrow as pa
+from pypaimon.common.predicate import Predicate
+from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.schema.data_types import AtomicType
+
_COL_REF_PATTERN = re.compile(r'\b([st])\.(\w+)\b')
+logger = logging.getLogger(__name__)
def _load_datafusion():
@@ -103,3 +109,157 @@ def extract_target_columns(condition: str) -> Set[str]:
stripped = _strip_string_literals(condition)
return {m.group(2) for m in _COL_REF_PATTERN.finditer(stripped)
if m.group(1) == "t"}
+
+
+def try_parse_self_merge_predicate(condition, fields) -> Optional[Predicate]:
+ """Best-effort conversion of a self-merge condition for scan pruning."""
+ if not isinstance(condition, str):
+ return None
+
+ # Conversion failure keeps the original unfiltered DataFusion execution.
+ try:
+ expression = _parse_self_merge_expression(condition, fields)
+ return _to_paimon_predicate(
+ expression,
+ PredicateBuilder(fields),
+ {field.name: field for field in fields},
+ )
+ except Exception:
+ logger.debug(
+ "Unable to push down self-merge condition %r",
+ condition,
+ exc_info=True,
+ )
+ return None
+
+
+def _parse_self_merge_expression(condition, fields):
+ from pypaimon.schema.data_types import PyarrowFieldParser
+ from pypaimon.table.special_fields import SpecialFields
+
+ pa_schema = PyarrowFieldParser.from_paimon_schema(fields)
+ arrays, names = [], []
+ for alias in ('s', 't'):
+ for field in pa_schema:
+ arrays.append(pa.array([], type=field.type))
+ names.append('{}.{}'.format(alias, field.name))
+ arrays.append(pa.array([], type=pa.int64()))
+ names.append('{}.{}'.format(alias, SpecialFields.ROW_ID.name))
+
+ batch = pa.RecordBatch.from_arrays(arrays, names)
+ context = _load_datafusion().SessionContext()
+ context.register_record_batches('_self_merge', [[batch]])
+ df_schema = context.table(
+ '_self_merge'
+ ).logical_plan().to_variant().schema()
+ return context.parse_sql_expr(rewrite_condition(condition), df_schema)
+
+
+def _to_paimon_predicate(expression, builder, fields_by_name):
+ kind = expression.variant_name()
+ node = expression.to_variant()
+
+ 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]
+ if op == 'AND':
+ return PredicateBuilder.and_predicates(predicates)
+ return PredicateBuilder.or_predicates(predicates)
+ return _comparison_predicate(node, builder, fields_by_name)
+
+ if kind == 'InList':
+ field = _datafusion_field(node.expr(), fields_by_name)
+ literals = [_datafusion_literal(item) for item in node.list()]
+ if (field is None or any(not found for found, _ in literals)
+ or not _safe_literals(field, [v for _, v in literals])):
+ return None
+ values = [value for _, value in literals]
+ if node.negated():
+ return builder.is_not_in(field.name, values)
+ return builder.is_in(field.name, values)
+
+ if kind == 'Between':
+ field = _datafusion_field(node.expr(), fields_by_name)
+ low_found, low = _datafusion_literal(node.low())
+ high_found, high = _datafusion_literal(node.high())
+ if (field is None or not low_found or not high_found
+ or not _safe_literals(field, [low, high])):
+ return None
+ if node.negated():
+ return builder.not_between(field.name, low, high)
+ return builder.between(field.name, low, high)
+
+ if kind in ('IsNull', 'IsNotNull'):
+ field = _datafusion_field(node.expr(), fields_by_name)
+ if field is None or not isinstance(field.type, AtomicType):
+ return None
+ if kind == 'IsNull':
+ return builder.is_null(field.name)
+ return builder.is_not_null(field.name)
+
+ return None
+
+
+def _comparison_predicate(node, builder, fields_by_name):
+ field = _datafusion_field(node.left(), fields_by_name)
+ found, literal = _datafusion_literal(node.right())
+ if field is None or not found or not _safe_literals(field, [literal]):
+ return None
+
+ methods = {
+ '=': builder.equal,
+ '!=': builder.not_equal,
+ '<': builder.less_than,
+ '<=': builder.less_or_equal,
+ '>': builder.greater_than,
+ '>=': builder.greater_or_equal,
+ }
+ method = methods.get(node.op())
+ if method is None:
+ return None
+ return method(field.name, literal)
+
+
+def _datafusion_field(expression, fields_by_name):
+ if expression.variant_name() != 'Column':
+ return None
+ name = expression.to_variant().name()
+ if not (name.startswith('s.') or name.startswith('t.')):
+ return None
+ return fields_by_name.get(name[2:])
+
+
+def _datafusion_literal(expression):
+ if expression.variant_name() != 'Literal':
+ return False, None
+ value = expression.python_value()
+ if isinstance(value, pa.Scalar):
+ value = value.as_py()
+ return True, value
+
+
+def _safe_literals(field, literals) -> bool:
+ if not isinstance(field.type, AtomicType):
+ return False
+ type_name = field.type.type.upper().split('(', 1)[0].strip()
+ if type_name in {'TINYINT', 'SMALLINT', 'INT', 'INTEGER', 'BIGINT'}:
+ return all(
+ isinstance(literal, int)
+ and not isinstance(literal, bool)
+ and -(1 << 63) <= literal <= (1 << 63) - 1
+ for literal in literals
+ )
+ if type_name == 'BOOLEAN':
+ return all(isinstance(literal, bool) for literal in literals)
+ if type_name in {'STRING', 'CHAR', 'VARCHAR'}:
+ return all(isinstance(literal, str) for literal in literals)
+ return False
diff --git a/paimon-python/pypaimon/ray/ray_paimon.py
b/paimon-python/pypaimon/ray/ray_paimon.py
index 78309200ac..2a5bbf0a26 100644
--- a/paimon-python/pypaimon/ray/ray_paimon.py
+++ b/paimon-python/pypaimon/ray/ray_paimon.py
@@ -60,6 +60,7 @@ def read_paimon(
ray_remote_args: Optional[Dict[str, Any]] = None,
concurrency: Optional[int] = None,
override_num_blocks: Optional[int] = None,
+ _preserve_current_schema: bool = False,
**read_args,
) -> "ray.data.Dataset":
"""Read a Paimon table into a Ray Dataset.
@@ -109,6 +110,7 @@ def read_paimon(
snapshot_id=snapshot_id,
tag_name=tag_name,
dynamic_options=dynamic_options,
+ preserve_current_schema=_preserve_current_schema,
)
if not split_provider.splits():
diff --git a/paimon-python/pypaimon/read/datasource/split_provider.py
b/paimon-python/pypaimon/read/datasource/split_provider.py
index eb953c573e..6060ccce80 100644
--- a/paimon-python/pypaimon/read/datasource/split_provider.py
+++ b/paimon-python/pypaimon/read/datasource/split_provider.py
@@ -97,6 +97,7 @@ class CatalogSplitProvider(SplitProvider):
snapshot_id: Optional[int] = None,
tag_name: Optional[str] = None,
dynamic_options: Optional[Dict[str, str]] = None,
+ preserve_current_schema: bool = False,
):
if not table_identifier:
raise ValueError("table_identifier is required")
@@ -132,6 +133,7 @@ class CatalogSplitProvider(SplitProvider):
self._snapshot_id = snapshot_id
self._tag_name = tag_name
self._dynamic_options = dynamic_options
+ self._preserve_current_schema = preserve_current_schema
self._table_cached = None
self._splits_cached = None
self._read_type_cached = None
@@ -150,7 +152,10 @@ class CatalogSplitProvider(SplitProvider):
if self._dynamic_options:
dynamic_options.update(self._dynamic_options)
if dynamic_options:
- table = table.copy(dynamic_options)
+ if self._preserve_current_schema:
+ table = table.copy_without_time_travel(dynamic_options)
+ else:
+ table = table.copy(dynamic_options)
self._table_cached = table
return self._table_cached
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 2e1e8b8e44..70a0a0d189 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
@@ -16,6 +16,7 @@
# limitations under the License.
################################################################################
+import datetime
import os
import shutil
import tempfile
@@ -2036,6 +2037,489 @@ class RayDataEvolutionMergeIntoTest(unittest.TestCase):
self.assertEqual(out['age'], [99, 99, 99])
self.assertEqual(out['name'], ['a', 'b', 'c'])
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_condition_pushes_down_predicate(self):
+ from pypaimon.common.options.core_options import (
+ CoreOptions, GlobalIndexSearchMode,
+ )
+ from pypaimon.ray.ray_paimon import read_paimon as real_read_paimon
+
+ target = self._create_table()
+ self._write(
+ target,
+ pa.Table.from_pydict(
+ {
+ 'id': pa.array([1, 2, 3], type=pa.int32()),
+ 'name': ['a', 'b', 'c'],
+ 'age': pa.array([10, 20, 30], type=pa.int32()),
+ },
+ schema=self.pa_schema,
+ ),
+ )
+ with patch(
+ 'pypaimon.ray.ray_paimon.read_paimon',
+ wraps=real_read_paimon,
+ ) as mock_read:
+ result = merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[WhenMatched.update(
+ {'age': lit(99)}, condition='t.id IN (1, 3)',
+ )],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(result['num_matched'], 2)
+ self.assertEqual(
+ self._read_sorted(target),
+ {
+ 'id': [1, 2, 3],
+ 'name': ['a', 'b', 'c'],
+ 'age': [99, 20, 99],
+ },
+ )
+ read_kwargs = mock_read.call_args[1]
+ predicate = read_kwargs['filter']
+ self.assertEqual(predicate.method, 'in')
+ self.assertEqual(predicate.field, 'id')
+ self.assertEqual(predicate.literals, [1, 3])
+ self.assertEqual(
+ read_kwargs['dynamic_options'][
+ CoreOptions.SCALAR_INDEX_SEARCH_MODE.key()
+ ],
+ GlobalIndexSearchMode.FULL.value,
+ )
+ self.assertTrue(read_kwargs['_preserve_current_schema'])
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_pushdown_handles_evolved_file_groups(self):
+ from pypaimon.schema.data_types import AtomicType
+ from pypaimon.schema.schema_change import SchemaChange
+
+ options = dict(self.de_options)
+ options.update({
+ 'global-index.enabled': 'true',
+ 'bucket': '-1',
+ })
+ target = self._create_table(options=options)
+ self._write(
+ target,
+ pa.Table.from_pydict(
+ {
+ 'id': pa.array([1, 2], type=pa.int32()),
+ 'name': ['a', 'b'],
+ 'age': pa.array([10, 20], type=pa.int32()),
+ },
+ schema=self.pa_schema,
+ ),
+ )
+
+ table = self.catalog.get_table(target)
+ self.assertGreater(table.create_global_index('id'), 0)
+
+ # This append is intentionally not covered by the existing index.
+ self._write(
+ target,
+ pa.Table.from_pydict(
+ {
+ 'id': pa.array([3, 4], type=pa.int32()),
+ 'name': ['c', 'd'],
+ 'age': pa.array([30, 40], type=pa.int32()),
+ },
+ schema=self.pa_schema,
+ ),
+ )
+
+ from pypaimon.index.index_file_handler import IndexFileHandler
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ indexed_ranges = {
+ (
+ entry.index_file.global_index_meta.row_range_start,
+ entry.index_file.global_index_meta.row_range_end,
+ )
+ for entry in IndexFileHandler(table).scan(snapshot)
+ }
+ self.assertEqual({(0, 1)}, indexed_ranges)
+
+ first_result = merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[WhenMatched.update(
+ {'age': lit(99)}, condition='t.id IN (2, 3)',
+ )],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+ self.assertEqual(first_result['num_matched'], 2)
+
+ self.catalog.alter_table(
+ target,
+ [SchemaChange.add_column('note', AtomicType('STRING'))],
+ False,
+ )
+
+ result = merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[WhenMatched.update(
+ {'name': lit('updated')},
+ condition='t.id IN (1, 4) AND t.note IS NULL',
+ )],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(result['num_matched'], 2)
+ self.assertEqual(
+ self._read_sorted(target),
+ {
+ 'id': [1, 2, 3, 4],
+ 'name': ['updated', 'b', 'c', 'updated'],
+ 'age': [10, 99, 99, 40],
+ 'note': [None, None, None, None],
+ },
+ )
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_delete_condition_pushes_down_predicate(self):
+ from pypaimon.ray.ray_paimon import read_paimon as real_read_paimon
+
+ options = dict(self.de_options)
+ options['deletion-vectors.enabled'] = 'true'
+ target = self._create_table(options=options)
+ self._write(target, self._source(ids=(1, 2, 3)))
+
+ with patch(
+ 'pypaimon.ray.ray_paimon.read_paimon',
+ wraps=real_read_paimon,
+ ) as mock_read:
+ result = merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[WhenMatched.delete(condition='s.id = 2')],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(result['num_matched'], 1)
+ self.assertEqual(self._read_sorted(target)['id'], [1, 3])
+ predicate = mock_read.call_args[1]['filter']
+ self.assertEqual((predicate.method, predicate.field,
predicate.literals),
+ ('equal', 'id', [2]))
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_multiple_conditions_push_down_or(self):
+ from pypaimon.ray.ray_paimon import read_paimon as real_read_paimon
+
+ target = self._create_table()
+ self._write(target, self._source(ids=(1, 2, 3)))
+
+ with patch(
+ 'pypaimon.ray.ray_paimon.read_paimon',
+ wraps=real_read_paimon,
+ ) as mock_read:
+ result = merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[
+ WhenMatched.update(
+ {'age': lit(11)}, condition='t.id = 1',
+ ),
+ WhenMatched.update(
+ {'age': lit(33)}, condition='s.id = 3',
+ ),
+ ],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(result['num_matched'], 2)
+ predicate = mock_read.call_args[1]['filter']
+ self.assertEqual(predicate.method, 'or')
+ self.assertEqual(
+ [(p.field, p.literals) for p in predicate.literals],
+ [('id', [1]), ('id', [3])],
+ )
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_unconditional_clause_disables_pushdown(self):
+ from pypaimon.ray.ray_paimon import read_paimon as real_read_paimon
+
+ target = self._create_table()
+ self._write(target, self._source(ids=(1, 2, 3)))
+
+ with patch(
+ 'pypaimon.ray.ray_paimon.read_paimon',
+ wraps=real_read_paimon,
+ ) as mock_read:
+ result = merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[
+ WhenMatched.update(
+ {'age': lit(11)}, condition='t.id = 1',
+ ),
+ WhenMatched.update({'age': lit(99)}),
+ ],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(result['num_matched'], 3)
+ self.assertNotIn('filter', mock_read.call_args[1])
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_column_comparison_fails_open(self):
+ from pypaimon.ray.ray_paimon import read_paimon as real_read_paimon
+
+ target = self._create_table()
+ self._write(target, self._source(ids=(1, 2, 3)))
+
+ with patch(
+ 'pypaimon.ray.ray_paimon.read_paimon',
+ wraps=real_read_paimon,
+ ) as mock_read:
+ result = merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[WhenMatched.update(
+ {'name': lit('same')}, condition='t.age = s.age',
+ )],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(result['num_matched'], 3)
+ self.assertNotIn('filter', mock_read.call_args[1])
+ self.assertEqual(self._read_sorted(target)['name'],
+ ['same', 'same', 'same'])
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_pushdown_preserves_field_case(self):
+ from pypaimon.ray.ray_paimon import read_paimon as real_read_paimon
+
+ case_schema = pa.schema([
+ ('UserID', pa.int32()),
+ ('Value', pa.int32()),
+ ])
+ target = f'default.tbl_{uuid.uuid4().hex[:8]}'
+ schema = Schema.from_pyarrow_schema(
+ case_schema, options=self.de_options,
+ )
+ self.catalog.create_table(target, schema, False)
+ self._write(target, pa.Table.from_pydict(
+ {'UserID': [1, 2, 3], 'Value': [10, 20, 30]},
+ schema=case_schema,
+ ))
+
+ with patch(
+ 'pypaimon.ray.ray_paimon.read_paimon',
+ wraps=real_read_paimon,
+ ) as mock_read:
+ result = merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[WhenMatched.update(
+ {'Value': lit(99)},
+ condition='t.UserID IN (1, 3)',
+ )],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(result['num_matched'], 2)
+ predicate = mock_read.call_args[1]['filter']
+ self.assertEqual(predicate.field, 'UserID')
+ table = self.catalog.get_table(target)
+ read_builder = table.new_read_builder()
+ splits = read_builder.new_scan().plan().splits()
+ out = read_builder.new_read().to_arrow(splits).sort_by('UserID')
+ self.assertEqual(out['Value'].to_pylist(), [99, 20, 99])
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_date_condition_fails_open(self):
+ from pypaimon.ray.ray_paimon import read_paimon as real_read_paimon
+
+ date_schema = pa.schema([
+ ('id', pa.int32()),
+ ('event_date', pa.date32()),
+ ('value', pa.int32()),
+ ])
+ target = f'default.tbl_{uuid.uuid4().hex[:8]}'
+ schema = Schema.from_pyarrow_schema(
+ date_schema, options=self.de_options,
+ )
+ self.catalog.create_table(target, schema, False)
+ self._write(target, pa.Table.from_pydict({
+ 'id': [1, 2],
+ 'event_date': [
+ datetime.date(2026, 1, 1),
+ datetime.date(2026, 1, 2),
+ ],
+ 'value': [10, 20],
+ }, schema=date_schema))
+
+ with patch(
+ 'pypaimon.ray.ray_paimon.read_paimon',
+ wraps=real_read_paimon,
+ ) as mock_read:
+ result = merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[WhenMatched.update(
+ {'value': lit(99)},
+ condition="t.event_date = '2026-01-01'",
+ )],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(result['num_matched'], 1)
+ self.assertNotIn('filter', mock_read.call_args[1])
+ table = self.catalog.get_table(target)
+ read_builder = table.new_read_builder()
+ splits = read_builder.new_scan().plan().splits()
+ out = read_builder.new_read().to_arrow(splits).sort_by('id')
+ self.assertEqual(out['value'].to_pylist(), [99, 20])
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_double_condition_fails_open(self):
+ from pypaimon.ray.ray_paimon import read_paimon as real_read_paimon
+
+ double_schema = pa.schema([
+ ('id', pa.int32()),
+ ('metric', pa.float64()),
+ ('value', pa.int32()),
+ ])
+ target = f'default.tbl_{uuid.uuid4().hex[:8]}'
+ schema = Schema.from_pyarrow_schema(
+ double_schema, options=self.de_options,
+ )
+ self.catalog.create_table(target, schema, False)
+ self._write(target, pa.Table.from_pydict({
+ 'id': [1, 2, 3],
+ 'metric': [float('nan'), -1.0, 1.0],
+ 'value': [10, 20, 30],
+ }, schema=double_schema))
+
+ with patch(
+ 'pypaimon.ray.ray_paimon.read_paimon',
+ wraps=real_read_paimon,
+ ) as mock_read:
+ result = merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[WhenMatched.update(
+ {'value': lit(99)}, condition='t.metric > 0',
+ )],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(result['num_matched'], 2)
+ self.assertNotIn('filter', mock_read.call_args[1])
+ table = self.catalog.get_table(target)
+ read_builder = table.new_read_builder()
+ splits = read_builder.new_scan().plan().splits()
+ out = read_builder.new_read().to_arrow(splits).sort_by('id')
+ self.assertEqual(out['value'].to_pylist(), [99, 20, 99])
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_like_condition_fails_open(self):
+ from pypaimon.ray.ray_paimon import read_paimon as real_read_paimon
+
+ string_schema = pa.schema([
+ ('id', pa.int32()),
+ ('text', pa.string()),
+ ('value', pa.int32()),
+ ])
+ target = f'default.tbl_{uuid.uuid4().hex[:8]}'
+ schema = Schema.from_pyarrow_schema(
+ string_schema, options=self.de_options,
+ )
+ self.catalog.create_table(target, schema, False)
+ self._write(target, pa.Table.from_pydict({
+ 'id': [0, 1, 2, 3, 4],
+ 'text': ['n', '\\n', '\n', 'line\nbreak', 'linebreak'],
+ 'value': [10, 20, 30, 40, 50],
+ }, schema=string_schema))
+
+ with patch(
+ 'pypaimon.ray.ray_paimon.read_paimon',
+ wraps=real_read_paimon,
+ ) as mock_read:
+ result = merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[WhenMatched.update(
+ {'value': lit(99)}, condition=r"t.text LIKE '%\n%'",
+ )],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(result['num_matched'], 4)
+ self.assertNotIn('filter', mock_read.call_args[1])
+ table = self.catalog.get_table(target)
+ read_builder = table.new_read_builder()
+ splits = read_builder.new_scan().plan().splits()
+ out = read_builder.new_read().to_arrow(splits).sort_by('id')
+ self.assertEqual(out['value'].to_pylist(), [99, 99, 30, 99, 99])
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_out_of_range_integer_fails_open(self):
+ from pypaimon.ray.ray_paimon import read_paimon as real_read_paimon
+
+ int_schema = pa.schema([
+ ('id', pa.int64()),
+ ('value', pa.int32()),
+ ])
+ target = f'default.tbl_{uuid.uuid4().hex[:8]}'
+ schema = Schema.from_pyarrow_schema(
+ int_schema, options=self.de_options,
+ )
+ self.catalog.create_table(target, schema, False)
+ self._write(target, pa.Table.from_pydict({
+ 'id': [-1, 0, 1],
+ 'value': [10, 20, 30],
+ }, schema=int_schema))
+
+ with patch(
+ 'pypaimon.ray.ray_paimon.read_paimon',
+ wraps=real_read_paimon,
+ ) as mock_read:
+ result = merge_into(
+ target=target,
+ source=target,
+ catalog_options=self.catalog_options,
+ on=['_ROW_ID'],
+ when_matched=[WhenMatched.update(
+ {'value': lit(99)},
+ condition='t.id < 9223372036854775808',
+ )],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(result['num_matched'], 3)
+ self.assertNotIn('filter', mock_read.call_args[1])
+ table = self.catalog.get_table(target)
+ read_builder = table.new_read_builder()
+ splits = read_builder.new_scan().plan().splits()
+ out = read_builder.new_read().to_arrow(splits).sort_by('id')
+ self.assertEqual(out['value'].to_pylist(), [99, 99, 99])
+
def test_self_merge_update_star(self):
target = self._create_table()
self._write(
@@ -2530,6 +3014,20 @@ class TargetProjectionTest(unittest.TestCase):
class MergeConditionUnitTest(unittest.TestCase):
+ @staticmethod
+ def _predicate_fields():
+ return Schema.from_pyarrow_schema(pa.schema([
+ ('id', pa.int32()),
+ ('name', pa.string()),
+ ('MixedCase', pa.int32()),
+ ('flag', pa.bool_()),
+ ('event_date', pa.date32()),
+ ('event_time', pa.timestamp('us')),
+ ('amount', pa.decimal128(30, 2)),
+ ('float_value', pa.float32()),
+ ('double_value', pa.float64()),
+ ])).fields
+
def test_rewrite_condition(self):
from pypaimon.ray.merge_condition import rewrite_condition
self.assertEqual(
@@ -2592,6 +3090,155 @@ class MergeConditionUnitTest(unittest.TestCase):
{'s.id', 't.id', 's.age', 't.age'},
)
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_parse_simple_self_merge_predicate(self):
+ from pypaimon.ray.merge_condition import (
+ try_parse_self_merge_predicate,
+ )
+ predicate = try_parse_self_merge_predicate(
+ 't.id IN (1, 3) AND s.name = \'s.literal\'',
+ self._predicate_fields(),
+ )
+ self.assertEqual(predicate.method, 'and')
+ self.assertEqual(
+ [(p.field, p.literals) for p in predicate.literals],
+ [('id', [1, 3]), ('name', ['s.literal'])],
+ )
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_parse_self_merge_predicate_ast_subset(self):
+ from pypaimon.ray.merge_condition import (
+ try_parse_self_merge_predicate,
+ )
+ cases = [
+ ('t.id != 1', 'notEqual', 'id', [1]),
+ ('t.id NOT IN (1, 2)', 'notIn', 'id', [1, 2]),
+ ('t.id BETWEEN 1 AND 2', 'between', 'id', [1, 2]),
+ ('t.id NOT BETWEEN 1 AND 2', 'notBetween', 'id', [1, 2]),
+ ('t.event_date IS NULL', 'isNull', 'event_date', None),
+ ('t.event_date IS NOT NULL', 'isNotNull', 'event_date', None),
+ ]
+ for condition, method, field, literals in cases:
+ with self.subTest(condition=condition):
+ predicate = try_parse_self_merge_predicate(
+ condition, self._predicate_fields(),
+ )
+ self.assertEqual(
+ (predicate.method, predicate.field, predicate.literals),
+ (method, field, literals),
+ )
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_predicate_preserves_field_case(self):
+ from pypaimon.ray.merge_condition import (
+ try_parse_self_merge_predicate,
+ )
+ predicate = try_parse_self_merge_predicate(
+ 't.MixedCase = 1', self._predicate_fields(),
+ )
+ self.assertEqual(predicate.field, 'MixedCase')
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_column_comparison_is_not_pushed_down(self):
+ from pypaimon.ray.merge_condition import (
+ try_parse_self_merge_predicate,
+ )
+ self.assertIsNone(try_parse_self_merge_predicate(
+ 't.id = s.id', self._predicate_fields(),
+ ))
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_function_is_not_pushed_down(self):
+ from pypaimon.ray.merge_condition import (
+ try_parse_self_merge_predicate,
+ )
+ self.assertIsNone(try_parse_self_merge_predicate(
+ 'abs(t.id) > 1', self._predicate_fields(),
+ ))
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_predicate_parse_failure_is_logged(self):
+ from pypaimon.ray.merge_condition import (
+ try_parse_self_merge_predicate,
+ )
+ with self.assertLogs(
+ 'pypaimon.ray.merge_condition', level='DEBUG') as logs:
+ self.assertIsNone(try_parse_self_merge_predicate(
+ 't.id =', self._predicate_fields(),
+ ))
+ self.assertIn('Unable to push down', '\n'.join(logs.output))
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_unsafe_literal_types_fail_open(self):
+ from pypaimon.ray.merge_condition import (
+ try_parse_self_merge_predicate,
+ )
+ conditions = [
+ "t.event_date = '2026-01-01'",
+ "t.event_time = '2026-01-01 01:02:03'",
+ 't.amount = 12345678901234567890.12',
+ 't.float_value > 0',
+ 't.double_value > 0',
+ ]
+ for condition in conditions:
+ with self.subTest(condition=condition):
+ self.assertIsNone(try_parse_self_merge_predicate(
+ condition, self._predicate_fields(),
+ ))
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_invalid_literals_fail_open(self):
+ from pypaimon.ray.merge_condition import (
+ try_parse_self_merge_predicate,
+ )
+ conditions = [
+ "t.flag = 'true'",
+ 't.name = bare_value',
+ 't.name IN (NULL)',
+ ]
+ for condition in conditions:
+ with self.subTest(condition=condition):
+ self.assertIsNone(try_parse_self_merge_predicate(
+ condition, self._predicate_fields(),
+ ))
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_boolean_literal_is_pushed_down(self):
+ from pypaimon.ray.merge_condition import (
+ try_parse_self_merge_predicate,
+ )
+ predicate = try_parse_self_merge_predicate(
+ 't.flag = TRUE', self._predicate_fields(),
+ )
+ self.assertEqual((predicate.field, predicate.literals),
+ ('flag', [True]))
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_like_condition_is_not_pushed_down(self):
+ from pypaimon.ray.merge_condition import (
+ try_parse_self_merge_predicate,
+ )
+ self.assertIsNone(try_parse_self_merge_predicate(
+ r"t.name LIKE '%\n%'", self._predicate_fields(),
+ ))
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_self_merge_out_of_range_integers_fail_open(self):
+ from pypaimon.ray.merge_condition import (
+ try_parse_self_merge_predicate,
+ )
+ conditions = [
+ 't.id < 9223372036854775808',
+ 't.id > -9223372036854775809',
+ 't.id IN (1, 9223372036854775808)',
+ 't.id BETWEEN -9223372036854775809 AND 1',
+ ]
+ for condition in conditions:
+ with self.subTest(condition=condition):
+ self.assertIsNone(try_parse_self_merge_predicate(
+ condition, self._predicate_fields(),
+ ))
+
@unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
def test_filter_batch(self):
from pypaimon.ray.merge_condition import filter_batch
