diff --git a/paimon-python/pypaimon/catalog/table_query_auth.py b/paimon-python/pypaimon/catalog/table_query_auth.py
index a1add858d0e3..04983ee97e72 100644
--- a/paimon-python/pypaimon/catalog/table_query_auth.py
+++ b/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:
diff --git a/paimon-python/pypaimon/common/predicate_json_parser.py b/paimon-python/pypaimon/common/predicate_json_parser.py
index bfce29d4cc12..5ab57be62098 100644
--- a/paimon-python/pypaimon/common/predicate_json_parser.py
+++ b/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:
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 000000000000..686b9ce46b1b
--- /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]}
+
+
+@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}]
diff --git a/paimon-python/pypaimon/tests/predicate_json_parser_test.py b/paimon-python/pypaimon/tests/predicate_json_parser_test.py
index 06cae7efa890..7dc1991d664a 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 @@ def setUp(self):
             "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 @@ def test_and(self):
         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 @@ def test_or(self):
         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 @@ def test_concat_ws_transform_in_predicate(self):
         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 2db77f6b9712..1c39f6c2cf34 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 @@ def test_multiple_filters_combined_with_and(self):
         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 cb7595546230..a973cb5b8815 100644
--- a/paimon-python/pypaimon/tests/video_format_test.py
+++ b/paimon-python/pypaimon/tests/video_format_test.py
@@ -350,7 +350,7 @@ def close(self):
 
         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 ee9ba053991a..0b48495c5b69 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>