[python] Use Native planning and reads for REST query authorization

paimon-python/pypaimon/common/predicate_json_parser.py

@@ -172,14 +172,14 @@ def _apply_predicate_transform(transform: dict, batch: pa.RecordBatch,
     elif name in _DATE_EXTRACT:
         return _date_extract(name, transform["fieldRef"], batch)
     elif name == "LENGTH":
-        _check_string_input("LENGTH input", transform["inputs"][0], batch)
+        _check_string_input("LENGTH input", transform["inputs"][0], batch, stored=True)
         source = _resolve_transform_input(transform["inputs"][0], batch)
         return pa.array(
             [None if v is None else len(v) for v in source.to_pylist()],
             type=pa.int32())
 
     elif name == "BIT_LENGTH":
-        _check_string_input("BIT_LENGTH input", transform["inputs"][0], batch)
+        _check_string_input("BIT_LENGTH input", transform["inputs"][0], batch, stored=True)
         source = _resolve_transform_input(transform["inputs"][0], batch)
         return pa.array(
             [None if v is None else len(v.encode("utf-8")) * 8
@@ -320,6 +320,24 @@ def _check_string_input(slot: str, value, batch: pa.RecordBatch, stored: bool =
         raise ValueError(f"{slot} field must be a string: {column.type}")
 
 
+_INTEGER_TYPE = re.compile(
+    r"\s*(?:TINYINT|SMALLINT|INT|INTEGER|BIGINT)(?=\s|\Z)"
+    r"(?:\s*(?:NOT\s+)?NULL)?\s*\Z", re.IGNORECASE)
+
+
+def _check_integer_input(slot, value, batch):
+    if value is None or (isinstance(value, int) and not isinstance(value, bool)):
+        return
+    if not isinstance(value, dict):
+        raise ValueError(f"{slot} must be an integer or a field: {value!r}")
+    stored = value.get("type")
+    if not isinstance(stored, str) or _INTEGER_TYPE.match(stored) is None:
+        raise ValueError(f"{slot} field must be an integer: {stored!r}")
+    column = _field_column(value, batch)
+    if not pa.types.is_integer(column.type):
+        raise ValueError(f"{slot} field must be an integer: {column.type}")
+
+
 class _Positions:
     """A position slot resolved lazily, as Java reads one only when a row reaches it."""
 
@@ -432,8 +450,9 @@ def _pad(inputs, direction, batch: pa.RecordBatch) -> pa.Array:
         raise ValueError(f"PAD takes 3 inputs, got {inputs!r}")
     if direction not in ("LEFT", "RIGHT"):
         raise ValueError(f"PAD direction must be LEFT or RIGHT: {direction!r}")
-    _check_string_input("PAD input", inputs[0], batch)
-    _check_string_input("PAD string", inputs[2], batch)
+    _check_string_input("PAD input", inputs[0], batch, stored=True)
+    _check_string_input("PAD string", inputs[2], batch, stored=True)
+    _check_integer_input("PAD length", inputs[1], batch)
     source = _resolve_transform_input(inputs[0], batch)
     pad = _resolve_transform_input(inputs[2], batch)
     lengths = _Positions(inputs[1], batch)
@@ -444,10 +463,11 @@ def _pad(inputs, direction, batch: pa.RecordBatch) -> pa.Array:
         value = src_list[i]
         pad_str = pad_list[i]
         raw_len = lengths.value(i)
-        if value is None or raw_len is None or pad_str is None:
+        length = _int_position(raw_len) if raw_len is not None else None
+        if value is None or length is None or pad_str is None:
             result.append(None)
             continue
-        result.append(_pad_one(value, _int_position(raw_len), pad_str, direction))
+        result.append(_pad_one(value, length, pad_str, direction))
     return pa.array(result, type=pa.string())
 
 
@@ -464,9 +484,9 @@ def _pad_one(source, length, pad, direction):
 def _translate(inputs, batch: pa.RecordBatch) -> pa.Array:
     if not isinstance(inputs, list) or len(inputs) != 3:
         raise ValueError(f"TRANSLATE takes 3 inputs, got {inputs!r}")
-    _check_string_input("TRANSLATE source", inputs[0], batch)
-    _check_string_input("TRANSLATE matching", inputs[1], batch)
-    _check_string_input("TRANSLATE replacement", inputs[2], batch)
+    _check_string_input("TRANSLATE source", inputs[0], batch, stored=True)
+    _check_string_input("TRANSLATE matching", inputs[1], batch, stored=True)
+    _check_string_input("TRANSLATE replacement", inputs[2], batch, stored=True)
     source = _resolve_transform_input(inputs[0], batch)
     matching = _resolve_transform_input(inputs[1], batch)
     replacement = _resolve_transform_input(inputs[2], batch)
@@ -499,8 +519,11 @@ def _translate_one(source, matching, replacement):
 def _overlay(inputs, batch: pa.RecordBatch) -> pa.Array:
     if not isinstance(inputs, list) or len(inputs) not in (3, 4):
         raise ValueError(f"OVERLAY takes 3 or 4 inputs, got {inputs!r}")
-    _check_string_input("OVERLAY input", inputs[0], batch)
-    _check_string_input("OVERLAY replacement", inputs[1], batch)
+    _check_string_input("OVERLAY input", inputs[0], batch, stored=True)
+    _check_string_input("OVERLAY replacement", inputs[1], batch, stored=True)
+    _check_integer_input("OVERLAY position", inputs[2], batch)
+    if len(inputs) == 4:
+        _check_integer_input("OVERLAY length", inputs[3], batch)
     source = _resolve_transform_input(inputs[0], batch)
     replacement = _resolve_transform_input(inputs[1], batch)
     positions = _Positions(inputs[2], batch)
@@ -513,15 +536,13 @@ def _overlay(inputs, batch: pa.RecordBatch) -> pa.Array:
         value = src_list[i]
         repl = repl_list[i]
         raw_pos = positions.value(i)
-        # SQL null propagation: any null argument yields a null result
-        if value is None or repl is None or raw_pos is None:
-            result.append(None)
-            continue
-        if has_length and lengths.value(i) is None:
+        # Java evaluates both integer arguments before propagating string NULLs.
+        pos = _int_position(raw_pos) if raw_pos is not None else None
+        raw_length = lengths.value(i) if has_length else None
+        length = _int_position(raw_length) if raw_length is not None else None
+        if value is None or repl is None or pos is None or (has_length and length is None):
             result.append(None)
             continue
-        pos = _int_position(raw_pos)
-        length = _int_position(lengths.value(i)) if has_length else None
         replaced = len(repl) if (length is None or length < 0) else length
         # pos - 1 and pos + replaced are 32-bit int expressions in Java's
         # OverlayTransform; wrap them so a boundary position overflows exactly

paimon-python/pypaimon/read/reader/filter_record_batch_reader.py

@@ -92,7 +92,13 @@ def _filter_batch(self, batch: pa.RecordBatch) -> Optional[pa.RecordBatch]:
         )
 
     def _filter_batch_by_row(self, batch: pa.RecordBatch) -> Optional[pa.RecordBatch]:
-        df = polars.from_arrow(pa.Table.from_batches([batch]))
+        # Predicate indices address logical fields. A record-reader adapter
+        # may prepend _row_kind, which is metadata rather than an operand.
+        columns = ([field.name for field in self.schema_fields]
+                   if self.schema_fields is not None else
+                   [name for name in batch.schema.names if name != "_row_kind"])
+        logical_batch = batch.select(columns)
+        df = polars.from_arrow(pa.Table.from_batches([logical_batch]))
         iterator = InternalRowWrapperIterator(
             self._iter_df_rows(df),
             df.width,

paimon-python/pypaimon/read/split_serializer.py

@@ -18,7 +18,8 @@
 """Read and write the cross-language ``SplitSerializer`` v1 binary format.
 
 This is the stable bridge between Java, PyPaimon and ``pypaimon_rust``. It
-supports DataSplit and IndexedSplit (including scores on the Python side) and
+supports DataSplit, IndexedSplit (including scores on the Python side),
+QueryAuthSplit with the captured rules, and
 the current DataSplit v9/DataFileMeta v21 layout. The reader remains compatible
 with DataSplit v8.
 """
@@ -31,6 +32,7 @@
 from pypaimon.manifest.schema.data_file_meta import DataFileMeta
 from pypaimon.manifest.schema.simple_stats import SimpleStats
 from pypaimon.read.split import DataSplit, Split
+from pypaimon.read.query_auth_split import QueryAuthSplit
 from pypaimon.schema.data_types import AtomicType, DataField
 from pypaimon.table.row.binary_row import BinaryRow
 from pypaimon.table.row.generic_row import (
@@ -44,6 +46,7 @@
 _SPLIT_SER_VERSION = 1
 _TYPE_DATA_SPLIT = 1
 _TYPE_INDEXED_SPLIT = 3
+_TYPE_QUERY_AUTH_SPLIT = 5
 _INDEXED_SPLIT_MAGIC = -938472394838495695
 _INDEXED_SPLIT_VERSION = 1
 _SPLIT_MAGIC = -2394839472490812314
@@ -290,7 +293,11 @@ def serialize_split_v1(split: Split, include_scores: bool = True) -> bytes:
     writer = _Writer()
     writer.i64(_SPLIT_SER_MAGIC)
     writer.i32(_SPLIT_SER_VERSION)
-    if isinstance(split, IndexedSplit):
+    if isinstance(split, QueryAuthSplit):
+        writer.i32(_TYPE_QUERY_AUTH_SPLIT)
+        writer.take(serialize_split_v1(split.split, include_scores))
+        _write_auth_result(writer, split.auth_result)
+    elif isinstance(split, IndexedSplit):
         data_split = split.data_split()
         if not isinstance(data_split, DataSplit):
             raise TypeError(
@@ -326,6 +333,32 @@ def serialize_split_v1(split: Split, include_scores: bool = True) -> bytes:
     return writer.finish()
 
 
+def _write_auth_result(writer, result):
+    writer.u8(result is not None)
+    if result is None:
+        return
+
+    def string(value):
+        if value is None:
+            writer.i32(-1)
+        else:
+            encoded = value.encode('utf-8')
+            writer.i32(len(encoded))
+            writer.take(encoded)
+
+    writer.u8(result.filter is not None)
+    if result.filter is not None:
+        writer.i32(len(result.filter))
+        for rule in result.filter:
+            string(rule)
+    writer.u8(result.column_masking is not None)
+    if result.column_masking is not None:
+        writer.i32(len(result.column_masking))
+        for column in sorted(result.column_masking, key=lambda name: name.encode('utf-16-be')):
+            string(column)
+            string(result.column_masking[column])
+
+
 def _serialized_row(row) -> bytes:
     if row is None:
         row = GenericRow([], [])
@@ -527,6 +560,14 @@ def deserialize_split_v1(data: bytes, partition_fields: List[DataField],
     merge-on-read; None for append tables.
     """
     r = _Reader(data)
+    split = _read_split_frame(r, partition_fields, key_fields)
+    r.finish()
+    return split
+
+
+def _read_split_frame(r, partition_fields, key_fields, depth=0):
+    if depth >= 32:
+        raise ValueError('QueryAuthSplit nesting exceeds 32')
     magic = r.i64()
     if magic != _SPLIT_SER_MAGIC:
         raise ValueError("bad SplitSerializer magic %d" % magic)
@@ -535,7 +576,11 @@ def deserialize_split_v1(data: bytes, partition_fields: List[DataField],
         raise ValueError(
             "unsupported SplitSerializer version %d (expected %d)" % (version, _SPLIT_SER_VERSION))
     type_id = r.i32()
-    if type_id == _TYPE_DATA_SPLIT:
+    if type_id == _TYPE_QUERY_AUTH_SPLIT:
+        split = QueryAuthSplit(
+            _read_split_frame(r, partition_fields, key_fields, depth + 1),
+            _read_auth_result(r))
+    elif type_id == _TYPE_DATA_SPLIT:
         split = _read_datasplit_body(r, partition_fields, key_fields)
     elif type_id == _TYPE_INDEXED_SPLIT:
         imagic = r.i64()
@@ -562,10 +607,38 @@ def deserialize_split_v1(data: bytes, partition_fields: List[DataField],
         split = IndexedSplit(data_split, row_ranges, scores)
     else:
         raise ValueError("unsupported split type id %d" % type_id)
-    r.finish()
     return split
 
 
+def _read_auth_result(reader):
+    from pypaimon.catalog.table_query_auth import TableQueryAuthResult
+
+    if reader.u8() == 0:
+        return None
+
+    def count(multiplier):
+        size = reader.i32()
+        if size < 0 or size > (len(reader.d) - reader.p) // (4 * multiplier):
+            raise ValueError('invalid QueryAuthSplit rule count')
+        return size
+
+    def string():
+        size = reader.i32()
+        if size < 0:
+            raise ValueError('NULL authorization rule in QueryAuthSplit')
+        return reader.take(size).decode('utf-8')
+
+    filters = [string() for _ in range(count(1))] if reader.u8() else None
+    masks = None
+    if reader.u8():
+        masks = {}
+        for _ in range(count(2)):
+            # Python 3.6/3.7 dict comprehensions evaluate values before keys.
+            column = string()
+            masks[column] = string()
+    return TableQueryAuthResult(filters, masks)
+
+
 def _read_scores(r: '_Reader') -> Optional[List[float]]:
     if r.u8() == 0:
         return None