[python] Use Native for transformed REST authorization filters

paimon-python/pypaimon/catalog/table_query_auth.py

@@ -20,7 +20,6 @@
 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 @@ def convert_plan(self, plan):
     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:

paimon-python/pypaimon/common/predicate_json_parser.py

@@ -75,22 +75,39 @@ def _wrap_int32(value: int) -> int:
 
 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 filter_fn(batch: pa.RecordBatch) -> pa.Array:
     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:

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]}
+
+
+@pytest.mark.parametrize('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]
+
+
+@pytest.mark.parametrize('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}]
+
+
+@pytest.mark.parametrize('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}]