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 4e1116627f [python] Widen the read with the columns query auth rules
need (#10033)
4e1116627f is described below
commit 4e1116627f51eb8ba3c662d166fca12ddbbab0a8
Author: Jiajia Li <[email protected]>
AuthorDate: Mon Sep 21 15:05:33 2026 +0800
[python] Widen the read with the columns query auth rules need (#10033)
---
paimon-python/pypaimon/catalog/table_query_auth.py | 39 +++--
paimon-python/pypaimon/read/table_read.py | 43 +++--
.../pypaimon/tests/deferred_blob_resolve_test.py | 4 +-
.../pypaimon/tests/query_auth_widening_test.py | 191 +++++++++++++++++++++
.../pypaimon/tests/table_query_auth_test.py | 10 +-
5 files changed, 249 insertions(+), 38 deletions(-)
diff --git a/paimon-python/pypaimon/catalog/table_query_auth.py
b/paimon-python/pypaimon/catalog/table_query_auth.py
index 07cb260e75..fc82a07ffc 100644
--- a/paimon-python/pypaimon/catalog/table_query_auth.py
+++ b/paimon-python/pypaimon/catalog/table_query_auth.py
@@ -284,21 +284,34 @@ class TableQueryAuthResult:
"nested field '{}'; the second copy would be raw."
.format(path[0], '.'.join(path)))
- def get_extra_fields_for_filter(
+ def get_extra_fields(
self,
read_fields: List[DataField],
table_fields: List[DataField],
) -> List[DataField]:
- if not self.filter:
+ """Columns the rules read that the projection does not carry: every
filter operand,
+ and the inputs of every mask whose target is readable. A mask whose
target stays
+ outside the projection is inert and widens nothing."""
+ if not self.has_restrictions:
return []
- read_field_names = {f.name for f in read_fields}
- extra = []
- for json_str in self.filter:
- referenced = extract_referenced_fields(json_str)
- for name in referenced:
- if name not in read_field_names:
- field = next((f for f in table_fields if f.name == name),
None)
- if field:
- extra.append(field)
- read_field_names.add(name)
- return extra
+ projected = {f.name for f in read_fields}
+ masking = self.parsed_column_masking()
+ readable = set(projected)
+ pending = list(readable)
+ needed = set()
+ for operand in self.filter_field_names():
+ needed.add(operand)
+ if operand not in readable:
+ readable.add(operand)
+ pending.append(operand)
+ while pending:
+ transform = masking.get(pending.pop())
+ if transform is None:
+ continue
+ for name in _collect_all_field_refs_from_transform(transform):
+ needed.add(name)
+ if name not in readable:
+ readable.add(name)
+ pending.append(name)
+ by_name = {f.name: f for f in table_fields}
+ return [by_name[name] for name in sorted(needed - projected) if name
in by_name]
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index 34bed8aa97..0215d8eec5 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -1399,17 +1399,23 @@ class TableRead:
) if push_down_limit else None
effective_read_type = read_type if read_type is not None else
self.read_type
scan_read_type = self._with_predicate_extra_fields(read_type) if
read_type is not None else self._scan_read_type
+ # the columns authorization widens the read with are plain top-level
ones, so each
+ # gets a one-segment path; without them the outer extraction sees
fewer paths than fields
+ nested_name_paths = self.nested_name_paths
+ if nested_name_paths and len(nested_name_paths) <
len(effective_read_type):
+ nested_name_paths = list(nested_name_paths) + [
+ [f.name] for f in effective_read_type[len(nested_name_paths):]]
if self.table.is_primary_key_table and (
getattr(split, 'is_streaming', False) or not
split.raw_convertible):
inner_read_type = scan_read_type
outer_extract_name_paths: Optional[List[List[str]]] = None
- if self.nested_name_paths and any(
- len(p) > 1 for p in self.nested_name_paths):
+ if nested_name_paths and any(
+ len(p) > 1 for p in nested_name_paths):
# Inner: full ROW for the merge function. Outer: extract
# the requested sub-paths back to the user's flat schema.
inner_read_type = self._with_predicate_extra_fields(
- self._widen_to_top_level_for_merge())
- outer_extract_name_paths = self.nested_name_paths
+ self._widen_to_top_level_for_merge(nested_name_paths))
+ outer_extract_name_paths = nested_name_paths
# When the user's projection drops a ``sequence.field``, the merge
# heap can't compare it. Inject the missing sequence field(s) into
@@ -1450,15 +1456,15 @@ class TableRead:
limit=effective_limit,
)
elif self.table.options.data_evolution_enabled():
- if self.nested_name_paths and any(
- len(p) > 1 for p in self.nested_name_paths):
+ if nested_name_paths and any(
+ len(p) > 1 for p in nested_name_paths):
if not self._only_map_key_nested_paths():
raise NotImplementedError(
"ROW nested-field projection on data-evolution tables "
"is not yet supported")
scan_read_type = self._with_predicate_extra_fields(
- self._widen_to_top_level_for_merge())
- outer_extract_name_paths = self.nested_name_paths
+ self._widen_to_top_level_for_merge(nested_name_paths))
+ outer_extract_name_paths = nested_name_paths
else:
outer_extract_name_paths = None
if (outer_extract_name_paths is None
@@ -1473,7 +1479,7 @@ class TableRead:
row_tracking_enabled=True,
nested_name_paths=(
None if outer_extract_name_paths
- else self.nested_name_paths),
+ else nested_name_paths),
outer_extract_name_paths=outer_extract_name_paths,
outer_flat_read_type=(
self.read_type if outer_extract_name_paths else None),
@@ -1485,16 +1491,16 @@ class TableRead:
else:
inner_read_type = scan_read_type
outer_extract_name_paths: Optional[List[List[str]]] = None
- if self.nested_name_paths and any(
- len(p) > 1 for p in self.nested_name_paths):
+ if nested_name_paths and any(
+ len(p) > 1 for p in nested_name_paths):
# Mirror the merge path: read the full top-level columns so
# the per-file field-id normalization applies (a leaf path is
# only valid against the latest schema, not each file's own
# names/types), then extract the requested sub-paths back to
# the user's flat schema.
inner_read_type = self._with_predicate_extra_fields(
- self._widen_to_top_level_for_merge())
- outer_extract_name_paths = self.nested_name_paths
+ self._widen_to_top_level_for_merge(nested_name_paths))
+ outer_extract_name_paths = nested_name_paths
if read_type is None and outer_extract_name_paths is None and
self._needs_output_projection():
outer_extract_name_paths = self._output_extract_name_paths()
return RawFileSplitRead(
@@ -1551,15 +1557,16 @@ class TableRead:
fields = SpecialFields.row_type_with_row_tracking(fields)
return fields
- def _widen_to_top_level_for_merge(self) -> List[DataField]:
- """Unique top-level fields from ``self.nested_name_paths``, in path
order."""
+ def _widen_to_top_level_for_merge(self, name_paths=None) ->
List[DataField]:
+ """Unique top-level fields from ``name_paths``, in path order."""
+ paths = self.nested_name_paths if name_paths is None else name_paths
table_fields_by_name = {f.name: f for f in self._table_read_fields()}
paths_by_top = {}
- for path in self.nested_name_paths or []:
+ for path in paths or []:
paths_by_top.setdefault(path[0], []).append(path)
seen = set()
widened: List[DataField] = []
- for path in self.nested_name_paths or []:
+ for path in paths or []:
top_name = path[0]
if top_name in seen:
continue
@@ -1650,7 +1657,7 @@ class TableRead:
auth_result.validate_read_type(
latest_fields, read_fields, self.nested_name_paths, table_fields)
- extra_fields = auth_result.get_extra_fields_for_filter(read_fields,
table_fields)
+ extra_fields = auth_result.get_extra_fields(read_fields, table_fields)
effective_read_type = read_fields
if extra_fields:
effective_read_type = read_fields + extra_fields
diff --git a/paimon-python/pypaimon/tests/deferred_blob_resolve_test.py
b/paimon-python/pypaimon/tests/deferred_blob_resolve_test.py
index 83d47fdc70..e23ff0e8df 100644
--- a/paimon-python/pypaimon/tests/deferred_blob_resolve_test.py
+++ b/paimon-python/pypaimon/tests/deferred_blob_resolve_test.py
@@ -69,7 +69,7 @@ class _RejectScoreOneAuthResult(TableQueryAuthResult):
super().__init__(filter=_RejectScoreOneAuthResult.filter,
column_masking=None)
@staticmethod
- def get_extra_fields_for_filter(read_fields, table_fields):
+ def get_extra_fields(read_fields, table_fields):
return []
@staticmethod
@@ -96,7 +96,7 @@ class _PayloadAuthResult(TableQueryAuthResult):
})], column_masking=None)
@staticmethod
- def get_extra_fields_for_filter(read_fields, table_fields):
+ def get_extra_fields(read_fields, table_fields):
return []
def extract_row_filter(self):
diff --git a/paimon-python/pypaimon/tests/query_auth_widening_test.py
b/paimon-python/pypaimon/tests/query_auth_widening_test.py
new file mode 100644
index 0000000000..db426f9705
--- /dev/null
+++ b/paimon-python/pypaimon/tests/query_auth_widening_test.py
@@ -0,0 +1,191 @@
+################################################################################
+# 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.
+################################################################################
+
+import json
+import os
+import shutil
+import tempfile
+import unittest
+
+import pyarrow as pa
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.catalog.table_query_auth import TableQueryAuthResult
+from pypaimon.schema.data_types import AtomicType, DataField
+
+
+def _field_ref(index, name, type_name="STRING"):
+ return {"name": "FIELD_REF",
+ "fieldRef": {"index": index, "name": name, "type": type_name}}
+
+
+def _null_mask():
+ return json.dumps({"name": "NULL"})
+
+
+def _leaf(index, name, function, literals, type_name="INT"):
+ return json.dumps({
+ "kind": "LEAF",
+ "transform": _field_ref(index, name, type_name),
+ "function": function,
+ "literals": literals,
+ })
+
+
+def _fields(*names):
+ return [DataField(i, n, AtomicType("STRING")) for i, n in enumerate(names)]
+
+
+class TestExtraFieldsWidening(unittest.TestCase):
+
+ def _extra(self, result, read, table):
+ return [f.name for f in result.get_extra_fields(
+ _fields(*read), _fields(*table))]
+
+ def test_filter_operands_widen(self):
+ result = TableQueryAuthResult(
+ filter=[_leaf(2, "score", "GREATER_THAN", [5])],
column_masking=None)
+ self.assertEqual(self._extra(result, ["id"], ["id", "name", "score"]),
+ ["score"])
+
+ def test_readable_mask_input_widens(self):
+ result = TableQueryAuthResult(filter=None, column_masking={
+ "secret": json.dumps(_field_ref(2, "salt"))})
+ self.assertEqual(
+ self._extra(result, ["id", "secret"], ["id", "secret", "salt"]),
+ ["salt"])
+
+ def test_unreadable_mask_target_widens_nothing(self):
+ result = TableQueryAuthResult(filter=None, column_masking={
+ "secret": json.dumps(_field_ref(2, "salt"))})
+ self.assertEqual(self._extra(result, ["id"], ["id", "secret",
"salt"]), [])
+
+ def test_widening_is_transitive(self):
+ result = TableQueryAuthResult(filter=None, column_masking={
+ "a": json.dumps(_field_ref(1, "b")),
+ "b": json.dumps(_field_ref(2, "c")),
+ })
+ self.assertEqual(self._extra(result, ["a"], ["a", "b", "c"]), ["b",
"c"])
+
+ def test_a_filter_operand_makes_its_mask_inputs_needed(self):
+ result = TableQueryAuthResult(
+ filter=[_leaf(1, "b", "GREATER_THAN", [5])],
+ column_masking={"b": json.dumps(_field_ref(2, "c"))})
+ self.assertEqual(self._extra(result, ["a"], ["a", "b", "c"]), ["b",
"c"])
+
+ def test_a_name_absent_from_the_table_is_skipped(self):
+ result = TableQueryAuthResult(
+ filter=[_leaf(9, "gone", "GREATER_THAN", [5])],
column_masking=None)
+ self.assertEqual(self._extra(result, ["id"], ["id"]), [])
+
+ def test_no_restrictions_widens_nothing(self):
+ self.assertEqual(
+ self._extra(TableQueryAuthResult(filter=None, column_masking=None),
+ ["id"], ["id", "score"]), [])
+
+
+class TestWidenedReadsDoNotCrash(unittest.TestCase):
+
+ @classmethod
+ def setUpClass(cls):
+ cls.tempdir = tempfile.mkdtemp()
+ cls.catalog = CatalogFactory.create(
+ {'warehouse': os.path.join(cls.tempdir, 'wh')})
+ cls.catalog.create_database('db', True)
+
+ cls.flat_schema = pa.schema([
+ ('id', pa.int32()), ('secret', pa.string()), ('salt', pa.string()),
+ ('score', pa.int32())])
+ cls._create('flat', cls.flat_schema, {
+ 'id': [1, 2], 'secret': ['a', 'b'], 'salt': ['x', 'y'],
+ 'score': [1, 9]})
+
+ cls.nested_schema = pa.schema([
+ ('id', pa.int32()),
+ ('s', pa.struct([('a', pa.string()), ('b', pa.string())])),
+ ('score', pa.int32())])
+ cls._create('nested', cls.nested_schema, {
+ 'id': [1, 2],
+ 's': [{'a': 'a1', 'b': 'b1'}, {'a': 'a2', 'b': 'b2'}],
+ 'score': [1, 9]})
+
+ @classmethod
+ def _create(cls, name, pa_schema, rows):
+ cls.catalog.create_table(
+ 'db.' + name, Schema.from_pyarrow_schema(pa_schema), False)
+ table = cls.catalog.get_table('db.' + name)
+ writer = table.new_batch_write_builder().new_write()
+ writer.write_arrow(pa.Table.from_pydict(rows, schema=pa_schema))
+ table.new_batch_write_builder().new_commit().commit(
+ writer.prepare_commit())
+ writer.close()
+
+ @classmethod
+ def tearDownClass(cls):
+ shutil.rmtree(cls.tempdir, ignore_errors=True)
+
+ def _read(self, name, projection, auth_result=None):
+ table = self.catalog.get_table('db.' + name)
+ if auth_result is not None:
+ table.catalog_environment.table_query_auth = (
+ lambda options, identifier: (lambda select: auth_result))
+ builder = table.new_read_builder()
+ if projection is not None:
+ builder = builder.with_projection(projection)
+ splits = builder.new_scan().plan().splits()
+ return builder.new_read().to_arrow(splits).to_pydict()
+
+ def test_mask_input_outside_the_projection(self):
+ out = self._read('flat', ['id', 'secret'], TableQueryAuthResult(
+ filter=None,
+ column_masking={'secret': json.dumps(_field_ref(2, 'salt'))}))
+ self.assertEqual(out, {'id': [1, 2], 'secret': ['x', 'y']})
+
+ def test_a_mask_reading_a_masked_column_is_still_refused(self):
+ with self.assertRaises(ValueError) as ctx:
+ self._read('flat', ['id', 'secret'], TableQueryAuthResult(
+ filter=None,
+ column_masking={'secret': json.dumps(_field_ref(2, 'salt')),
+ 'salt': _null_mask()}))
+ self.assertIn("which is masked too", str(ctx.exception))
+
+ def test_filter_operand_outside_the_projection(self):
+ out = self._read('flat', ['id', 'secret'], TableQueryAuthResult(
+ filter=[_leaf(3, 'score', 'GREATER_THAN', [5])],
+ column_masking=None))
+ self.assertEqual(out, {'id': [2], 'secret': ['b']})
+
+ def test_nested_projection_with_filter_on_an_unprojected_column(self):
+ out = self._read('nested', ['id', 's.a'], TableQueryAuthResult(
+ filter=[_leaf(2, 'score', 'GREATER_THAN', [5])],
+ column_masking=None))
+ self.assertEqual(out, {'id': [2], 's_a': ['a2']})
+
+ def test_nested_projection_with_a_mask_input_outside_the_projection(self):
+ out = self._read('nested', ['id', 's.a'], TableQueryAuthResult(
+ filter=None,
+ column_masking={'id': json.dumps(_field_ref(2, 'score', 'INT'))}))
+ self.assertEqual(out, {'id': [1, 9], 's_a': ['a1', 'a2']})
+
+ def test_nested_projection_without_authorization_is_unchanged(self):
+ self.assertEqual(self._read('nested', ['id', 's.a']),
+ {'id': [1, 2], 's_a': ['a1', 'a2']})
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/table_query_auth_test.py
b/paimon-python/pypaimon/tests/table_query_auth_test.py
index 0f28ace877..05deff7101 100644
--- a/paimon-python/pypaimon/tests/table_query_auth_test.py
+++ b/paimon-python/pypaimon/tests/table_query_auth_test.py
@@ -161,7 +161,7 @@ class
TestTableQueryAuthResultConvertPlan(unittest.TestCase):
def test_blank_filter_no_extra_fields(self):
result = TableQueryAuthResult([""], None)
- extra = result.get_extra_fields_for_filter(
+ extra = result.get_extra_fields(
[_FakeField("a")], [_FakeField("a"), _FakeField("b")])
self.assertEqual(extra, [])
@@ -247,7 +247,7 @@ class
TestTableQueryAuthResultExtraFields(unittest.TestCase):
_FakeField("dept"),
]
result = TableQueryAuthResult([_simple_filter_json("dept")], None)
- extra = result.get_extra_fields_for_filter(read_fields, table_fields)
+ extra = result.get_extra_fields(read_fields, table_fields)
self.assertEqual(len(extra), 1)
self.assertEqual(extra[0].name, "dept")
@@ -255,12 +255,12 @@ class
TestTableQueryAuthResultExtraFields(unittest.TestCase):
read_fields = [_FakeField("name"), _FakeField("dept")]
table_fields = read_fields + [_FakeField("age")]
result = TableQueryAuthResult([_simple_filter_json("dept")], None)
- extra = result.get_extra_fields_for_filter(read_fields, table_fields)
+ extra = result.get_extra_fields(read_fields, table_fields)
self.assertEqual(len(extra), 0)
def test_no_extra_when_no_filter(self):
result = TableQueryAuthResult(None, None)
- extra = result.get_extra_fields_for_filter(
+ extra = result.get_extra_fields(
[_FakeField("a")], [_FakeField("a"), _FakeField("b")]
)
self.assertEqual(len(extra), 0)
@@ -271,7 +271,7 @@ class
TestTableQueryAuthResultExtraFields(unittest.TestCase):
read_fields = [_FakeField("name")]
table_fields = [_FakeField("name"), _FakeField("dept")]
result = TableQueryAuthResult([f1, f2], None)
- extra = result.get_extra_fields_for_filter(read_fields, table_fields)
+ extra = result.get_extra_fields(read_fields, table_fields)
self.assertEqual(len(extra), 1)
