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 27658695ac [python] Support LENGTH, BIT_LENGTH, OVERLAY, PAD and
TRANSLATE in query-auth rules (#10179)
27658695ac is described below
commit 27658695ac14b0d39e6902d4387ddc736cd1869d
Author: jackylee <[email protected]>
AuthorDate: Sat Oct 3 17:11:58 2026 +0800
[python] Support LENGTH, BIT_LENGTH, OVERLAY, PAD and TRANSLATE in
query-auth rules (#10179)
---
.../pypaimon/common/predicate_json_parser.py | 139 +++++++++++++++++++++
.../pypaimon/tests/predicate_json_parser_test.py | 87 +++++++++++++
2 files changed, 226 insertions(+)
diff --git a/paimon-python/pypaimon/common/predicate_json_parser.py
b/paimon-python/pypaimon/common/predicate_json_parser.py
index c67c6442ca..a6474908ef 100644
--- a/paimon-python/pypaimon/common/predicate_json_parser.py
+++ b/paimon-python/pypaimon/common/predicate_json_parser.py
@@ -28,6 +28,17 @@ _MAX_STOP = 2 ** 31 - 1
_INT_MIN, _INT_MAX = -2 ** 31, 2 ** 31 - 1
+
+def _wrap_int32(value: int) -> int:
+ """Wrap ``value`` into signed 32-bit, mirroring Java ``int`` overflow.
+
+ OVERLAY's position arithmetic (``pos - 1`` and ``pos + replaced``) is
+ evaluated by ``OverlayTransform.transform`` as 32-bit ``int``, so a large
+ position wraps rather than growing unboundedly. Query-auth rules must admit
+ exactly the rows Java admits, so compute these the same way.
+ """
+ return ((value + 2 ** 31) % 2 ** 32) - 2 ** 31
+
# Integer.parseInt syntax: an optional sign and Unicode decimal digits, which
# Character.digit accepts, but no whitespace or underscore, which int() would.
# Java reads UTF-16 chars, so a supplementary-plane digit fails there.
@@ -160,6 +171,29 @@ 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)
+ 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)
+ source = _resolve_transform_input(transform["inputs"][0], batch)
+ return pa.array(
+ [None if v is None else len(v.encode("utf-8")) * 8
+ for v in source.to_pylist()],
+ type=pa.int32())
+
+ elif name == "TRANSLATE":
+ return _translate(transform["inputs"], batch)
+
+ elif name == "OVERLAY":
+ return _overlay(transform["inputs"], batch)
+
+ elif name == "PAD":
+ return _pad(transform["inputs"], transform.get("direction"), batch)
raise ValueError(f"Unknown transform type: {name}")
@@ -393,6 +427,111 @@ def _resolve_transform_input(inp, batch: pa.RecordBatch)
-> pa.Array:
return pa.array([str(inp)] * len(batch), type=pa.string())
+def _pad(inputs, direction, batch: pa.RecordBatch) -> pa.Array:
+ if not isinstance(inputs, list) or len(inputs) != 3:
+ 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)
+ source = _resolve_transform_input(inputs[0], batch)
+ pad = _resolve_transform_input(inputs[2], batch)
+ lengths = _Positions(inputs[1], batch)
+ src_list = source.to_pylist()
+ pad_list = pad.to_pylist()
+ result = []
+ for i in range(len(source)):
+ 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:
+ result.append(None)
+ continue
+ result.append(_pad_one(value, _int_position(raw_len), pad_str,
direction))
+ return pa.array(result, type=pa.string())
+
+
+def _pad_one(source, length, pad, direction):
+ needed = length - len(source)
+ if needed <= 0 or pad == "":
+ # Truncate to the first `length` characters (empty for a non-positive
length).
+ return source[:length] if length > 0 else ""
+ pad_chars = len(pad)
+ padding = pad * (needed // pad_chars) + pad[:needed % pad_chars]
+ return padding + source if direction == "LEFT" else source + padding
+
+
+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)
+ source = _resolve_transform_input(inputs[0], batch)
+ matching = _resolve_transform_input(inputs[1], batch)
+ replacement = _resolve_transform_input(inputs[2], batch)
+ result = []
+ for src, mat, rep in zip(source.to_pylist(), matching.to_pylist(),
+ replacement.to_pylist()):
+ result.append(_translate_one(src, mat, rep))
+ return pa.array(result, type=pa.string())
+
+
+def _translate_one(source, matching, replacement):
+ if source is None or matching is None or replacement is None:
+ return None
+ # First mapping wins per source character; a replacement code point of 0
+ # (or a missing one) deletes the character, mirroring Java's
TranslateTransform.
+ dictionary = {}
+ for i, ch in enumerate(matching):
+ if ch not in dictionary:
+ mapped = replacement[i] if i < len(replacement) else None
+ dictionary[ch] = mapped if (mapped is not None and ord(mapped) !=
0) else None
+ out = []
+ for ch in source:
+ if ch not in dictionary:
+ out.append(ch)
+ elif dictionary[ch] is not None:
+ out.append(dictionary[ch])
+ return "".join(out)
+
+
+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)
+ source = _resolve_transform_input(inputs[0], batch)
+ replacement = _resolve_transform_input(inputs[1], batch)
+ positions = _Positions(inputs[2], batch)
+ has_length = len(inputs) == 4
+ lengths = _Positions(inputs[3], batch) if has_length else None
+ src_list = source.to_pylist()
+ repl_list = replacement.to_pylist()
+ result = []
+ for i in range(len(source)):
+ 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:
+ 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
+ # as Java's does (otherwise an auth rule admits different rows here).
+ head = _substring_sql(value, 1, _wrap_int32(pos - 1))
+ tail = _substring_sql(value, _wrap_int32(pos + replaced), _INT_MAX)
+ result.append(head + repl + tail)
+ return pa.array(result, type=pa.string())
+
+
def _concat_ws(sep: pa.Array, value_arrays: list) -> pa.Array:
sep_list = sep.to_pylist()
val_lists = [v.to_pylist() for v in value_arrays]
diff --git a/paimon-python/pypaimon/tests/predicate_json_parser_test.py
b/paimon-python/pypaimon/tests/predicate_json_parser_test.py
index 8c96c1a04e..fba06c35fc 100644
--- a/paimon-python/pypaimon/tests/predicate_json_parser_test.py
+++ b/paimon-python/pypaimon/tests/predicate_json_parser_test.py
@@ -872,3 +872,90 @@ class TestDateExtractTransforms(unittest.TestCase):
self.assertEqual(
parse_predicate_to_batch_filter(pred)(self._ts_batch()).to_pylist(),
[False, True, False])
+
+
+class TestStringTransforms(unittest.TestCase):
+ """LENGTH / BIT_LENGTH / TRANSLATE / OVERLAY / PAD, mirroring the Java
+ transforms of the same name (see TransformJsonSerdeTest for the wire
shape)."""
+
+ def _apply(self, transform, column):
+ batch = pa.RecordBatch.from_pydict({"s": column})
+ return _apply_predicate_transform(transform, batch).to_pylist()
+
+ def _field(self):
+ return {"index": 0, "name": "s", "type": "STRING"}
+
+ def test_length(self):
+ out = self._apply({"name": "LENGTH", "inputs": [self._field()]},
+ ["hello", "hi", None])
+ self.assertEqual(out, [5, 2, None])
+
+ def test_bit_length(self):
+ # "é" is two UTF-8 bytes -> 16 bits.
+ out = self._apply({"name": "BIT_LENGTH", "inputs": [self._field()]},
+ ["hello", "é", None])
+ self.assertEqual(out, [40, 16, None])
+
+ def test_translate_maps_and_deletes(self):
+ mapped = self._apply(
+ {"name": "TRANSLATE", "inputs": [self._field(), "el", "ip"]},
+ ["hello", None])
+ self.assertEqual(mapped, ["hippo", None])
+ # A shorter replacement deletes the unmatched matching characters.
+ deleted = self._apply(
+ {"name": "TRANSLATE", "inputs": [self._field(), "l", ""]},
+ ["hello"])
+ self.assertEqual(deleted, ["heo"])
+
+ def test_overlay_with_and_without_length(self):
+ four = self._apply(
+ {"name": "OVERLAY", "inputs": [self._field(), "XX", 2, 1]},
+ ["hello", None])
+ self.assertEqual(four, ["hXXllo", None])
+ three = self._apply(
+ {"name": "OVERLAY", "inputs": [self._field(), "XX", 2]},
+ ["hello"])
+ self.assertEqual(three, ["hXXlo"])
+
+ def test_overlay_position_arithmetic_overflows_like_java(self):
+ # pos + replaced overflows 32-bit int in Java's OverlayTransform:
+ # OVERLAY('hello', 'x', 1, 2147483647) is 'xhell' there (the wrapped
+ # end position lands inside the string), not 'x'. The auth rule must
+ # admit the same rows in PyPaimon.
+ out = self._apply(
+ {"name": "OVERLAY", "inputs": [self._field(), "x", 1, 2147483647]},
+ ["hello"])
+ self.assertEqual(out, ["xhell"])
+
+ def test_pad_left_right_and_truncate(self):
+ left = self._apply(
+ {"name": "PAD", "inputs": [self._field(), 5, "ab"], "direction":
"LEFT"},
+ ["hi", None])
+ self.assertEqual(left, ["abahi", None])
+ right = self._apply(
+ {"name": "PAD", "inputs": [self._field(), 5, "ab"], "direction":
"RIGHT"},
+ ["hi"])
+ self.assertEqual(right, ["hiaba"])
+ # A target shorter than the source truncates.
+ self.assertEqual(
+ self._apply(
+ {"name": "PAD", "inputs": [self._field(), 3, "x"],
"direction": "LEFT"},
+ ["hello"]),
+ ["hel"])
+
+ def test_length_wired_through_a_leaf_filter(self):
+ pred = json.dumps({
+ "kind": "LEAF",
+ "transform": {"name": "LENGTH", "inputs": [self._field()]},
+ "function": "GREATER_THAN",
+ "literals": [3],
+ })
+ batch = pa.RecordBatch.from_pydict({"s": ["hello", "hi", "abcd"]})
+ self.assertEqual(
+ parse_predicate_to_batch_filter(pred)(batch).to_pylist(),
+ [True, False, True])
+
+ def test_unknown_transform_still_raises(self):
+ with self.assertRaises(ValueError):
+ self._apply({"name": "NO_SUCH_TRANSFORM", "inputs":
[self._field()]},
+ ["hello"])
