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 8de6fdbda9 [python] Use Native for transformed REST authorization
filters (#10548)
8de6fdbda9 is described below
commit 8de6fdbda982e821bc91745dfe01b03f98c654f9
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Oct 14 09:55:00 2026 +0800
[python] Use Native for transformed REST authorization filters (#10548)
---
paimon-python/pypaimon/catalog/table_query_auth.py | 18 +--
.../pypaimon/common/predicate_json_parser.py | 138 +++++++++++++++++---
.../pypaimon/tests/native_transform_filter_test.py | 145 +++++++++++++++++++++
.../pypaimon/tests/predicate_json_parser_test.py | 128 ++++++++++++++++++
.../pypaimon/tests/table_query_auth_test.py | 15 +++
paimon-python/pypaimon/tests/video_format_test.py | 2 +-
pom.xml | 1 +
7 files changed, 414 insertions(+), 33 deletions(-)
diff --git a/paimon-python/pypaimon/catalog/table_query_auth.py
b/paimon-python/pypaimon/catalog/table_query_auth.py
index a1add858d0..04983ee97e 100644
--- a/paimon-python/pypaimon/catalog/table_query_auth.py
+++ b/paimon-python/pypaimon/catalog/table_query_auth.py
@@ -20,7 +20,6 @@ import json
from typing import Callable, Dict, List, Optional
import pyarrow as pa
-import pyarrow.compute as pc
from pypaimon.common.predicate_json_parser import (
_collect_all_field_refs_from_transform,
@@ -138,16 +137,13 @@ class TableQueryAuthResult:
def extract_row_filter(self) -> Optional[Callable[[pa.RecordBatch],
pa.Array]]:
if not self.filter:
return None
- filters = [parse_predicate_to_batch_filter(json_str) for json_str in
self.filter]
- if len(filters) == 1:
- return filters[0]
-
- def combined(batch: pa.RecordBatch) -> pa.Array:
- result = filters[0](batch)
- for f in filters[1:]:
- result = pc.and_(result, f(batch))
- return result
- return combined
+ if len(self.filter) == 1:
+ return parse_predicate_to_batch_filter(self.filter[0])
+ # Java PredicateBuilder.and evaluates entries in their original order.
+ return parse_predicate_to_batch_filter(json.dumps({
+ 'kind': 'COMPOUND', 'function': 'AND',
+ 'children': [json.loads(rule) for rule in self.filter],
+ }))
def parsed_column_masking(self) -> Dict[str, dict]:
if not self.column_masking:
diff --git a/paimon-python/pypaimon/common/predicate_json_parser.py
b/paimon-python/pypaimon/common/predicate_json_parser.py
index bfce29d4cc..5ab57be620 100644
--- a/paimon-python/pypaimon/common/predicate_json_parser.py
+++ b/paimon-python/pypaimon/common/predicate_json_parser.py
@@ -75,22 +75,39 @@ _DATE_EXTRACT = {
def parse_predicate_to_batch_filter(json_str: str) ->
Callable[[pa.RecordBatch], pa.Array]:
data = json.loads(json_str)
- return _build_filter(data)
+ execute = _build_filter(data)
+ validate = _build_filter(data, short_circuit=False)
+ validated_schema = None
+
+ def filter_fn(batch):
+ nonlocal validated_schema
+ # Java deserializes every child before row evaluation. Validate all
+ # definitions on typed empty arrays even if short-circuiting skips
them.
+ if validated_schema != batch.schema:
+ validate(batch.slice(0, 0))
+ validated_schema = batch.schema
+ return execute(batch)
+
+ return filter_fn
-def _build_filter(data: dict) -> Callable[[pa.RecordBatch], pa.Array]:
+def _build_filter(data: dict, short_circuit=True) ->
Callable[[pa.RecordBatch], pa.Array]:
kind = data["kind"]
if kind == "LEAF":
return _build_leaf_filter(data)
elif kind == "COMPOUND":
- return _build_compound_filter(data)
+ return _build_compound_filter(data, short_circuit)
raise ValueError(f"Unknown predicate kind: {kind}")
def _build_leaf_filter(data: dict) -> Callable:
transform = data["transform"]
function = data["function"]
- literals = data.get("literals", [])
+ literals = data.get("literals")
+ if literals is None and function in ('TRUE', 'FALSE', 'IS_NULL',
'IS_NOT_NULL', 'IS_NAN'):
+ literals = []
+ if not isinstance(literals, list):
+ raise ValueError(f"{function} needs a literal list")
def filter_fn(batch: pa.RecordBatch) -> pa.Array:
value_array = _apply_predicate_transform(transform, batch)
@@ -99,22 +116,31 @@ def _build_leaf_filter(data: dict) -> Callable:
return filter_fn
-def _build_compound_filter(data: dict) -> Callable:
+def _build_compound_filter(data: dict, short_circuit=True) -> Callable:
function = data["function"]
- child_filters = [_build_filter(child) for child in data["children"]]
+ if function not in ('AND', 'OR'):
+ raise ValueError(f"Unknown compound function: {function}")
+ children = data['children']
+ if not children:
+ raise ValueError('Compound predicate needs children')
+ child_filters = [_build_filter(child, short_circuit) for child in children]
def filter_fn(batch: pa.RecordBatch) -> pa.Array:
- if function == "AND":
- result = child_filters[0](batch)
- for cf in child_filters[1:]:
- result = pc.and_(result, cf(batch))
- return result
- elif function == "OR":
- result = child_filters[0](batch)
- for cf in child_filters[1:]:
- result = pc.or_(result, cf(batch))
- return result
- raise ValueError(f"Unknown compound function: {function}")
+ result = child_filters[0](batch)
+ for cf in child_filters[1:]:
+ if not short_circuit:
+ result = pc.and_(result, cf(batch)) if function == 'AND' else
pc.or_(result, cf(batch))
+ continue
+ # Java evaluates each later child only for undecided rows. Retain
+ # their original positions so nested compounds keep row order.
+ pending = result if function == 'AND' else pc.invert(result)
+ if not pc.any(pending).as_py():
+ break
+ if pc.all(pending).as_py():
+ result = cf(batch)
+ continue
+ result = pc.replace_with_mask(result, pending,
cf(batch.filter(pending)))
+ return result
return filter_fn
@@ -128,8 +154,30 @@ def _apply_predicate_transform(transform: dict, batch:
pa.RecordBatch,
elif name == "CAST":
col = batch.column(transform["fieldRef"]["name"])
- target_type = _paimon_type_to_arrow(transform["type"])
- return pc.cast(col, target_type, safe=False)
+ target = transform['type']
+ width = re.fullmatch(r"\s*(CHAR|VARCHAR|BINARY|VARBINARY)(?=\s|\(|\Z)"
+ r"(?:\s*\(\s*([0-9]+)\s*\))?"
+ r"(?:\s*(?:NOT\s+)?NULL)?\s*", target,
flags=re.IGNORECASE)
+ if width is not None:
+ kind, length = width.group(1).upper(), int(width.group(2) or 1)
+ if length < 1 or length > _INT_MAX:
+ raise ValueError('Cast width must be between 1 and 2147483647')
+ target_type = pa.string() if kind in ('CHAR', 'VARCHAR') else
pa.binary()
+ else:
+ target_type = _paimon_type_to_arrow(target)
+ stored = _paimon_type_to_arrow(transform['fieldRef']['type'])
+ col = _cast_input_for_java_executor(col, stored, target_type)
+ result = pc.cast(col, target_type, safe=False)
+ # Java CastTransform enforces widths; Arrow uses unbounded arrays.
+ if width is not None:
+ pad = kind in ('CHAR', 'BINARY')
+ padding = ' ' if kind == 'CHAR' else b'\x00'
+ result = pa.array([
+ None if value is None else
+ (value[:length] + padding * max(0, length - len(value)) if pad
else value[:length])
+ for value in result.to_pylist()
+ ], type=target_type)
+ return result
elif name == "UPPER":
input_col = _resolve_transform_input(transform["inputs"][0], batch)
@@ -142,7 +190,7 @@ def _apply_predicate_transform(transform: dict, batch:
pa.RecordBatch,
elif name == "CONCAT":
resolved = [_resolve_transform_input(inp, batch) for inp in
transform["inputs"]]
if not resolved:
- return pa.nulls(len(batch), type=pa.string())
+ return pa.array([''] * len(batch), type=pa.string())
return pc.binary_join_element_wise(*resolved, "")
elif name == "CONCAT_WS":
@@ -193,6 +241,43 @@ def _apply_predicate_transform(transform: dict, batch:
pa.RecordBatch,
raise ValueError(f"Unknown transform type: {name}")
+def _cast_input_for_java_executor(col, stored, target):
+ # copyWithNewInputs retains Java's original executor. Character, Boolean
+ # and byte executors require their original runtime value class.
+ needs_class = (((pa.types.is_string(stored) or
pa.types.is_large_string(stored)) and
+ not (pa.types.is_string(col.type) or
pa.types.is_large_string(col.type))) or
+ (pa.types.is_boolean(stored) and not
pa.types.is_boolean(col.type)) or
+ ((pa.types.is_binary(stored) or
pa.types.is_large_binary(stored)) and
+ not (pa.types.is_binary(col.type) or
pa.types.is_large_binary(col.type))))
+ if needs_class:
+ if col.null_count != len(col):
+ raise ValueError('CAST source no longer has the runtime type its
Java executor expects')
+ return pa.nulls(len(col), type=target)
+
+ numeric = pa.types.is_integer(stored) or pa.types.is_floating(stored) or
pa.types.is_decimal(stored)
+ if numeric and pa.types.is_string(target) and col.type != stored:
+ # NumericToString accepts Object. DATE and TIME getters return Integer,
+ # so this executor prints epoch days / milliseconds, not calendar text.
+ if pa.types.is_date(col.type):
+ return col.cast(pa.date32(), safe=False).cast(pa.int32())
+ if pa.types.is_time(col.type):
+ return col.cast(pa.time32('ms'), safe=False).cast(pa.int32())
+ if pa.types.is_decimal(col.type):
+ # Paimon Decimal.toString uses BigDecimal.toPlainString. Arrow may
+ # emit an exponent for a small scaled decimal, including zero.
+ return pa.array([None if value is None else format(value, 'f') for
value in col.to_pylist()],
+ type=pa.string())
+ if not (pa.types.is_integer(col.type) or pa.types.is_decimal(col.type)
or
+ pa.types.is_boolean(col.type) or pa.types.is_string(col.type)
or pa.types.is_large_string(col.type)):
+ # Timestamp.toString uses LocalDateTime text, and other objects
+ # have their own formatting. An Arrow cast would authorize other
+ # rows; refuse the unsupported rebound instead.
+ if col.null_count != len(col):
+ raise ValueError('CAST rebound source requires its original
Java executor formatting')
+ return pa.nulls(len(col), type=target)
+ return col
+
+
def _date_extract(name: str, field_ref: dict, batch: pa.RecordBatch) ->
pa.Array:
"""Java calendar extraction on DATE days or Timestamp milliseconds.
@@ -653,7 +738,18 @@ def _apply_leaf_function(function: str, value_array:
pa.Array, literals: list, b
two-valued logic (null β False) so that compound AND/OR behaves
identically to Java.
"""
- converted = [_convert_literal(lit, value_array.type) for lit in literals]
+ if (pa.types.is_binary(value_array.type) or
pa.types.is_large_binary(value_array.type)) and any(
+ literal is not None for literal in literals):
+ raise ValueError('Java predicates do not support non-null binary
literals')
+ converted = []
+ for literal in literals:
+ value = _convert_literal(literal, value_array.type)
+ if value is not None:
+ # Java deserializes every literal using the expression's output
+ # type before evaluation, including unused literals and NULL tests.
+ checked = pa.scalar(value, type=value_array.type)
+ value = checked if isinstance(value, pa.Scalar) else
checked.as_py()
+ converted.append(value)
if function == "EQUAL":
if converted[0] is None:
diff --git a/paimon-python/pypaimon/tests/native_transform_filter_test.py
b/paimon-python/pypaimon/tests/native_transform_filter_test.py
new file mode 100644
index 0000000000..686b9ce46b
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_transform_filter_test.py
@@ -0,0 +1,145 @@
+# 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.
+
+"""REST transformed row filters through Python and strict Native reads."""
+
+import json
+from contextlib import ExitStack
+from unittest.mock import patch
+
+import pytest
+
+from pypaimon.api.api_response import AuthTableQueryResponse
+from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.read.query_auth_split import QueryAuthSplit
+from pypaimon.read.split_serializer import deserialize_split_v1,
serialize_split_v1
+from pypaimon.tests import native_calendar_mask_test, native_query_auth_test
+
+rest_catalog = native_query_auth_test.rest_catalog
+# Exercise the same grants against append, primary-key and data-evolution
tables.
+guarded_table = native_query_auth_test.guarded_table
+calendar_table = native_calendar_mask_test.calendar_table
+_read = native_query_auth_test._read
+_field = native_query_auth_test._field
+
+pytestmark = [pytest.mark.native_plan, pytest.mark.skipif(
+ not native_runtime_available(), reason='Rust reader required')]
+
+
+def _leaf(transform, function, literals):
+ return {'kind': 'LEAF', 'transform': transform, 'function': function,
'literals': literals}
+
+
+_SOURCE = _field('source')
+_LENGTH = {'name': 'LENGTH', 'inputs': [_SOURCE]}
+_UPPER = {'name': 'UPPER', 'inputs': [_SOURCE]}
+
+
[email protected]('transform,function,literals,expected', [
+ (_UPPER, 'EQUAL', ['ABCABC'], [2]),
+ ({'name': 'LOWER', 'inputs': ['ABC']}, 'EQUAL', ['abc'], [1, 2, 3, 4]),
+ ({'name': 'CONCAT', 'inputs': [_SOURCE, 'x']}, 'EQUAL', ['abcabcx'], [2]),
+ ({'name': 'CONCAT', 'inputs': []}, 'EQUAL', [''], [1, 2, 3, 4]),
+ ({'name': 'CONCAT_WS', 'inputs': ['-', _SOURCE, None]}, 'EQUAL', [''], [3,
4]),
+ ({'name': 'SUBSTRING', 'inputs': [_SOURCE, 2, 2]}, 'EQUAL', ['πδΉ'], [1]),
+ ({'name': 'SUBSTRING',
+ 'inputs': [_SOURCE, _field('position', 'BIGINT'), _field('length',
'INT')]}, 'EQUAL', ['ab'], [2]),
+ ({'name': 'TRIM', 'trimFlag': 'BOTH', 'inputs': [_SOURCE, 'η²δΉac']},
'EQUAL', ['π'], [1]),
+ ({'name': 'BIT_LENGTH', 'inputs': [_SOURCE]}, 'GREATER_THAN', [48], [1]),
+ ({'name': 'PAD', 'direction': 'LEFT', 'inputs': [_SOURCE, 5, 'πx']},
'EQUAL', ['πxη²πδΉ'], [1]),
+ ({'name': 'TRANSLATE', 'inputs': [_SOURCE, 'η²πδΉaba', 'XY']}, 'EQUAL',
['XY'], [1]),
+ ({'name': 'OVERLAY', 'inputs': [_SOURCE, 'x', 2, 1]}, 'EQUAL', ['axcabc'],
[2]),
+ ({'name': 'NULL'}, 'IS_NULL', [], [1, 2, 3, 4]),
+ ({'name': 'CAST', 'fieldRef': _SOURCE, 'type': 'VARCHAR(1)'}, 'EQUAL',
['η²'], [1]),
+ ({'name': 'CAST', 'fieldRef': _SOURCE, 'type': 'CHAR(8)'}, 'EQUAL',
['abcabc '], [2]),
+ ({'name': 'CAST', 'fieldRef': _SOURCE, 'type': 'CHAR'}, 'EQUAL', [' '],
[3]),
+ ({'name': 'CAST', 'fieldRef': _SOURCE, 'type': 'BINARY(8)'}, 'IS_NULL',
[], [4]),
+ ({'name': 'CAST', 'fieldRef': _field('id', 'INT'), 'type': 'BIGINT'},
'EQUAL', [2], [2]),
+ (_LENGTH, 'EQUAL', [3], [1]), (_LENGTH, 'NOT_EQUAL', [3], [2, 3]),
+ (_LENGTH, 'LESS_THAN', [3], [3]), (_LENGTH, 'LESS_OR_EQUAL', [3], [1, 3]),
+ (_LENGTH, 'GREATER_THAN', [3], [2]), (_LENGTH, 'GREATER_OR_EQUAL', [3],
[1, 2]),
+ (_LENGTH, 'BETWEEN', [3, 6], [1, 2]), (_LENGTH, 'NOT_BETWEEN', [1, 4], [2,
3]),
+ (_LENGTH, 'IN', [3, None], [1]), (_LENGTH, 'NOT_IN', [3], [2, 3]),
+ (_LENGTH, 'NOT_IN', [3, None], []), (_LENGTH, 'IS_NULL', [], [4]),
+ (_LENGTH, 'IS_NOT_NULL', [], [1, 2, 3]), (_LENGTH, 'TRUE', [], [1, 2, 3,
4]),
+ (_LENGTH, 'FALSE', [], []), (_LENGTH, 'EQUAL', [None], []), (_LENGTH,
'NOT_EQUAL', [None], []),
+ (_UPPER, 'STARTS_WITH', ['AB'], [2]), (_UPPER, 'ENDS_WITH', ['BC'], [2]),
+ (_UPPER, 'CONTAINS', ['BCA'], [2]), (_UPPER, 'LIKE', ['A_C%'], [2]),
+])
+def test_transformed_filters_read_hidden_operands_natively(guarded_table,
transform, function, literals, expected):
+ table, server = guarded_table
+ server.table_query_auth_store[table.identifier.get_full_name()] =
AuthTableQueryResponse(
+ filter=[json.dumps(_leaf(transform, function, literals))])
+ for native in [False, True]:
+ result = _read(table, native, ['id']).sort_by([('id', 'ascending')])
+ assert result.to_pylist() == [{'id': id_} for id_ in expected]
+
+
[email protected]('field,type_', [('d', 'DATE'), ('ts', 'TIMESTAMP(3)'),
+ ('us', 'TIMESTAMP(6)'), ('ns',
'TIMESTAMP(9)')])
+def test_all_calendar_filters_read_hidden_operands_natively(calendar_table,
field, type_):
+ table, server = calendar_table
+ cases = [
+ ('YEAR', 2024, [1, 2]), ('MONTH', 12, [2, 3]), ('DAY', 30, [2]),
+ ('HOUR', 0 if field == 'd' else 23, [1, 2, 3] if field == 'd' else [1,
3]),
+ ('MINUTE', 0 if field == 'd' else 59, [1, 2, 3] if field == 'd' else
[1, 3]),
+ ('SECOND', 0 if field == 'd' else 59, [1, 2, 3] if field == 'd' else
[1, 3]),
+ ('QUARTER', 4, [2, 3]), ('DAY_OF_YEAR', 365, [2, 3]),
+ ('WEEKDAY', 0, [2]), ('ISO_DAY_OF_WEEK', 1, [2]), ('DAY_OF_WEEK', 2,
[2]),
+ ('WEEK', 1, [2, 3]), ('YEAR_OF_WEEK', 2025, [2]),
+ ]
+ for part, value, expected in cases:
+ server.table_query_auth_store[table.identifier.get_full_name()] =
AuthTableQueryResponse(
+ filter=[json.dumps(_leaf({'name': part, 'fieldRef': _field(field,
type_)}, 'EQUAL', [value]))])
+ for native in [False, True]:
+ result = _read(table, native, ['id']).sort_by([('id',
'ascending')])
+ assert result.to_pylist() == [{'id': id_} for id_ in expected],
(part, field, native)
+
+
+def
test_transformed_auth_filter_precedes_mask_user_filter_and_limit(guarded_table):
+ table, server = guarded_table
+ server.table_query_auth_store[table.identifier.get_full_name()] =
AuthTableQueryResponse(
+ filter=[json.dumps(_leaf({'name': 'UPPER', 'inputs':
[_field('value')]}, 'EQUAL', ['SECRET'])),
+ json.dumps(_leaf({'name': 'SUBSTRING', 'inputs': [_SOURCE, 1,
1]}, 'EQUAL', ['a']))],
+ column_masking={'value': json.dumps({'name': 'CONCAT', 'inputs':
['visible']})})
+ predicate = PredicateBuilder(table.fields).equal('value', 'visible')
+ for native in [False, True]:
+ assert _read(table, native, ['id'], predicate, limit=1).to_pylist() ==
[{'id': 2}]
+
+
[email protected]('mode', ['disable-native', 'setup-failure',
'wire-roundtrip'])
+def test_transformed_filter_fallback_retains_captured_policy(guarded_table,
mode):
+ table, server = guarded_table
+ server.table_query_auth_store[table.identifier.get_full_name()] =
AuthTableQueryResponse(
+ filter=[json.dumps(_leaf(_UPPER, 'EQUAL', ['ABCABC']))])
+ table = table.copy({'scan.native-plan.enabled': 'true',
'read.native.enabled': 'true'})
+ builder = table.new_read_builder().with_projection(['id'])
+ scan = builder.new_scan()
+ with patch.object(scan.file_scanner, 'scan',
side_effect=AssertionError('Native plan fell back')):
+ splits = scan.plan().splits()
+ assert splits and all(isinstance(split, QueryAuthSplit) for split in
splits)
+ server.table_query_auth_store[table.identifier.get_full_name()] =
AuthTableQueryResponse()
+ if mode == 'disable-native':
+ builder = table.copy({'read.native.enabled':
'false'}).new_read_builder().with_projection(['id'])
+ elif mode == 'wire-roundtrip':
+ splits = [deserialize_split_v1(serialize_split_v1(split), []) for
split in splits]
+ with ExitStack() as stack:
+ if mode == 'setup-failure':
+ stack.enter_context(patch('pypaimon.read.native_plan.native_read',
+ side_effect=RuntimeError('setup
failed')))
+ assert builder.new_read().to_arrow(splits).to_pylist() == [{'id': 2}]
diff --git a/paimon-python/pypaimon/tests/predicate_json_parser_test.py
b/paimon-python/pypaimon/tests/predicate_json_parser_test.py
index 06cae7efa8..7dc1991d66 100644
--- a/paimon-python/pypaimon/tests/predicate_json_parser_test.py
+++ b/paimon-python/pypaimon/tests/predicate_json_parser_test.py
@@ -178,6 +178,18 @@ class TestCompoundPredicates(unittest.TestCase):
"id": [1, 2, 3, 4, 5],
})
+ def test_short_circuit_only_evaluates_rows_reaching_each_child(self):
+ batch = pa.RecordBatch.from_pydict({'id': [1, 2], 's': ['abc', 'xyz'],
'position': [1, 2147483648]})
+ first = json.loads(_make_leaf('id', 'EQUAL', [1]))
+ second = {'kind': 'LEAF', 'function': 'EQUAL', 'literals': ['a'],
'transform': {
+ 'name': 'SUBSTRING', 'inputs': [
+ {'name': 's', 'type': 'STRING'}, {'name': 'position', 'type':
'BIGINT'}, 1]}}
+ for function, compare, expected in [('AND', 1, [True, False]), ('OR',
2, [True, True])]:
+ first['literals'] = [compare]
+ rule = {'kind': 'COMPOUND', 'function': function, 'children':
[first, second]}
+ with self.subTest(function=function):
+
self.assertEqual(parse_predicate_to_batch_filter(json.dumps(rule))(batch).to_pylist(),
expected)
+
def test_and(self):
pred = json.dumps({
"kind": "COMPOUND",
@@ -196,6 +208,22 @@ class TestCompoundPredicates(unittest.TestCase):
f = parse_predicate_to_batch_filter(pred)
self.assertEqual(f(self.batch).to_pylist(), [False, False, True, True,
False])
+ def test_nested_compounds_keep_original_mixed_row_positions(self):
+ batch = pa.RecordBatch.from_pydict({
+ 'id': [1, 2, 3, 4], 's': ['x', 'abc', '', None], 'position':
[2147483648, 1, 0, 2147483648]})
+
+ def guard(id_):
+ return {'kind': 'LEAF', 'function': 'EQUAL', 'literals': [id_],
+ 'transform': {'name': 'FIELD_REF', 'fieldRef': {'name':
'id', 'type': 'BIGINT'}}}
+
+ substring = {'kind': 'LEAF', 'function': 'EQUAL', 'literals': ['a'],
'transform': {
+ 'name': 'SUBSTRING', 'inputs': [
+ {'name': 's', 'type': 'STRING'}, {'name': 'position', 'type':
'BIGINT'}, 1]}}
+ rule = {'kind': 'COMPOUND', 'function': 'OR', 'children': [guard(1), {
+ 'kind': 'COMPOUND', 'function': 'AND', 'children': [guard(2),
substring]}]}
+
self.assertEqual(parse_predicate_to_batch_filter(json.dumps(rule))(batch).to_pylist(),
+ [True, True, False, False])
+
def test_or(self):
pred = json.dumps({
"kind": "COMPOUND",
@@ -214,6 +242,47 @@ class TestCompoundPredicates(unittest.TestCase):
f = parse_predicate_to_batch_filter(pred)
self.assertEqual(f(self.batch).to_pylist(), [True, False, False,
False, True])
+ def test_unused_literals_and_unreachable_children_still_validate(self):
+ batch = pa.RecordBatch.from_pydict({'s': ['abc', None]})
+ for function in ['TRUE', 'FALSE', 'IS_NULL', 'IS_NOT_NULL', 'EQUAL',
'NOT_EQUAL', 'IN']:
+ for literals in [['bad'], [None, 'bad']]:
+ rule = {'kind': 'LEAF', 'transform': {'name': 'LENGTH',
'inputs': [{'name': 's', 'type': 'STRING'}]},
+ 'function': function, 'literals': literals}
+ unreachable = {'kind': 'COMPOUND', 'function': 'OR',
'children': [
+ {'kind': 'LEAF', 'transform': {'name': 'NULL'},
'function': 'TRUE', 'literals': []}, rule]}
+ for definition in [rule, unreachable]:
+ with self.subTest(function=function, literals=literals),
self.assertRaises((ValueError, TypeError)):
+
parse_predicate_to_batch_filter(json.dumps(definition))(batch)
+
+ def test_unary_and_constant_literals_can_be_null_or_omitted(self):
+ batch = pa.RecordBatch.from_pydict({'s': ['abc', None]})
+ for function, expected in [('TRUE', [True, True]), ('FALSE', [False,
False]),
+ ('IS_NULL', [False, True]), ('IS_NOT_NULL',
[True, False])]:
+ rule = {'kind': 'LEAF', 'transform': {'name': 'LENGTH', 'inputs':
[{'name': 's', 'type': 'STRING'}]},
+ 'function': function, 'literals': None}
+
self.assertEqual(parse_predicate_to_batch_filter(json.dumps(rule))(batch).to_pylist(),
expected)
+ rule.pop('literals')
+
self.assertEqual(parse_predicate_to_batch_filter(json.dumps(rule))(batch).to_pylist(),
expected)
+
+ def test_invalid_compound_never_becomes_allow_all(self):
+ for function, children in [('AND', []), ('OR', []), ('NOT', [])]:
+ with self.assertRaises(ValueError):
+ parse_predicate_to_batch_filter(json.dumps({'kind':
'COMPOUND', 'function': function,
+ 'children':
children}))
+
+ def test_binary_predicate_literals_follow_java_rejection(self):
+ batch = pa.RecordBatch.from_pydict({'s': ['abc', None]})
+ for type_ in ['BINARY(3)', 'VARBINARY(3)']:
+ transform = {'name': 'CAST', 'fieldRef': {'name': 's', 'type':
'STRING'}, 'type': type_}
+ rule = {'kind': 'LEAF', 'transform': transform, 'function':
'IS_NULL', 'literals': []}
+
self.assertEqual(parse_predicate_to_batch_filter(json.dumps(rule))(batch).to_pylist(),
[False, True])
+ rule.update(function='EQUAL', literals=[None])
+
self.assertEqual(parse_predicate_to_batch_filter(json.dumps(rule))(batch).to_pylist(),
[False, False])
+ for function in ['EQUAL', 'TRUE', 'IS_NULL']:
+ rule.update(function=function, literals=['YQAA'])
+ with self.subTest(type=type_, function=function),
self.assertRaises(ValueError):
+ parse_predicate_to_batch_filter(json.dumps(rule))(batch)
+
class TestPredicateTransforms(unittest.TestCase):
@@ -303,6 +372,65 @@ class TestPredicateTransforms(unittest.TestCase):
f = parse_predicate_to_batch_filter(pred)
self.assertEqual(f(batch).to_pylist(), [True, False])
+ def test_cast_enforces_java_character_and_byte_widths(self):
+ batch = pa.RecordBatch.from_pydict({'s': ['πδΈ', 'a', None]})
+ for type_, expected in [
+ ('VARCHAR(1)', ['π', 'a', None]), ('CHAR(3)', ['πδΈ ', 'a ',
None]),
+ ('BINARY(6)', ['πδΈ'.encode('utf8')[:6], b'a' + b'\x00' * 5, None]),
+ ('VARBINARY(1)', [b'\xf0', b'a', None]),
+ ('VARCHAR', ['π', 'a', None]), ('CHAR', ['π', 'a', None]),
+ ('BINARY', [b'\xf0', b'a', None]), ('VARBINARY', [b'\xf0', b'a',
None]),
+ ('CHAR(3) NULL', ['πδΈ ', 'a ', None]), ('CHAR(3)NOT NULL', ['πδΈ
', 'a ', None]),
+ ]:
+ transform = {'name': 'CAST', 'fieldRef': {'name': 's', 'type':
'STRING'}, 'type': type_}
+ with self.subTest(type=type_):
+ self.assertEqual(_apply_predicate_transform(transform,
batch).to_pylist(), expected)
+
+ def test_rebound_cast_keeps_original_java_runtime_class(self):
+ transform = {'name': 'CAST', 'fieldRef': {'name': 's', 'type':
'STRING'}, 'type': 'VARCHAR(5)'}
+ with self.assertRaises(ValueError):
+ _apply_predicate_transform(transform,
pa.RecordBatch.from_pydict({'s': [42]}))
+ nulls = pa.RecordBatch.from_pydict({'s': pa.array([None],
type=pa.int32())})
+ self.assertEqual(_apply_predicate_transform(transform,
nulls).to_pylist(), [None])
+ # NumericToString accepts Object, including a rebound string column.
+ transform['fieldRef']['type'] = 'INT'
+ strings = pa.RecordBatch.from_pydict({'s': ['abc', None]})
+ self.assertEqual(_apply_predicate_transform(transform,
strings).to_pylist(), ['abc', None])
+
+ def
test_rebound_numeric_to_string_uses_java_internal_temporal_values(self):
+ transform = {'name': 'CAST', 'fieldRef': {'name': 's', 'type': 'INT'},
'type': 'STRING'}
+ for type_, values, expected in [
+ (pa.date32(), [0, -1, None], ['0', '-1', None]),
+ (pa.time32('ms'), [0, 3600001, None], ['0', '3600001', None]),
+ (pa.time64('us'), [0, 3600001123, None], ['0', '3600001', None]),
+ ]:
+ batch = pa.RecordBatch.from_pydict({'s': pa.array(values,
type=type_)})
+ with self.subTest(type=type_):
+ self.assertEqual(_apply_predicate_transform(transform,
batch).to_pylist(), expected)
+
+ def test_rebound_numeric_to_string_uses_decimal_plain_text(self):
+ from decimal import Decimal
+ transform = {'name': 'CAST', 'fieldRef': {'name': 's', 'type': 'INT'},
'type': 'STRING'}
+ values = ['0.0000000123', '0.0000000000', '-0.0000000123', None]
+ array = pa.array([Decimal(v) if v is not None else None for v in
values], type=pa.decimal128(38, 10))
+ batch = pa.RecordBatch.from_pydict({'s': array})
+ self.assertEqual(_apply_predicate_transform(transform,
batch).to_pylist(), values)
+ transform['type'] = 'VARCHAR(3)'
+ self.assertEqual(_apply_predicate_transform(transform,
batch).to_pylist(), ['0.0', '0.0', '-0.', None])
+
+ def test_rebound_numeric_to_string_refuses_other_object_formatting(self):
+ transform = {'name': 'CAST', 'fieldRef': {'name': 's', 'type': 'INT'},
'type': 'STRING'}
+ for array in [pa.array([0, None], type=pa.timestamp('ms')),
pa.array([b'a', None]),
+ pa.array([1.0, None], type=pa.float64())]:
+ with self.subTest(type=array.type), self.assertRaises(ValueError):
+ _apply_predicate_transform(transform,
pa.RecordBatch.from_pydict({'s': array}))
+ nulls = pa.RecordBatch.from_pydict({'s': pa.nulls(2,
type=array.type)})
+ self.assertEqual(_apply_predicate_transform(transform,
nulls).to_pylist(), [None, None])
+
+ def test_empty_concat_is_the_empty_string(self):
+ self.assertEqual(_apply_predicate_transform({'name': 'CONCAT',
'inputs': []}, self.batch).to_pylist(),
+ [''] * len(self.batch))
+
class TestPaimonTypeToArrow(unittest.TestCase):
diff --git a/paimon-python/pypaimon/tests/table_query_auth_test.py
b/paimon-python/pypaimon/tests/table_query_auth_test.py
index 2db77f6b97..1c39f6c2cf 100644
--- a/paimon-python/pypaimon/tests/table_query_auth_test.py
+++ b/paimon-python/pypaimon/tests/table_query_auth_test.py
@@ -227,6 +227,21 @@ class
TestTableQueryAuthResultExtractRowFilter(unittest.TestCase):
mask = fn(batch)
self.assertEqual(mask.to_pylist(), [False, False, False])
+ def test_filter_entries_short_circuit_in_server_order(self):
+ first = _simple_filter_json('dept', 'eng')
+ second = json.dumps({
+ 'kind': 'LEAF', 'function': 'EQUAL', 'literals': ['a'],
+ 'transform': {'name': 'SUBSTRING', 'inputs': [
+ {'name': 's', 'type': 'STRING'}, {'name': 'position', 'type':
'BIGINT'}, 1]},
+ })
+ batch = pa.RecordBatch.from_pydict({
+ 'dept': ['eng', 'sales'], 's': ['abc', 'xyz'], 'position': [1,
2147483648]})
+ result = TableQueryAuthResult([first, second],
None).extract_row_filter()
+ self.assertEqual(result(batch).to_pylist(), [True, False])
+ # Reversing the response's order executes the throwing transform first.
+ with self.assertRaises(ValueError):
+ TableQueryAuthResult([second, first],
None).extract_row_filter()(batch)
+
class TestTableQueryAuthResultExtraFields(unittest.TestCase):
diff --git a/paimon-python/pypaimon/tests/video_format_test.py
b/paimon-python/pypaimon/tests/video_format_test.py
index cb75955462..a973cb5b88 100644
--- a/paimon-python/pypaimon/tests/video_format_test.py
+++ b/paimon-python/pypaimon/tests/video_format_test.py
@@ -350,7 +350,7 @@ class VideoFormatTest(unittest.TestCase):
reader = mock.Mock()
reader.new_input_stream.side_effect = lambda unused: JindoInputFile(
- JindoStream())
+ JindoStream(), owner=reader)
descriptor = VideoFrameDescriptor(
"oss://bucket/video.mp4", 0, -1, 0, -1, 0)
value = Blob.from_descriptor(reader, descriptor)
diff --git a/pom.xml b/pom.xml
index ee9ba05399..0b48495c5b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -676,6 +676,7 @@ under the License.
<!-- Binary compatibility fixtures -->
<exclude>paimon-common/src/test/resources/fmindex-v1-golden.base64</exclude>
<exclude>paimon-core/src/test/resources/compatibility/**</exclude>
+
<exclude>paimon-python/pypaimon/tests/resources/calendar_extract_java.jsonl</exclude>
<exclude>paimon-python/pypaimon/sample/data/**</exclude>
<exclude>**/*.proto</exclude>
<exclude>**/Cargo*</exclude>
