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 8b43f291eb [python] Fix DECIMAL Arrow predicates across scales (#10317)
8b43f291eb is described below
commit 8b43f291eb8e341fa6569ffa8360601c666bbb8b
Author: zhigang <[email protected]>
AuthorDate: Fri Oct 2 09:41:15 2026 +0800
[python] Fix DECIMAL Arrow predicates across scales (#10317)
---
.../pypaimon/common/decimal_arrow_predicate.py | 142 ++++++++++++++
paimon-python/pypaimon/common/predicate.py | 53 ++++-
.../read/reader/filter_record_batch_reader.py | 2 +-
paimon-python/pypaimon/read/split_read.py | 4 +-
.../pypaimon/tests/decimal_predicate_test.py | 216 +++++++++++++++++++++
.../pypaimon/tests/schema_evolution_read_test.py | 13 ++
paimon-python/pypaimon/tests/where_parser_test.py | 3 +
7 files changed, 425 insertions(+), 8 deletions(-)
diff --git a/paimon-python/pypaimon/common/decimal_arrow_predicate.py
b/paimon-python/pypaimon/common/decimal_arrow_predicate.py
new file mode 100644
index 0000000000..b53ae95a7f
--- /dev/null
+++ b/paimon-python/pypaimon/common/decimal_arrow_predicate.py
@@ -0,0 +1,142 @@
+# 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 operator
+from decimal import Decimal
+from typing import List, Optional, Sequence, Tuple, Union
+
+import pyarrow
+from pyarrow import dataset as pyarrow_dataset
+
+DecimalLiteral = Union[Decimal, int]
+
+
+def _decimal_scaled_integer(value: DecimalLiteral, scale: int) -> Tuple[int,
int]:
+ """Return the floor and remainder of the exact value in column units."""
+ numerator, denominator = Decimal(value).as_integer_ratio()
+ if scale >= 0:
+ numerator *= 10**scale
+ else:
+ denominator *= 10**-scale
+ return divmod(numerator, denominator)
+
+
+def _decimal_scalar(unscaled: int, field_type: pyarrow.DataType) ->
pyarrow.Scalar:
+ """Build a scalar at the column's precision and scale."""
+ digits = tuple(int(digit) for digit in str(abs(unscaled)))
+ return pyarrow.scalar(
+ Decimal((int(unscaled < 0), digits, -field_type.scale)),
type=field_type
+ )
+
+
+def _decimal_comparison_on_field_scale(
+ field: pyarrow_dataset.Expression,
+ field_type: pyarrow.DataType,
+ method: str,
+ value: Optional[DecimalLiteral],
+) -> pyarrow_dataset.Expression:
+ """Compare against the column's representable values without rounding."""
+ # https://github.com/apache/arrow/issues/41011
+ # TODO: Recheck this workaround after upgrading PyArrow.
+ if value is None:
+ return field.is_valid() & field.is_null()
+ floor, remainder = _decimal_scaled_integer(value, field_type.scale)
+ ceiling = floor + (remainder != 0)
+ if method in ('equal', 'notEqual'):
+ if remainder:
+ return (
+ field.is_valid()
+ if method == 'notEqual'
+ else field.is_valid() & field.is_null()
+ )
+ threshold = floor
+ elif method in ('lessThan', 'greaterOrEqual'):
+ threshold = ceiling
+ else:
+ threshold = floor
+
+ compare = {
+ 'equal': operator.eq,
+ 'notEqual': operator.ne,
+ 'lessThan': operator.lt,
+ 'lessOrEqual': operator.le,
+ 'greaterThan': operator.gt,
+ 'greaterOrEqual': operator.ge,
+ }[method]
+ maximum = 10**field_type.precision - 1
+ if method in ('equal', 'notEqual'):
+ if abs(threshold) > maximum:
+ return (
+ field.is_valid()
+ if method == 'notEqual'
+ else field.is_valid() & field.is_null()
+ )
+ else:
+ low, high = compare(-maximum, threshold), compare(maximum, threshold)
+ if low == high:
+ return field.is_valid() if low else field.is_valid() &
field.is_null()
+ return compare(field, _decimal_scalar(threshold, field_type))
+
+
+def decimal_arrow_expression(
+ field: pyarrow_dataset.Expression,
+ field_type: pyarrow.DataType,
+ method: str,
+ literals: Sequence[Optional[DecimalLiteral]],
+) -> pyarrow_dataset.Expression:
+ """Compose exact DECIMAL comparisons for one predicate leaf."""
+
+ def compare(
+ comparison: str, value: Optional[DecimalLiteral]
+ ) -> pyarrow_dataset.Expression:
+ return _decimal_comparison_on_field_scale(field, field_type,
comparison, value)
+
+ if method in (
+ 'equal',
+ 'notEqual',
+ 'lessThan',
+ 'lessOrEqual',
+ 'greaterThan',
+ 'greaterOrEqual',
+ ):
+ return compare(method, literals[0])
+ if method == 'between':
+ return compare('greaterOrEqual', literals[0]) & compare(
+ 'lessOrEqual', literals[1]
+ )
+ if method == 'notBetween':
+ return compare('lessThan', literals[0]) | compare('greaterThan',
literals[1])
+ if method == 'notIn' and any(value is None for value in literals):
+ return field.is_valid() & field.is_null()
+ if method in ('in', 'notIn'):
+ maximum = 10**field_type.precision - 1
+ scalars: List[pyarrow.Scalar] = []
+ for value in literals:
+ if value is None:
+ continue
+ unscaled, remainder = _decimal_scaled_integer(value,
field_type.scale)
+ if not remainder and abs(unscaled) <= maximum:
+ scalars.append(_decimal_scalar(unscaled, field_type))
+ if not scalars:
+ return (
+ field.is_valid()
+ if method == 'notIn'
+ else field.is_valid() & field.is_null()
+ )
+ matches = field.isin(scalars)
+ return (~matches if method == 'notIn' else matches) & field.is_valid()
+ raise ValueError(f'Unsupported decimal predicate method: {method}')
diff --git a/paimon-python/pypaimon/common/predicate.py
b/paimon-python/pypaimon/common/predicate.py
index d93bbedfea..8967e2cecb 100644
--- a/paimon-python/pypaimon/common/predicate.py
+++ b/paimon-python/pypaimon/common/predicate.py
@@ -18,13 +18,17 @@
import re
from abc import ABC, ABCMeta, abstractmethod
from dataclasses import dataclass
-from typing import Any, Dict, List, Optional
-from typing import ClassVar
+from decimal import Decimal
+from typing import Any, ClassVar, Dict, List, Optional, Sequence, cast
import pyarrow
from pyarrow import compute as pyarrow_compute
from pyarrow import dataset as pyarrow_dataset
+from pypaimon.common.decimal_arrow_predicate import (
+ DecimalLiteral,
+ decimal_arrow_expression,
+)
from pypaimon.manifest.schema.simple_stats import SimpleStats
from pypaimon.table.row.internal_row import InternalRow
@@ -124,15 +128,22 @@ class Predicate:
return tester.test_by_stats(min_value, max_value, self.literals)
raise ValueError(f"Unsupported predicate method: {self.method}")
- def to_arrow(self) -> Any:
+ def to_arrow(self, schema: Optional[pyarrow.Schema] = None) -> Any:
+ """Convert to Arrow, using the input schema for exact decimal
comparisons."""
if self.method == 'and':
return _combine_arrow_expressions(
- [p.to_arrow() for p in self.literals],
+ [
+ p.to_arrow(schema) if schema is not None else p.to_arrow()
+ for p in self.literals
+ ],
lambda left, right: left & right,
)
if self.method == 'or':
return _combine_arrow_expressions(
- [p.to_arrow() for p in self.literals],
+ [
+ p.to_arrow(schema) if schema is not None else p.to_arrow()
+ for p in self.literals
+ ],
lambda left, right: left | right,
)
@@ -183,9 +194,39 @@ class Predicate:
return None
field = pyarrow_dataset.field(self.field)
+ literals = self.literals
+ if schema is not None and self.field in schema.names:
+ field_type = schema.field(self.field).type
+ if (
+ pyarrow.types.is_decimal(field_type)
+ and literals
+ and all(
+ value is None or isinstance(value, (Decimal, int))
+ for value in literals
+ )
+ and self.method
+ in (
+ 'equal',
+ 'notEqual',
+ 'lessThan',
+ 'lessOrEqual',
+ 'greaterThan',
+ 'greaterOrEqual',
+ 'in',
+ 'notIn',
+ 'between',
+ 'notBetween',
+ )
+ ):
+ return decimal_arrow_expression(
+ field,
+ field_type,
+ self.method,
+ cast(Sequence[Optional[DecimalLiteral]], literals),
+ )
tester = Predicate.testers.get(self.method)
if tester:
- return tester.test_by_arrow(field, self.literals)
+ return tester.test_by_arrow(field, literals)
raise ValueError("Unsupported predicate method:
{}".format(self.method))
diff --git a/paimon-python/pypaimon/read/reader/filter_record_batch_reader.py
b/paimon-python/pypaimon/read/reader/filter_record_batch_reader.py
index 2100dd9361..22dfc43ab6 100644
--- a/paimon-python/pypaimon/read/reader/filter_record_batch_reader.py
+++ b/paimon-python/pypaimon/read/reader/filter_record_batch_reader.py
@@ -70,7 +70,7 @@ class FilterRecordBatchReader(RecordBatchReader):
def _filter_batch(self, batch: pa.RecordBatch) -> Optional[pa.RecordBatch]:
if not self._use_arrow_filter:
return self._filter_batch_by_row(batch)
- expr = self.predicate.to_arrow()
+ expr = self.predicate.to_arrow(batch.schema)
if expr is None:
return self._filter_batch_by_row(batch)
result = ds.InMemoryDataset(pa.Table.from_batches([batch])).scanner(
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index a26b6e6803..50a2c0ea7a 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -642,7 +642,9 @@ class SplitRead(ABC):
read_predicate =
(trim_predicate_by_fields(self.push_down_predicate, read_file_fields)
if schema_id == self.table.table_schema.id else
None)
read_arrow_predicate = (
- read_predicate.to_arrow()
+ read_predicate.to_arrow(
+ PyarrowFieldParser.from_paimon_schema(schema_fields)
+ )
if read_predicate and self._arrow_filter_pushdown_enabled
else None
)
diff --git a/paimon-python/pypaimon/tests/decimal_predicate_test.py
b/paimon-python/pypaimon/tests/decimal_predicate_test.py
new file mode 100644
index 0000000000..b8512491a2
--- /dev/null
+++ b/paimon-python/pypaimon/tests/decimal_predicate_test.py
@@ -0,0 +1,216 @@
+# 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.
+
+from decimal import Decimal, localcontext
+import operator
+from pathlib import Path
+from typing import Callable, List, Optional, Union
+
+import pyarrow as pa
+import pyarrow.dataset as ds
+import pyarrow.parquet as pq
+import pytest
+
+from pypaimon import Schema
+from pypaimon.common.predicate import Predicate
+from pypaimon.filesystem.local_file_io import LocalFileIO
+from pypaimon.read.reader.format_pyarrow_reader import FormatPyArrowReader
+from pypaimon.read.reader.filter_record_batch_reader import
FilterRecordBatchReader
+
+
+COMPARISONS = [
+ ('equal', operator.eq),
+ ('notEqual', operator.ne),
+ ('lessThan', operator.lt),
+ ('lessOrEqual', operator.le),
+ ('greaterThan', operator.gt),
+ ('greaterOrEqual', operator.ge),
+]
+
+
[email protected]('method,compare', COMPARISONS)
[email protected](
+ 'bound',
+ [
+ Decimal('2'),
+ Decimal('24.000'),
+ Decimal('1.001'),
+ Decimal('-1.001'),
+ Decimal('1E-41'),
+ Decimal('1E+100'),
+ Decimal('-1E+100'),
+ 24,
+ ],
+)
+def test_decimal_comparison_matches_python(
+ method: str,
+ compare: Callable[[Decimal, Union[Decimal, int]], bool],
+ bound: Union[Decimal, int],
+) -> None:
+ decimal_values: List[Optional[Decimal]] = [
+ Decimal(value)
+ for value in (
+ '-24.00',
+ '-1.01',
+ '-1.00',
+ '0.00',
+ '1.00',
+ '1.01',
+ '2.00',
+ '24.00',
+ )
+ ] + [None]
+ table = pa.table({'value': pa.array(decimal_values, type=pa.decimal128(15,
2))})
+ predicate = Predicate(method, 0, 'value', [bound])
+ expected = [
+ value for value in decimal_values if value is not None and
compare(value, bound)
+ ]
+ # A small context must not round a high precision bound or column value.
+ with localcontext() as context:
+ context.prec = 6
+ expr = predicate.to_arrow(table.schema)
+ actual = ds.Scanner.from_batches(table.to_reader(), filter=expr).to_table()
+ assert actual.column('value').to_pylist() == expected
+
+
[email protected](
+ 'method,literals,expected',
+ [
+ ('in', [Decimal('1'), Decimal('2.001'), None], [Decimal('1.00')]),
+ ('notIn', [Decimal('1'), None], []),
+ (
+ 'between',
+ [Decimal('-1'), Decimal('1.001')],
+ [Decimal('-1.00'), Decimal('1.00')],
+ ),
+ ('notBetween', [Decimal('-1'), Decimal('1.001')], [Decimal('2.00')]),
+ (
+ 'in',
+ [Decimal('1E-41'), Decimal('1E+100'), Decimal('1.00')],
+ [Decimal('1.00')],
+ ),
+ (
+ 'notIn',
+ [Decimal('1E-41'), Decimal('1E+100'), Decimal('1.00')],
+ [Decimal('-1.00'), Decimal('2.00')],
+ ),
+ ],
+)
+def test_decimal_set_and_range_filters(
+ method: str, literals: List[Optional[Decimal]], expected: List[Decimal]
+) -> None:
+ values = [Decimal('-1.00'), Decimal('1.00'), Decimal('2.00'), None]
+ table = pa.table({'value': pa.array(values, type=pa.decimal128(15, 2))})
+ predicate = Predicate(method, 0, 'value', literals)
+ actual = ds.Scanner.from_batches(
+ table.to_reader(), filter=predicate.to_arrow(table.schema)
+ ).to_table()
+ assert actual.column('value').to_pylist() == expected
+
+
+def test_decimal_large_literal_preserves_precision() -> None:
+ values = [Decimal('9007199254740992.00'), Decimal('9007199254740993.00')]
+ table = pa.table({'value': pa.array(values, type=pa.decimal128(38, 2))})
+ predicate = Predicate('equal', 0, 'value',
[Decimal('9007199254740993.00')])
+ actual = ds.Scanner.from_batches(
+ table.to_reader(), filter=predicate.to_arrow(table.schema)
+ ).to_table()
+ assert actual.column('value').to_pylist() == [values[1]]
+
+
[email protected](
+ 'bound,expected',
+ [
+ (Decimal('24.000'), [Decimal('1.00'), Decimal('17.00')]),
+ (Decimal('24.001'), [Decimal('1.00'), Decimal('17.00'),
Decimal('24.00')]),
+ ],
+)
+def test_decimal_bound_preserves_row_group_pruning(
+ tmp_path: Path, bound: Decimal, expected: List[Decimal]
+) -> None:
+ values = [Decimal('1.00'), Decimal('17.00'), Decimal('24.00'),
Decimal('50.00')]
+ table = pa.table({'value': pa.array(values, type=pa.decimal128(15, 2))})
+ path = tmp_path / 'decimals.parquet'
+ pq.write_table(table, path, row_group_size=1)
+ dataset = ds.dataset(path)
+ predicate = Predicate('lessThan', 0, 'value', [bound])
+ expression = predicate.to_arrow(table.schema)
+ selected_groups = [
+ group
+ for fragment in dataset.get_fragments()
+ for group in fragment.split_by_row_group(filter=expression)
+ ]
+ assert len(selected_groups) == len(expected)
+ assert dataset.to_table(filter=expression).column('value').to_pylist() ==
expected
+
+
+def test_decimal_batch_filter_after_normalization(tmp_path: Path) -> None:
+ batch = pa.record_batch(
+ [
+ pa.array(
+ [Decimal('1.00'), Decimal('17.00'), Decimal('24.00'), None],
+ type=pa.decimal128(15, 2),
+ )
+ ],
+ names=['value'],
+ )
+ # Exercise the real post-read filter separately from physical file
pushdown.
+ path = tmp_path / 'batch.parquet'
+ pq.write_table(pa.Table.from_batches([batch]), path)
+ source = FormatPyArrowReader(
+ LocalFileIO(),
+ 'parquet',
+ str(path),
+ Schema.from_pyarrow_schema(batch.schema).fields,
+ None,
+ )
+ reader = FilterRecordBatchReader(
+ source, Predicate('lessThan', 0, 'value', [Decimal('24')])
+ )
+ try:
+ first = reader.read_arrow_batch()
+ assert first is not None
+ assert first.column(0).to_pylist() == [Decimal('1.00'),
Decimal('17.00')]
+ assert reader.read_arrow_batch() is None
+ finally:
+ reader.close()
+
+
+def test_decimal_composed_predicate_uses_schema() -> None:
+ values = [Decimal('-1.00'), Decimal('0.00'), Decimal('1.00'),
Decimal('2.00')]
+ table = pa.table({'value': pa.array(values, type=pa.decimal128(15, 2))})
+ predicate = Predicate(
+ 'or',
+ None,
+ None,
+ [
+ Predicate(
+ 'and',
+ None,
+ None,
+ [
+ Predicate('greaterOrEqual', 0, 'value', [Decimal('0')]),
+ Predicate('lessThan', 0, 'value', [Decimal('1.001')]),
+ ],
+ ),
+ Predicate('equal', 0, 'value', [Decimal('2')]),
+ ],
+ )
+ actual = ds.Scanner.from_batches(
+ table.to_reader(), filter=predicate.to_arrow(table.schema)
+ ).to_table()
+ assert actual.column('value').to_pylist() == values[1:]
diff --git a/paimon-python/pypaimon/tests/schema_evolution_read_test.py
b/paimon-python/pypaimon/tests/schema_evolution_read_test.py
index ea02304657..144129cb52 100644
--- a/paimon-python/pypaimon/tests/schema_evolution_read_test.py
+++ b/paimon-python/pypaimon/tests/schema_evolution_read_test.py
@@ -304,6 +304,19 @@ class SchemaEvolutionReadTest(unittest.TestCase):
schema=new_schema)
self.assertEqual(expected, actual)
+ if name == 'decimal_scale_up':
+ read_builder = table.new_read_builder()
+ predicate = read_builder.new_predicate_builder().less_than(
+ 'v', decimal.Decimal('4.5601')
+ )
+ read_builder.with_filter(predicate).with_projection(['k'])
+ filtered = read_builder.new_read().to_arrow(
+ self._scan_table(read_builder)
+ )
+ self.assertEqual(
+ [1, 2, 4], sorted(filtered.column('k').to_pylist())
+ )
+
def test_schema_evolution_type_lossy_old_file_only(self):
# Reading ONLY old-schema files after a lossy type change (no
# newer-schema file in the splits). The output type must equal the
diff --git a/paimon-python/pypaimon/tests/where_parser_test.py
b/paimon-python/pypaimon/tests/where_parser_test.py
index 4f38d71a91..762f4db485 100644
--- a/paimon-python/pypaimon/tests/where_parser_test.py
+++ b/paimon-python/pypaimon/tests/where_parser_test.py
@@ -508,6 +508,9 @@ class WhereParserScanTest(unittest.TestCase):
# any stored value; it must return no rows rather than round into a
match.
self.assertEqual(self._scan_ids("price = 99.999"), [])
+ def test_decimal_fractional_range_uses_column_scale(self) -> None:
+ self.assertEqual(self._scan_ids("price < 50.001"), [2])
+
def test_high_precision_decimal_integer_literal_matches(self):
# A DECIMAL(38, 2) integer-form literal exceeds the default decimal
# context; it must still rescale and match the stored scale-2 value.
