You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "MgjLLL (via GitHub)" <gi...@apache.org> on 2026/06/05 11:36:24 UTC
[PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
MgjLLL opened a new pull request, #8136:
URL: https://github.com/apache/paimon/pull/8136
### Purpose
Adds query-auth support to the Python client so it honors the row-level filter and column masking rules returned by a REST catalog, matching the existing JVM client behavior.
When the new option `query-auth.enabled` is set to `true`, before producing a `Plan` the client calls `POST /v1/.../databases/{db}/tables/{tb}/auth` with the projected fields, receives `{ filter, columnMasking }`, and applies them on the read path:
- `RESTApi.auth_table_query` issues the call (new request/response models `AuthTableQueryRequest` / `AuthTableQueryResponse`, new path in `ResourcePaths.auth_table`).
- `TableQueryAuth` / `TableQueryAuthResult` (`catalog/table_query_auth.py`) wrap the result and convert each split to a `QueryAuthSplit`.
- `predicate_json_parser` (`common/predicate_json_parser.py`) parses Paimon predicate JSON into a PyArrow compute filter (EQ/NEQ/LT/LTEQ/GT/GTEQ/IS_NULL/IS_NOT_NULL/IN/NOT_IN/STARTS_WITH/ENDS_WITH/CONTAINS/AND/OR/NOT).
- `AuthFilterReader` / `AuthMaskingReader` / `ColumnProjectReader` (`read/reader/auth_masking_reader.py`) implement row filtering, column masking transforms (`NULL`, `FIELD_REF`, `CAST`, `UPPER`, `LOWER`, `CONCAT`, `CONCAT_WS`) and final projection back to the user's requested columns.
- `read_builder` / `stream_read_builder` / `table_read` / `table_scan` / `file_store_table` / `catalog_environment` / `rest_catalog` are wired to invoke the auth call and pull extra fields required only by the auth filter.
Behavior is gated by the new `CoreOptions.QUERY_AUTH_ENABLED` (`query-auth.enabled`, default `false`), so existing users see no change.
### Tests
Three new test files (994+ lines, all passing locally under `pytest`):
- `paimon-python/pypaimon/tests/predicate_json_parser_test.py` — covers each predicate kind, nested AND/OR/NOT, type coercion, null handling, and `extract_referenced_fields`.
- `paimon-python/pypaimon/tests/auth_masking_reader_test.py` — covers each masking transform, missing-field validation, and projection back to the user-requested columns.
- `paimon-python/pypaimon/tests/table_query_auth_test.py` — end-to-end coverage: REST catalog calls `auth_table_query`, the result is plumbed into the plan, splits become `QueryAuthSplit`, and reads return filtered + masked rows.
Local check:
```
cd paimon-python
python -m pytest pypaimon/tests/predicate_json_parser_test.py \
pypaimon/tests/auth_masking_reader_test.py \
pypaimon/tests/table_query_auth_test.py -q
flake8 --config dev/cfg.ini pypaimon/ # 已在改动范围内通过
```
### API and Format
- New catalog option: `query-auth.enabled` (boolean, default `false`).
- New REST endpoint consumed by the client: `POST /v1/{prefix}/databases/{db}/tables/{tb}/auth`. Request `{ "select": [...] }`, response `{ "filter": [<predicate-json>...], "columnMasking": { <col>: <transform-json>, ... } }`. The contract follows the existing Java client; no server-side change is required for catalogs that already implement query auth.
- No change to existing user-facing Python APIs. New types (`AuthTableQueryRequest`, `AuthTableQueryResponse`, `TableQueryAuth`, `TableQueryAuthResult`, `QueryAuthSplit`, `AuthFilterReader`, `AuthMaskingReader`, `ColumnProjectReader`) are additive and live under existing modules.
- File format / on-disk layout: unchanged.
### Documentation
The new option `query-auth.enabled` should be reflected in the Python configuration reference. Happy to add the docs entry in this PR or in a follow-up — please advise.
This closes #8135
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #8136:
URL: https://github.com/apache/paimon/pull/8136#issuecomment-4637972034
I found a few correctness issues in the query-auth paths introduced here:
1. `paimon-python/pypaimon/read/datasource/split_provider.py:127` constructs `ReadBuilder(self._ensure_table())` directly. That bypasses `FileStoreTable.new_read_builder()`, which is where the REST query auth is injected. As a result, `pypaimon.ray.read_paimon(...)` can read REST tables without applying server-side row filters or column masking. I think this should either call `self._ensure_table().new_read_builder()` or explicitly pass the table's query auth into the builder, with a Ray regression test for row filtering/masking.
2. `paimon-python/pypaimon/read/stream_read_builder.py:117` stores `_query_auth`, but `new_streaming_scan()` does not pass it into `AsyncStreamingTableScan`. The plans returned from `streaming_table_scan.py:322` and `streaming_table_scan.py:386` also do not go through `auth_result.convert_plan()`. So `table.new_stream_read_builder()` skips row filters and column masking for both the initial scan and later delta/changelog scans. The streaming scan should preserve and apply query auth before returning each plan.
3. The auth reader wrappers currently assume the inner reader supports `read_arrow_batch()` (`paimon-python/pypaimon/read/reader/auth_masking_reader.py:38` and `:66`). For primary-key tables with non-raw-convertible splits, `TableRead` can create a `MergeFileSplitRead`, whose `create_reader()` returns the normal row `RecordReader` path rather than a `RecordBatchReader`. Wrapping that in `AuthFilterReader`/`AuthMaskingReader` will fail with `AttributeError` when query auth is enabled. This needs either a row-reader auth path, conversion to a batch-capable reader before wrapping, or routing/rejecting these splits explicitly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #8136:
URL: https://github.com/apache/paimon/pull/8136#issuecomment-4954310415
`scan_with_stats() did not call 'auth_result.convert_plan(plan)', causing the explain/aft plan to bypass' QueryAuthSplit '; A pure blank row filter will also trigger 'JSONDecodeError'.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "MgjLLL (via GitHub)" <gi...@apache.org>.
MgjLLL commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3458051844
##########
paimon-python/pypaimon/read/reader/auth_masking_reader.py:
##########
@@ -0,0 +1,224 @@
+################################################################################
+# 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
+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,
+ _paimon_type_to_arrow,
+)
+from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
+
+
+class RecordReaderToBatchAdapter(RecordBatchReader):
+
+ def __init__(self, inner, schema: pa.Schema, chunk_size: int = 65536, include_row_kind: bool = False):
+ self._inner = inner
+ self._schema = schema
+ self._chunk_size = chunk_size
+ self._exhausted = False
+ self._pending_iterator = None
+ self._include_row_kind = include_row_kind
+
+ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
+ if self._exhausted:
+ return None
+ row_tuples = []
+ row_kinds = []
+ while len(row_tuples) < self._chunk_size:
+ if self._pending_iterator is not None:
+ row = self._pending_iterator.next()
+ while row is not None:
+ row_tuples.append(
+ row.row_tuple[row.offset:row.offset + row.arity])
+ if self._include_row_kind:
+ row_kinds.append(row.get_row_kind().to_string())
+ if len(row_tuples) >= self._chunk_size:
+ return self._flush(row_tuples, row_kinds)
+ row = self._pending_iterator.next()
+ self._pending_iterator = None
+
+ row_iterator = self._inner.read_batch()
+ if row_iterator is None:
+ self._exhausted = True
+ break
+ self._pending_iterator = row_iterator
+
+ if not row_tuples:
+ return None
+ return self._flush(row_tuples, row_kinds)
+
+ def _flush(self, row_tuples, row_kinds=None):
+ columns_data = list(zip(*row_tuples))
+ pydict = {
+ name: list(col)
+ for name, col in zip(self._schema.names, columns_data)
+ }
+ batch = pa.RecordBatch.from_pydict(pydict, schema=self._schema)
+ if row_kinds:
+ row_kind_array = pa.array(row_kinds, type=pa.string())
+ row_kind_field = pa.field("_row_kind", pa.string())
+ new_schema = pa.schema([row_kind_field] + list(batch.schema))
+ columns = [row_kind_array] + [batch.column(i) for i in range(batch.num_columns)]
+ batch = pa.RecordBatch.from_arrays(columns, schema=new_schema)
+ return batch
+
+ def close(self):
+ self._inner.close()
+
+
+class AuthFilterReader(RecordBatchReader):
+
+ def __init__(self, inner_reader: RecordBatchReader, filter_fn: Callable[[pa.RecordBatch], pa.Array]):
+ self._inner = inner_reader
+ self._filter_fn = filter_fn
+
+ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
+ batch = self._inner.read_arrow_batch()
+ if batch is None:
+ return None
+ mask = self._filter_fn(batch)
+ return batch.filter(mask)
+
+ def close(self):
+ self._inner.close()
+
+
+class AuthMaskingReader(RecordBatchReader):
+
+ def __init__(self, inner_reader: RecordBatchReader, masking_rules: Dict[str, str], read_fields: List):
+ self._inner = inner_reader
+ self._masking_rules = masking_rules
+ self._read_fields = read_fields
+ read_field_names = {f.name for f in read_fields}
+ # Filter to projected columns, then validate field references.
+ self._parsed_rules = {
+ col: json.loads(tj) for col, tj in masking_rules.items()
+ if col in read_field_names
+ }
+ for col_name, transform in self._parsed_rules.items():
+ for ref_name in _collect_all_field_refs_from_transform(transform):
+ if ref_name not in read_field_names:
+ raise RuntimeError(
+ f"Column masking refers to field '{ref_name}' which is not present "
+ f"in output row type. Available fields: {read_field_names}"
+ )
+
+ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
+ batch = self._inner.read_arrow_batch()
+ if batch is None:
+ return None
+ original_batch = batch
+ masked_columns = {}
+ for col_name, transform in self._parsed_rules.items():
+ if col_name in original_batch.schema.names:
+ col_idx = original_batch.schema.get_field_index(col_name)
+ target_col_type = original_batch.schema.field(col_idx).type
+ masked_columns[col_idx] = self._apply_transform(transform, original_batch, target_col_type)
+ for col_idx, masked_array in masked_columns.items():
+ original_field = original_batch.schema.field(col_idx)
+ if masked_array.type != original_field.type:
+ masked_array = pc.cast(masked_array, original_field.type)
+ batch = batch.set_column(
+ col_idx, pa.field(original_field.name, original_field.type, nullable=True), masked_array)
+ return batch
+
+ def close(self):
+ self._inner.close()
+
+ def _apply_transform(
+ self,
+ transform: dict,
+ original_batch: pa.RecordBatch,
+ target_col_type: pa.DataType,
+ ) -> pa.Array:
+ name = transform["name"]
+
+ if name == "NULL":
+ return pa.nulls(len(original_batch), type=target_col_type)
+
+ elif name == "FIELD_REF":
+ ref_name = transform["fieldRef"]["name"]
+ return original_batch.column(ref_name)
+
+ elif name == "CAST":
+ ref_name = transform["fieldRef"]["name"]
+ source_col = original_batch.column(ref_name)
+ target_type = _paimon_type_to_arrow(transform["type"])
+ return pc.cast(source_col, target_type)
+
+ elif name == "UPPER":
+ input_col = self._resolve_input(transform["inputs"][0], original_batch)
+ return pc.utf8_upper(input_col)
+
+ elif name == "LOWER":
+ input_col = self._resolve_input(transform["inputs"][0], original_batch)
+ return pc.utf8_lower(input_col)
+
+ elif name == "CONCAT":
+ return self._apply_concat(transform["inputs"], original_batch)
+
+ elif name == "CONCAT_WS":
+ return self._apply_concat_ws(transform["inputs"], original_batch)
+
+ raise ValueError(f"Unknown transform type: {name}")
+
+ def _resolve_input(self, inp, original_batch: pa.RecordBatch) -> pa.Array:
+ if isinstance(inp, dict):
+ return original_batch.column(inp["name"])
+ elif isinstance(inp, str):
+ return pa.array([inp] * len(original_batch), type=pa.string())
+ elif inp is None:
+ return pa.nulls(len(original_batch), type=pa.string())
+ return pa.array([str(inp)] * len(original_batch), type=pa.string())
+
+ def _apply_concat(self, inputs: list, original_batch: pa.RecordBatch) -> pa.Array:
+ resolved = [self._resolve_input(inp, original_batch) for inp in inputs]
+ if not resolved:
+ return pa.nulls(len(original_batch), type=pa.string())
+ return pc.binary_join_element_wise(*resolved, "")
+
+ def _apply_concat_ws(self, inputs: list, original_batch: pa.RecordBatch) -> pa.Array:
+ if len(inputs) < 2:
+ return pa.nulls(len(original_batch), type=pa.string())
+ sep = self._resolve_input(inputs[0], original_batch)
+ values = [self._resolve_input(inp, original_batch) for inp in inputs[1:]]
+ return pc.binary_join_element_wise(*values, sep, null_handling='skip')
Review Comment:
Fixed. PyArrow's `binary_join_element_wise` with `null_handling='skip'` returns an empty array (length 0) when all values are null — confirmed by testing. Now pad the result to batch length and use `pc.if_else(sep_null, None, pc.if_else(pc.is_valid(result), result, ""))` to match Java's `BinaryString.concatWs` behavior: all-null values → empty string, null separator → null. Same fix applied in the predicate transform path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "MgjLLL (via GitHub)" <gi...@apache.org>.
MgjLLL commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3394107057
##########
paimon-python/pypaimon/common/predicate_json_parser.py:
##########
@@ -0,0 +1,290 @@
+################################################################################
+# 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 re
+from typing import Callable
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+
+def parse_predicate_to_batch_filter(json_str: str) -> Callable[[pa.RecordBatch], pa.Array]:
+ data = json.loads(json_str)
+ return _build_filter(data)
+
+
+def _build_filter(data: dict) -> Callable[[pa.RecordBatch], pa.Array]:
+ kind = data["kind"]
+ if kind == "LEAF":
+ return _build_leaf_filter(data)
+ elif kind == "COMPOUND":
+ return _build_compound_filter(data)
+ raise ValueError(f"Unknown predicate kind: {kind}")
+
+
+def _build_leaf_filter(data: dict) -> Callable:
+ transform = data["transform"]
+ function = data["function"]
+ literals = data.get("literals", [])
+
+ def filter_fn(batch: pa.RecordBatch) -> pa.Array:
+ value_array = _apply_predicate_transform(transform, batch)
+ return _apply_leaf_function(function, value_array, literals, len(batch))
+
+ return filter_fn
+
+
+def _build_compound_filter(data: dict) -> Callable:
+ function = data["function"]
+ child_filters = [_build_filter(child) for child in data["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}")
+
+ return filter_fn
+
+
+def _apply_predicate_transform(transform: dict, batch: pa.RecordBatch) -> pa.Array:
+ name = transform["name"]
+
+ if name == "FIELD_REF":
+ return batch.column(transform["fieldRef"]["name"])
+
+ elif name == "CAST":
+ col = batch.column(transform["fieldRef"]["name"])
+ target_type = _paimon_type_to_arrow(transform["type"])
+ return pc.cast(col, target_type)
+
+ elif name == "UPPER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_upper(input_col)
+
+ elif name == "LOWER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_lower(input_col)
+
+ 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 pc.binary_join_element_wise(*resolved, "")
+
+ elif name == "CONCAT_WS":
+ sep = _resolve_transform_input(transform["inputs"][0], batch)
+ values = [_resolve_transform_input(inp, batch) for inp in transform["inputs"][1:]]
+ if not values:
+ return pa.nulls(len(batch), type=pa.string())
+ return pc.binary_join_element_wise(*values, sep, null_handling='skip')
+
+ elif name == "NULL":
+ return pa.nulls(len(batch), type=pa.bool_())
+
+ raise ValueError(f"Unknown transform type in predicate: {name}")
+
+
+def _resolve_transform_input(inp, batch: pa.RecordBatch) -> pa.Array:
+ if isinstance(inp, dict):
+ return batch.column(inp["name"])
+ elif isinstance(inp, str):
+ return pa.array([inp] * len(batch), type=pa.string())
+ elif inp is None:
+ return pa.nulls(len(batch), type=pa.string())
+ return pa.array([str(inp)] * len(batch), type=pa.string())
+
+
+def _apply_leaf_function(function: str, value_array: pa.Array, literals: list, batch_len: int) -> pa.Array:
+ converted = [_convert_literal(lit, value_array.type) for lit in literals]
+
+ if function == "EQUAL":
+ return pc.equal(value_array, converted[0])
+ elif function == "NOT_EQUAL":
+ return pc.not_equal(value_array, converted[0])
+ elif function == "LESS_THAN":
+ return pc.less(value_array, converted[0])
+ elif function == "LESS_OR_EQUAL":
+ return pc.less_equal(value_array, converted[0])
+ elif function == "GREATER_THAN":
+ return pc.greater(value_array, converted[0])
+ elif function == "GREATER_OR_EQUAL":
+ return pc.greater_equal(value_array, converted[0])
+ elif function == "IS_NULL":
+ return pc.is_null(value_array)
+ elif function == "IS_NOT_NULL":
+ return pc.is_valid(value_array)
+ elif function == "IN":
+ return pc.is_in(value_array, pa.array(converted, type=value_array.type))
+ elif function == "NOT_IN":
+ return pc.invert(pc.is_in(value_array, pa.array(converted, type=value_array.type)))
+ elif function == "BETWEEN":
+ return pc.and_(pc.greater_equal(value_array, converted[0]),
+ pc.less_equal(value_array, converted[1]))
+ elif function == "NOT_BETWEEN":
+ return pc.or_(pc.less(value_array, converted[0]),
+ pc.greater(value_array, converted[1]))
+ elif function == "STARTS_WITH":
+ return pc.starts_with(value_array, converted[0])
+ elif function == "ENDS_WITH":
+ return pc.ends_with(value_array, converted[0])
+ elif function == "CONTAINS":
+ return pc.match_substring(value_array, converted[0])
+ elif function == "LIKE":
+ raw = literals[0]
+ escaped = re.escape(raw)
+ pattern = escaped.replace("%", ".*").replace("_", ".")
Review Comment:
Ported Like.sqlToRegexLike: backslash is the default escape character, \\% / \\_ are literal, invalid escape sequences raise. LIKE admin\_% now matches admin_foo (not adminXfoo), so policy predicates evaluate identically to the JVM client.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "MgjLLL (via GitHub)" <gi...@apache.org>.
MgjLLL commented on PR #8136:
URL: https://github.com/apache/paimon/pull/8136#issuecomment-4648668415
> I found a few correctness issues in the query-auth paths introduced here:
>
> 1. `paimon-python/pypaimon/read/datasource/split_provider.py:127` constructs `ReadBuilder(self._ensure_table())` directly. That bypasses `FileStoreTable.new_read_builder()`, which is where the REST query auth is injected. As a result, `pypaimon.ray.read_paimon(...)` can read REST tables without applying server-side row filters or column masking. I think this should either call `self._ensure_table().new_read_builder()` or explicitly pass the table's query auth into the builder, with a Ray regression test for row filtering/masking.
> 2. `paimon-python/pypaimon/read/stream_read_builder.py:117` stores `_query_auth`, but `new_streaming_scan()` does not pass it into `AsyncStreamingTableScan`. The plans returned from `streaming_table_scan.py:322` and `streaming_table_scan.py:386` also do not go through `auth_result.convert_plan()`. So `table.new_stream_read_builder()` skips row filters and column masking for both the initial scan and later delta/changelog scans. The streaming scan should preserve and apply query auth before returning each plan.
> 3. The auth reader wrappers currently assume the inner reader supports `read_arrow_batch()` (`paimon-python/pypaimon/read/reader/auth_masking_reader.py:38` and `:66`). For primary-key tables with non-raw-convertible splits, `TableRead` can create a `MergeFileSplitRead`, whose `create_reader()` returns the normal row `RecordReader` path rather than a `RecordBatchReader`. Wrapping that in `AuthFilterReader`/`AuthMaskingReader` will fail with `AttributeError` when query auth is enabled. This needs either a row-reader auth path, conversion to a batch-capable reader before wrapping, or routing/rejecting these splits explicitly.
@JingsongLi All 3 issues fixed (+ 1 additional parallel path bypass found during analysis). See updated PR description. PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3380850757
##########
paimon-python/pypaimon/catalog/table_query_auth.py:
##########
@@ -0,0 +1,87 @@
+################################################################################
+# 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.
+################################################################################
+
+from typing import Callable, Dict, List, Optional
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+from pypaimon.common.predicate_json_parser import (
+ extract_referenced_fields,
+ parse_predicate_to_batch_filter,
+)
+from pypaimon.schema.data_types import DataField
+
+
+class TableNoPermissionException(Exception):
+ MSG = "Table %s has no permission. Cause by %s."
+
+ def __init__(self, identifier, cause=None):
+ cause_msg = str(cause) if cause else ""
+ super().__init__(self.MSG % (identifier, cause_msg))
+ self.identifier = identifier
+ self.__cause__ = cause
+
+
+class TableQueryAuthResult:
+
+ def __init__(self, filter: Optional[List[str]], column_masking: Optional[Dict[str, str]]):
+ self.filter = filter
+ self.column_masking = column_masking
+
+ def convert_plan(self, plan):
+ from pypaimon.read.query_auth_split import QueryAuthSplit
+ from pypaimon.read.plan import Plan
+
+ if not self.filter and not self.column_masking:
+ return plan
+ auth_splits = [QueryAuthSplit(split, self) for split in plan.splits()]
+ return Plan(auth_splits)
Review Comment:
Java `TableScan.Plan` does not carry a snapshot id, but Python `Plan` does and the update / row-id update paths use it as `check_from_snapshot`. Wrapping the plan here drops `plan.snapshot_id`, so a query-auth table planned from a non-empty snapshot becomes `snapshot_id=None`; `table_update` then emits commit messages with `-1`, which disables the row-id conflict checks (and related global-index update checks). Please preserve the original plan metadata, e.g. `Plan(auth_splits, snapshot_id=plan.snapshot_id)`.
##########
paimon-python/pypaimon/common/predicate_json_parser.py:
##########
@@ -0,0 +1,290 @@
+################################################################################
+# 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 re
+from typing import Callable
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+
+def parse_predicate_to_batch_filter(json_str: str) -> Callable[[pa.RecordBatch], pa.Array]:
+ data = json.loads(json_str)
+ return _build_filter(data)
+
+
+def _build_filter(data: dict) -> Callable[[pa.RecordBatch], pa.Array]:
+ kind = data["kind"]
+ if kind == "LEAF":
+ return _build_leaf_filter(data)
+ elif kind == "COMPOUND":
+ return _build_compound_filter(data)
+ raise ValueError(f"Unknown predicate kind: {kind}")
+
+
+def _build_leaf_filter(data: dict) -> Callable:
+ transform = data["transform"]
+ function = data["function"]
+ literals = data.get("literals", [])
+
+ def filter_fn(batch: pa.RecordBatch) -> pa.Array:
+ value_array = _apply_predicate_transform(transform, batch)
+ return _apply_leaf_function(function, value_array, literals, len(batch))
+
+ return filter_fn
+
+
+def _build_compound_filter(data: dict) -> Callable:
+ function = data["function"]
+ child_filters = [_build_filter(child) for child in data["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}")
+
+ return filter_fn
+
+
+def _apply_predicate_transform(transform: dict, batch: pa.RecordBatch) -> pa.Array:
+ name = transform["name"]
+
+ if name == "FIELD_REF":
+ return batch.column(transform["fieldRef"]["name"])
+
+ elif name == "CAST":
+ col = batch.column(transform["fieldRef"]["name"])
+ target_type = _paimon_type_to_arrow(transform["type"])
+ return pc.cast(col, target_type)
+
+ elif name == "UPPER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_upper(input_col)
+
+ elif name == "LOWER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_lower(input_col)
+
+ 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 pc.binary_join_element_wise(*resolved, "")
+
+ elif name == "CONCAT_WS":
+ sep = _resolve_transform_input(transform["inputs"][0], batch)
+ values = [_resolve_transform_input(inp, batch) for inp in transform["inputs"][1:]]
+ if not values:
+ return pa.nulls(len(batch), type=pa.string())
+ return pc.binary_join_element_wise(*values, sep, null_handling='skip')
+
+ elif name == "NULL":
+ return pa.nulls(len(batch), type=pa.bool_())
+
+ raise ValueError(f"Unknown transform type in predicate: {name}")
+
+
+def _resolve_transform_input(inp, batch: pa.RecordBatch) -> pa.Array:
+ if isinstance(inp, dict):
+ return batch.column(inp["name"])
+ elif isinstance(inp, str):
+ return pa.array([inp] * len(batch), type=pa.string())
+ elif inp is None:
+ return pa.nulls(len(batch), type=pa.string())
+ return pa.array([str(inp)] * len(batch), type=pa.string())
+
+
+def _apply_leaf_function(function: str, value_array: pa.Array, literals: list, batch_len: int) -> pa.Array:
+ converted = [_convert_literal(lit, value_array.type) for lit in literals]
+
+ if function == "EQUAL":
+ return pc.equal(value_array, converted[0])
+ elif function == "NOT_EQUAL":
+ return pc.not_equal(value_array, converted[0])
+ elif function == "LESS_THAN":
+ return pc.less(value_array, converted[0])
+ elif function == "LESS_OR_EQUAL":
+ return pc.less_equal(value_array, converted[0])
+ elif function == "GREATER_THAN":
+ return pc.greater(value_array, converted[0])
+ elif function == "GREATER_OR_EQUAL":
+ return pc.greater_equal(value_array, converted[0])
+ elif function == "IS_NULL":
+ return pc.is_null(value_array)
+ elif function == "IS_NOT_NULL":
+ return pc.is_valid(value_array)
+ elif function == "IN":
+ return pc.is_in(value_array, pa.array(converted, type=value_array.type))
+ elif function == "NOT_IN":
+ return pc.invert(pc.is_in(value_array, pa.array(converted, type=value_array.type)))
+ elif function == "BETWEEN":
+ return pc.and_(pc.greater_equal(value_array, converted[0]),
+ pc.less_equal(value_array, converted[1]))
+ elif function == "NOT_BETWEEN":
+ return pc.or_(pc.less(value_array, converted[0]),
+ pc.greater(value_array, converted[1]))
+ elif function == "STARTS_WITH":
+ return pc.starts_with(value_array, converted[0])
+ elif function == "ENDS_WITH":
+ return pc.ends_with(value_array, converted[0])
+ elif function == "CONTAINS":
+ return pc.match_substring(value_array, converted[0])
+ elif function == "LIKE":
+ raw = literals[0]
+ escaped = re.escape(raw)
+ pattern = escaped.replace("%", ".*").replace("_", ".")
Review Comment:
This does not match the JVM `LIKE` semantics. Java treats backslash as the default escape character before expanding `%` / `_`, so a policy predicate like `LIKE admin\\_%` matches `admin_foo` and not `adminXfoo`. Escaping the whole string first and then replacing every `%` / `_` makes escaped wildcards behave as wildcards (or requires a literal backslash), so Python can allow/deny different rows from the Java client for the same auth filter. Please port the Java `Like.sqlToRegexLike` behavior, including invalid escape handling.
##########
paimon-python/pypaimon/read/table_read.py:
##########
@@ -611,6 +611,66 @@ def _widen_to_top_level_for_merge(self) -> List[DataField]:
widened.append(field)
return widened
+ def _create_reader_for_split(self, split):
+ from pypaimon.read.query_auth_split import QueryAuthSplit
+
+ auth_result = None
+ if isinstance(split, QueryAuthSplit):
+ auth_result = split.auth_result
+ split = split.split
+
+ if auth_result is not None:
+ return self._authed_reader(split, auth_result)
+ else:
+ return self._create_split_read(split).create_reader()
+
+ def _authed_reader(self, split, auth_result):
+ from pypaimon.read.reader.auth_masking_reader import (
+ AuthFilterReader, AuthMaskingReader, ColumnProjectReader)
+
+ table_fields = self.table.fields
+ read_fields = self.read_type
+
+ extra_fields = auth_result.get_extra_fields_for_filter(read_fields, table_fields)
+ effective_read_type = read_fields
+ if extra_fields:
+ effective_read_type = read_fields + extra_fields
+
+ reader = self._create_split_read_with_read_type(split, effective_read_type).create_reader()
+
+ if not isinstance(reader, RecordBatchReader):
+ from pypaimon.read.reader.auth_masking_reader import RecordReaderToBatchAdapter
+ schema = PyarrowFieldParser.from_paimon_schema(effective_read_type)
+ reader = RecordReaderToBatchAdapter(reader, schema)
+
+ filter_fn = auth_result.extract_row_filter()
+ if filter_fn:
+ reader = AuthFilterReader(reader, filter_fn)
+
+ if auth_result.column_masking:
+ reader = AuthMaskingReader(reader, auth_result.column_masking, effective_read_type)
+
+ if extra_fields:
+ original_columns = [f.name for f in read_fields]
+ reader = ColumnProjectReader(reader, original_columns)
+
+ return reader
+
+ def _create_split_read_with_read_type(self, split, read_type):
Review Comment:
This auth-specific construction bypasses the normal PK read path above. In `_create_split_read`, PK tables inject missing `sequence.field` columns into the inner read type and then project them back out, matching the Java `withReadType` + outer projection behavior. Here, if query auth is enabled and the user projects `id,val` from a PK table with `sequence.field=ts`, `MergeFileSplitRead` is built without `ts`; that can either fail with `sequence.field ... not found` or merge by file sequence instead of the configured user sequence. Please reuse the existing `_create_split_read` widening/project-back logic for `effective_read_type`, or factor it so the auth path cannot drift from the normal PK path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3549174341
##########
paimon-python/pypaimon/write/table_update.py:
##########
@@ -245,7 +245,9 @@ def _update_by_predicate(
else:
read_builder.with_projection([SpecialFields.ROW_ID.name])
- splits = read_builder.new_scan().plan().splits()
+ scan = read_builder.new_scan()
+ scan._enable_auth = False
Review Comment:
Maybe just assert no row filter auth?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "MgjLLL (via GitHub)" <gi...@apache.org>.
MgjLLL commented on PR #8136:
URL: https://github.com/apache/paimon/pull/8136#issuecomment-4678321413
@JingsongLi All 3 issues fixed (+ several related correctness issues found during follow-up self-review). New commit `7d037c6a8` squashes the round-2 fixes into one increment on top of `482fdadd7`. PTAL.
## Fixes for the three review comments
### 1. `catalog/table_query_auth.py` — wrapping drops `Plan.snapshot_id`
`convert_plan` now returns `Plan(auth_splits, snapshot_id=plan.snapshot_id)`.
Without this, query-auth tables planned from a non-empty snapshot lost row-id conflict / global-index update checks (`check_from_snapshot=-1`). New test `TestConvertPlanPreservesSnapshotId` covers both non-null and `None` cases.
### 2. `read/table_read.py` — `_create_split_read_with_read_type` bypasses PK widening
Removed the parallel method. `_create_split_read` now accepts an optional `read_type`, so the auth path goes through the same `sequence.field` injection + outer projection used by the normal PK read path.
`id,val` projection over a PK table with `sequence.field=ts` no longer fails or merges by file sequence.
### 3. `common/predicate_json_parser.py` — `LIKE` escape mismatches JVM
Ported `Like.sqlToRegexLike`: backslash is the default escape character, `\%` / `\_` are literal, invalid escape sequences raise.
`LIKE admin\_%` now matches `admin_foo` (not `adminXfoo`), so policy predicates evaluate identically to the JVM client.
## Related correctness fixes (self-found, all on the same auth surface)
- **Streaming path** previously stored `_query_auth` but never applied it. `StreamReadBuilder` now forwards `query_auth` and `read_type` to `AsyncStreamingTableScan`, which calls `_apply_auth` on initial / follow-up / catch-up plans. Catch-up uses `_create_initial_plan_raw` to avoid double auth.
- **Iterator / parallel / Ray** paths consolidated through `_create_reader_for_split`. `to_iterator` gains a `RecordBatchReader` branch for PyTorch / generic iterator consumers. `scan_with_stats` applies auth for parity with `plan()`.
- **Row kind preserved on PK + auth**: `RecordReaderToBatchAdapter` encodes `row_kind` into a `_row_kind` column when `include_row_kind=True`; `ColumnProjectReader` keeps `_row_kind` even when user projection drops it; `to_iterator` restores it via `OffsetRow.set_row_kind_byte` without leaking the column into `row_tuple`.
- **Adapter chunking**: when an inner `read_batch` yields more rows than `chunk_size`, the surplus rows are carried over to the next flush instead of being dropped.
- **`QueryAuthSplit` attribute delegation**: `__getattr__` forwards `file_size`, `file_paths`, `data_deletion_files`, `raw_convertible`, etc. to the inner split (Ray / explain / Daft paths), while guarding `_`-prefixed names to avoid pickle recursion.
- **Daft `explain_scan` consistency**: `ExplainSplitInfo` now carries `has_auth`, populated from `QueryAuthSplit` in `_build_explain_result`. Daft's `_split_has_auth` accepts both real splits and the explain descriptor, so reported `reader_mode` and `fallback_reason="query auth active"` match the actual read path.
- **Pickle safety when auth is disabled**: `table_query_auth` returns `None` instead of a local lambda, keeping `FileStoreTable` Ray-pickle-safe.
## Tests
- New tests for snapshot_id preservation, row_kind through adapter / project reader, chunk-size carry-over, and `QueryAuthSplit` attribute delegation incl. pickle round-trip.
- 45 query-auth related tests pass:
```
pytest pypaimon/tests/ -k "explain or query_auth or table_query_auth"
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "MgjLLL (via GitHub)" <gi...@apache.org>.
MgjLLL commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3394107057
##########
paimon-python/pypaimon/common/predicate_json_parser.py:
##########
@@ -0,0 +1,290 @@
+################################################################################
+# 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 re
+from typing import Callable
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+
+def parse_predicate_to_batch_filter(json_str: str) -> Callable[[pa.RecordBatch], pa.Array]:
+ data = json.loads(json_str)
+ return _build_filter(data)
+
+
+def _build_filter(data: dict) -> Callable[[pa.RecordBatch], pa.Array]:
+ kind = data["kind"]
+ if kind == "LEAF":
+ return _build_leaf_filter(data)
+ elif kind == "COMPOUND":
+ return _build_compound_filter(data)
+ raise ValueError(f"Unknown predicate kind: {kind}")
+
+
+def _build_leaf_filter(data: dict) -> Callable:
+ transform = data["transform"]
+ function = data["function"]
+ literals = data.get("literals", [])
+
+ def filter_fn(batch: pa.RecordBatch) -> pa.Array:
+ value_array = _apply_predicate_transform(transform, batch)
+ return _apply_leaf_function(function, value_array, literals, len(batch))
+
+ return filter_fn
+
+
+def _build_compound_filter(data: dict) -> Callable:
+ function = data["function"]
+ child_filters = [_build_filter(child) for child in data["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}")
+
+ return filter_fn
+
+
+def _apply_predicate_transform(transform: dict, batch: pa.RecordBatch) -> pa.Array:
+ name = transform["name"]
+
+ if name == "FIELD_REF":
+ return batch.column(transform["fieldRef"]["name"])
+
+ elif name == "CAST":
+ col = batch.column(transform["fieldRef"]["name"])
+ target_type = _paimon_type_to_arrow(transform["type"])
+ return pc.cast(col, target_type)
+
+ elif name == "UPPER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_upper(input_col)
+
+ elif name == "LOWER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_lower(input_col)
+
+ 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 pc.binary_join_element_wise(*resolved, "")
+
+ elif name == "CONCAT_WS":
+ sep = _resolve_transform_input(transform["inputs"][0], batch)
+ values = [_resolve_transform_input(inp, batch) for inp in transform["inputs"][1:]]
+ if not values:
+ return pa.nulls(len(batch), type=pa.string())
+ return pc.binary_join_element_wise(*values, sep, null_handling='skip')
+
+ elif name == "NULL":
+ return pa.nulls(len(batch), type=pa.bool_())
+
+ raise ValueError(f"Unknown transform type in predicate: {name}")
+
+
+def _resolve_transform_input(inp, batch: pa.RecordBatch) -> pa.Array:
+ if isinstance(inp, dict):
+ return batch.column(inp["name"])
+ elif isinstance(inp, str):
+ return pa.array([inp] * len(batch), type=pa.string())
+ elif inp is None:
+ return pa.nulls(len(batch), type=pa.string())
+ return pa.array([str(inp)] * len(batch), type=pa.string())
+
+
+def _apply_leaf_function(function: str, value_array: pa.Array, literals: list, batch_len: int) -> pa.Array:
+ converted = [_convert_literal(lit, value_array.type) for lit in literals]
+
+ if function == "EQUAL":
+ return pc.equal(value_array, converted[0])
+ elif function == "NOT_EQUAL":
+ return pc.not_equal(value_array, converted[0])
+ elif function == "LESS_THAN":
+ return pc.less(value_array, converted[0])
+ elif function == "LESS_OR_EQUAL":
+ return pc.less_equal(value_array, converted[0])
+ elif function == "GREATER_THAN":
+ return pc.greater(value_array, converted[0])
+ elif function == "GREATER_OR_EQUAL":
+ return pc.greater_equal(value_array, converted[0])
+ elif function == "IS_NULL":
+ return pc.is_null(value_array)
+ elif function == "IS_NOT_NULL":
+ return pc.is_valid(value_array)
+ elif function == "IN":
+ return pc.is_in(value_array, pa.array(converted, type=value_array.type))
+ elif function == "NOT_IN":
+ return pc.invert(pc.is_in(value_array, pa.array(converted, type=value_array.type)))
+ elif function == "BETWEEN":
+ return pc.and_(pc.greater_equal(value_array, converted[0]),
+ pc.less_equal(value_array, converted[1]))
+ elif function == "NOT_BETWEEN":
+ return pc.or_(pc.less(value_array, converted[0]),
+ pc.greater(value_array, converted[1]))
+ elif function == "STARTS_WITH":
+ return pc.starts_with(value_array, converted[0])
+ elif function == "ENDS_WITH":
+ return pc.ends_with(value_array, converted[0])
+ elif function == "CONTAINS":
+ return pc.match_substring(value_array, converted[0])
+ elif function == "LIKE":
+ raw = literals[0]
+ escaped = re.escape(raw)
+ pattern = escaped.replace("%", ".*").replace("_", ".")
Review Comment:
Ported Like.sqlToRegexLike: backslash is the default escape character, \\% / \\_ are literal, invalid escape sequences raise. LIKE admin\_% now matches admin_foo (not adminXfoo), so policy predicates evaluate identically to the JVM client.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "MgjLLL (via GitHub)" <gi...@apache.org>.
MgjLLL commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3394104278
##########
paimon-python/pypaimon/catalog/table_query_auth.py:
##########
@@ -0,0 +1,87 @@
+################################################################################
+# 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.
+################################################################################
+
+from typing import Callable, Dict, List, Optional
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+from pypaimon.common.predicate_json_parser import (
+ extract_referenced_fields,
+ parse_predicate_to_batch_filter,
+)
+from pypaimon.schema.data_types import DataField
+
+
+class TableNoPermissionException(Exception):
+ MSG = "Table %s has no permission. Cause by %s."
+
+ def __init__(self, identifier, cause=None):
+ cause_msg = str(cause) if cause else ""
+ super().__init__(self.MSG % (identifier, cause_msg))
+ self.identifier = identifier
+ self.__cause__ = cause
+
+
+class TableQueryAuthResult:
+
+ def __init__(self, filter: Optional[List[str]], column_masking: Optional[Dict[str, str]]):
+ self.filter = filter
+ self.column_masking = column_masking
+
+ def convert_plan(self, plan):
+ from pypaimon.read.query_auth_split import QueryAuthSplit
+ from pypaimon.read.plan import Plan
+
+ if not self.filter and not self.column_masking:
+ return plan
+ auth_splits = [QueryAuthSplit(split, self) for split in plan.splits()]
+ return Plan(auth_splits)
Review Comment:
convert_plan now returns Plan(auth_splits, snapshot_id=plan.snapshot_id). Without this, query-auth tables planned from a non-empty snapshot lost row-id conflict / global-index update checks (check_from_snapshot=-1). New test TestConvertPlanPreservesSnapshotId covers both non-null and None cases.
##########
paimon-python/pypaimon/read/table_read.py:
##########
@@ -611,6 +611,66 @@ def _widen_to_top_level_for_merge(self) -> List[DataField]:
widened.append(field)
return widened
+ def _create_reader_for_split(self, split):
+ from pypaimon.read.query_auth_split import QueryAuthSplit
+
+ auth_result = None
+ if isinstance(split, QueryAuthSplit):
+ auth_result = split.auth_result
+ split = split.split
+
+ if auth_result is not None:
+ return self._authed_reader(split, auth_result)
+ else:
+ return self._create_split_read(split).create_reader()
+
+ def _authed_reader(self, split, auth_result):
+ from pypaimon.read.reader.auth_masking_reader import (
+ AuthFilterReader, AuthMaskingReader, ColumnProjectReader)
+
+ table_fields = self.table.fields
+ read_fields = self.read_type
+
+ extra_fields = auth_result.get_extra_fields_for_filter(read_fields, table_fields)
+ effective_read_type = read_fields
+ if extra_fields:
+ effective_read_type = read_fields + extra_fields
+
+ reader = self._create_split_read_with_read_type(split, effective_read_type).create_reader()
+
+ if not isinstance(reader, RecordBatchReader):
+ from pypaimon.read.reader.auth_masking_reader import RecordReaderToBatchAdapter
+ schema = PyarrowFieldParser.from_paimon_schema(effective_read_type)
+ reader = RecordReaderToBatchAdapter(reader, schema)
+
+ filter_fn = auth_result.extract_row_filter()
+ if filter_fn:
+ reader = AuthFilterReader(reader, filter_fn)
+
+ if auth_result.column_masking:
+ reader = AuthMaskingReader(reader, auth_result.column_masking, effective_read_type)
+
+ if extra_fields:
+ original_columns = [f.name for f in read_fields]
+ reader = ColumnProjectReader(reader, original_columns)
+
+ return reader
+
+ def _create_split_read_with_read_type(self, split, read_type):
Review Comment:
Removed the parallel method. _create_split_read now accepts an optional read_type, so the auth path goes through the same sequence.field injection + outer projection used by the normal PK read path. id,val projection over a PK table with sequence.field=ts no longer fails or merges by file sequence.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3450555496
##########
paimon-python/pypaimon/catalog/catalog_environment.py:
##########
@@ -117,3 +117,12 @@ def empty() -> 'CatalogEnvironment':
catalog_loader=None,
supports_version_management=False
)
+
+ def table_query_auth(self, options, identifier):
+ if not options.query_auth_enabled or self.catalog_loader is None:
+ return None
+
+ def auth(select):
Review Comment:
This nested `auth` function is stored on `FileStoreTable` as `_query_auth_fn`, which makes auth-enabled tables/read tasks unpickleable (`AttributeError: Can't get local object ...`). Ray/Daft and the existing serialization paths pickle tables or read tasks, so this breaks those paths once `query-auth.enabled` is true. Please use a top-level callable/dataclass or reconstruct the catalog auth function lazily instead of storing a local closure.
##########
paimon-python/pypaimon/catalog/table_query_auth.py:
##########
@@ -0,0 +1,87 @@
+################################################################################
+# 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.
+################################################################################
+
+from typing import Callable, Dict, List, Optional
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+from pypaimon.common.predicate_json_parser import (
+ extract_referenced_fields,
+ parse_predicate_to_batch_filter,
+)
+from pypaimon.schema.data_types import DataField
+
+
+class TableNoPermissionException(Exception):
Review Comment:
This defines a second `TableNoPermissionException` that is unrelated to `pypaimon.catalog.catalog_exception.TableNoPermissionException`. `RESTCatalog.auth_table_query` raises this new type on 403, so callers catching the catalog exception will miss query-auth denials. Please reuse the existing catalog exception or make this subclass it.
##########
paimon-python/pypaimon/common/predicate_json_parser.py:
##########
@@ -0,0 +1,344 @@
+################################################################################
+# 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 re
+from typing import Callable
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+
+def parse_predicate_to_batch_filter(json_str: str) -> Callable[[pa.RecordBatch], pa.Array]:
+ data = json.loads(json_str)
+ return _build_filter(data)
+
+
+def _build_filter(data: dict) -> Callable[[pa.RecordBatch], pa.Array]:
+ kind = data["kind"]
+ if kind == "LEAF":
+ return _build_leaf_filter(data)
+ elif kind == "COMPOUND":
+ return _build_compound_filter(data)
+ raise ValueError(f"Unknown predicate kind: {kind}")
+
+
+def _build_leaf_filter(data: dict) -> Callable:
+ transform = data["transform"]
+ function = data["function"]
+ literals = data.get("literals", [])
+
+ def filter_fn(batch: pa.RecordBatch) -> pa.Array:
+ value_array = _apply_predicate_transform(transform, batch)
+ return _apply_leaf_function(function, value_array, literals, len(batch))
+
+ return filter_fn
+
+
+def _build_compound_filter(data: dict) -> Callable:
+ function = data["function"]
+ child_filters = [_build_filter(child) for child in data["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}")
+
+ return filter_fn
+
+
+def _apply_predicate_transform(transform: dict, batch: pa.RecordBatch) -> pa.Array:
+ name = transform["name"]
+
+ if name == "FIELD_REF":
+ return batch.column(transform["fieldRef"]["name"])
+
+ elif name == "CAST":
+ col = batch.column(transform["fieldRef"]["name"])
+ target_type = _paimon_type_to_arrow(transform["type"])
+ return pc.cast(col, target_type)
+
+ elif name == "UPPER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_upper(input_col)
+
+ elif name == "LOWER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_lower(input_col)
+
+ 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 pc.binary_join_element_wise(*resolved, "")
+
+ elif name == "CONCAT_WS":
+ sep = _resolve_transform_input(transform["inputs"][0], batch)
+ values = [_resolve_transform_input(inp, batch) for inp in transform["inputs"][1:]]
+ if not values:
+ return pa.nulls(len(batch), type=pa.string())
+ return pc.binary_join_element_wise(*values, sep, null_handling='skip')
+
+ elif name == "NULL":
+ return pa.nulls(len(batch), type=pa.bool_())
+
+ raise ValueError(f"Unknown transform type in predicate: {name}")
+
+
+def _resolve_transform_input(inp, batch: pa.RecordBatch) -> pa.Array:
+ if isinstance(inp, dict):
+ return batch.column(inp["name"])
+ elif isinstance(inp, str):
+ return pa.array([inp] * len(batch), type=pa.string())
+ elif inp is None:
+ return pa.nulls(len(batch), type=pa.string())
+ return pa.array([str(inp)] * len(batch), type=pa.string())
+
+
+def _apply_leaf_function(function: str, value_array: pa.Array, literals: list, batch_len: int) -> pa.Array:
+ """Null literal yields False to match Java LeafBinaryFunction/LeafTernaryFunction semantics."""
+ converted = [_convert_literal(lit, value_array.type) for lit in literals]
+
+ if function == "EQUAL":
+ return pc.equal(value_array, converted[0])
+ elif function == "NOT_EQUAL":
+ return pc.not_equal(value_array, converted[0])
+ elif function == "LESS_THAN":
+ return pc.less(value_array, converted[0])
+ elif function == "LESS_OR_EQUAL":
+ return pc.less_equal(value_array, converted[0])
+ elif function == "GREATER_THAN":
+ return pc.greater(value_array, converted[0])
+ elif function == "GREATER_OR_EQUAL":
+ return pc.greater_equal(value_array, converted[0])
+ elif function == "IS_NULL":
+ return pc.is_null(value_array)
+ elif function == "IS_NOT_NULL":
+ return pc.is_valid(value_array)
+ elif function == "IN":
+ return pc.is_in(value_array, pa.array(converted, type=value_array.type))
Review Comment:
Java's `In.test` returns false when the field value is null and skips null literals. `pc.is_in` treats `NULL IN (..., NULL)` as true, so an auth row filter like `dept IN ('eng', NULL)` would leak rows where `dept` is null. Please drop null literals and wrap the result with `pc.if_else(pc.is_valid(value_array), ..., False)`; if no non-null literals remain, the result should be all false.
##########
paimon-python/pypaimon/read/reader/auth_masking_reader.py:
##########
@@ -0,0 +1,224 @@
+################################################################################
+# 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
+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,
+ _paimon_type_to_arrow,
+)
+from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
+
+
+class RecordReaderToBatchAdapter(RecordBatchReader):
+
+ def __init__(self, inner, schema: pa.Schema, chunk_size: int = 65536, include_row_kind: bool = False):
+ self._inner = inner
+ self._schema = schema
+ self._chunk_size = chunk_size
+ self._exhausted = False
+ self._pending_iterator = None
+ self._include_row_kind = include_row_kind
+
+ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
+ if self._exhausted:
+ return None
+ row_tuples = []
+ row_kinds = []
+ while len(row_tuples) < self._chunk_size:
+ if self._pending_iterator is not None:
+ row = self._pending_iterator.next()
+ while row is not None:
+ row_tuples.append(
+ row.row_tuple[row.offset:row.offset + row.arity])
+ if self._include_row_kind:
+ row_kinds.append(row.get_row_kind().to_string())
+ if len(row_tuples) >= self._chunk_size:
+ return self._flush(row_tuples, row_kinds)
+ row = self._pending_iterator.next()
+ self._pending_iterator = None
+
+ row_iterator = self._inner.read_batch()
+ if row_iterator is None:
+ self._exhausted = True
+ break
+ self._pending_iterator = row_iterator
+
+ if not row_tuples:
+ return None
+ return self._flush(row_tuples, row_kinds)
+
+ def _flush(self, row_tuples, row_kinds=None):
+ columns_data = list(zip(*row_tuples))
+ pydict = {
+ name: list(col)
+ for name, col in zip(self._schema.names, columns_data)
+ }
+ batch = pa.RecordBatch.from_pydict(pydict, schema=self._schema)
+ if row_kinds:
+ row_kind_array = pa.array(row_kinds, type=pa.string())
+ row_kind_field = pa.field("_row_kind", pa.string())
+ new_schema = pa.schema([row_kind_field] + list(batch.schema))
+ columns = [row_kind_array] + [batch.column(i) for i in range(batch.num_columns)]
+ batch = pa.RecordBatch.from_arrays(columns, schema=new_schema)
+ return batch
+
+ def close(self):
+ self._inner.close()
+
+
+class AuthFilterReader(RecordBatchReader):
+
+ def __init__(self, inner_reader: RecordBatchReader, filter_fn: Callable[[pa.RecordBatch], pa.Array]):
+ self._inner = inner_reader
+ self._filter_fn = filter_fn
+
+ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
+ batch = self._inner.read_arrow_batch()
+ if batch is None:
+ return None
+ mask = self._filter_fn(batch)
+ return batch.filter(mask)
+
+ def close(self):
+ self._inner.close()
+
+
+class AuthMaskingReader(RecordBatchReader):
+
+ def __init__(self, inner_reader: RecordBatchReader, masking_rules: Dict[str, str], read_fields: List):
+ self._inner = inner_reader
+ self._masking_rules = masking_rules
+ self._read_fields = read_fields
+ read_field_names = {f.name for f in read_fields}
+ # Filter to projected columns, then validate field references.
+ self._parsed_rules = {
+ col: json.loads(tj) for col, tj in masking_rules.items()
+ if col in read_field_names
+ }
+ for col_name, transform in self._parsed_rules.items():
+ for ref_name in _collect_all_field_refs_from_transform(transform):
+ if ref_name not in read_field_names:
+ raise RuntimeError(
+ f"Column masking refers to field '{ref_name}' which is not present "
+ f"in output row type. Available fields: {read_field_names}"
+ )
+
+ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
+ batch = self._inner.read_arrow_batch()
+ if batch is None:
+ return None
+ original_batch = batch
+ masked_columns = {}
+ for col_name, transform in self._parsed_rules.items():
+ if col_name in original_batch.schema.names:
+ col_idx = original_batch.schema.get_field_index(col_name)
+ target_col_type = original_batch.schema.field(col_idx).type
+ masked_columns[col_idx] = self._apply_transform(transform, original_batch, target_col_type)
+ for col_idx, masked_array in masked_columns.items():
+ original_field = original_batch.schema.field(col_idx)
+ if masked_array.type != original_field.type:
+ masked_array = pc.cast(masked_array, original_field.type)
+ batch = batch.set_column(
+ col_idx, pa.field(original_field.name, original_field.type, nullable=True), masked_array)
+ return batch
+
+ def close(self):
+ self._inner.close()
+
+ def _apply_transform(
+ self,
+ transform: dict,
+ original_batch: pa.RecordBatch,
+ target_col_type: pa.DataType,
+ ) -> pa.Array:
+ name = transform["name"]
+
+ if name == "NULL":
+ return pa.nulls(len(original_batch), type=target_col_type)
+
+ elif name == "FIELD_REF":
+ ref_name = transform["fieldRef"]["name"]
+ return original_batch.column(ref_name)
+
+ elif name == "CAST":
+ ref_name = transform["fieldRef"]["name"]
+ source_col = original_batch.column(ref_name)
+ target_type = _paimon_type_to_arrow(transform["type"])
+ return pc.cast(source_col, target_type)
+
+ elif name == "UPPER":
+ input_col = self._resolve_input(transform["inputs"][0], original_batch)
+ return pc.utf8_upper(input_col)
+
+ elif name == "LOWER":
+ input_col = self._resolve_input(transform["inputs"][0], original_batch)
+ return pc.utf8_lower(input_col)
+
+ elif name == "CONCAT":
+ return self._apply_concat(transform["inputs"], original_batch)
+
+ elif name == "CONCAT_WS":
+ return self._apply_concat_ws(transform["inputs"], original_batch)
+
+ raise ValueError(f"Unknown transform type: {name}")
+
+ def _resolve_input(self, inp, original_batch: pa.RecordBatch) -> pa.Array:
+ if isinstance(inp, dict):
+ return original_batch.column(inp["name"])
+ elif isinstance(inp, str):
+ return pa.array([inp] * len(original_batch), type=pa.string())
+ elif inp is None:
+ return pa.nulls(len(original_batch), type=pa.string())
+ return pa.array([str(inp)] * len(original_batch), type=pa.string())
+
+ def _apply_concat(self, inputs: list, original_batch: pa.RecordBatch) -> pa.Array:
+ resolved = [self._resolve_input(inp, original_batch) for inp in inputs]
+ if not resolved:
+ return pa.nulls(len(original_batch), type=pa.string())
+ return pc.binary_join_element_wise(*resolved, "")
+
+ def _apply_concat_ws(self, inputs: list, original_batch: pa.RecordBatch) -> pa.Array:
+ if len(inputs) < 2:
+ return pa.nulls(len(original_batch), type=pa.string())
+ sep = self._resolve_input(inputs[0], original_batch)
+ values = [self._resolve_input(inp, original_batch) for inp in inputs[1:]]
+ return pc.binary_join_element_wise(*values, sep, null_handling='skip')
Review Comment:
`binary_join_element_wise(..., null_handling='skip')` does not match Java `BinaryString.concatWs` when the separator is non-null and all value inputs for a row are null. Java returns an empty string for that row, but Arrow returns a shorter array, so `set_column` can fail with `ArrowInvalid: Added column's length must match record batch's length`. Please normalize this case so the result always has `len(original_batch)` rows; the predicate-side `CONCAT_WS` path has the same issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "MgjLLL (via GitHub)" <gi...@apache.org>.
MgjLLL commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3458015698
##########
paimon-python/pypaimon/common/predicate_json_parser.py:
##########
@@ -0,0 +1,344 @@
+################################################################################
+# 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 re
+from typing import Callable
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+
+def parse_predicate_to_batch_filter(json_str: str) -> Callable[[pa.RecordBatch], pa.Array]:
+ data = json.loads(json_str)
+ return _build_filter(data)
+
+
+def _build_filter(data: dict) -> Callable[[pa.RecordBatch], pa.Array]:
+ kind = data["kind"]
+ if kind == "LEAF":
+ return _build_leaf_filter(data)
+ elif kind == "COMPOUND":
+ return _build_compound_filter(data)
+ raise ValueError(f"Unknown predicate kind: {kind}")
+
+
+def _build_leaf_filter(data: dict) -> Callable:
+ transform = data["transform"]
+ function = data["function"]
+ literals = data.get("literals", [])
+
+ def filter_fn(batch: pa.RecordBatch) -> pa.Array:
+ value_array = _apply_predicate_transform(transform, batch)
+ return _apply_leaf_function(function, value_array, literals, len(batch))
+
+ return filter_fn
+
+
+def _build_compound_filter(data: dict) -> Callable:
+ function = data["function"]
+ child_filters = [_build_filter(child) for child in data["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}")
+
+ return filter_fn
+
+
+def _apply_predicate_transform(transform: dict, batch: pa.RecordBatch) -> pa.Array:
+ name = transform["name"]
+
+ if name == "FIELD_REF":
+ return batch.column(transform["fieldRef"]["name"])
+
+ elif name == "CAST":
+ col = batch.column(transform["fieldRef"]["name"])
+ target_type = _paimon_type_to_arrow(transform["type"])
+ return pc.cast(col, target_type)
+
+ elif name == "UPPER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_upper(input_col)
+
+ elif name == "LOWER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_lower(input_col)
+
+ 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 pc.binary_join_element_wise(*resolved, "")
+
+ elif name == "CONCAT_WS":
+ sep = _resolve_transform_input(transform["inputs"][0], batch)
+ values = [_resolve_transform_input(inp, batch) for inp in transform["inputs"][1:]]
+ if not values:
+ return pa.nulls(len(batch), type=pa.string())
+ return pc.binary_join_element_wise(*values, sep, null_handling='skip')
+
+ elif name == "NULL":
+ return pa.nulls(len(batch), type=pa.bool_())
+
+ raise ValueError(f"Unknown transform type in predicate: {name}")
+
+
+def _resolve_transform_input(inp, batch: pa.RecordBatch) -> pa.Array:
+ if isinstance(inp, dict):
+ return batch.column(inp["name"])
+ elif isinstance(inp, str):
+ return pa.array([inp] * len(batch), type=pa.string())
+ elif inp is None:
+ return pa.nulls(len(batch), type=pa.string())
+ return pa.array([str(inp)] * len(batch), type=pa.string())
+
+
+def _apply_leaf_function(function: str, value_array: pa.Array, literals: list, batch_len: int) -> pa.Array:
+ """Null literal yields False to match Java LeafBinaryFunction/LeafTernaryFunction semantics."""
+ converted = [_convert_literal(lit, value_array.type) for lit in literals]
+
+ if function == "EQUAL":
+ return pc.equal(value_array, converted[0])
+ elif function == "NOT_EQUAL":
+ return pc.not_equal(value_array, converted[0])
+ elif function == "LESS_THAN":
+ return pc.less(value_array, converted[0])
+ elif function == "LESS_OR_EQUAL":
+ return pc.less_equal(value_array, converted[0])
+ elif function == "GREATER_THAN":
+ return pc.greater(value_array, converted[0])
+ elif function == "GREATER_OR_EQUAL":
+ return pc.greater_equal(value_array, converted[0])
+ elif function == "IS_NULL":
+ return pc.is_null(value_array)
+ elif function == "IS_NOT_NULL":
+ return pc.is_valid(value_array)
+ elif function == "IN":
+ return pc.is_in(value_array, pa.array(converted, type=value_array.type))
Review Comment:
Fixed. Now drop null literals from the value set and wrap with pc.if_else(pc.is_valid(value_array), in_mask, False). If all literals are null, returns all-false. Also added null literal guards for EQUAL/NOT_EQUAL/LT/LE/GT/GE to match Java's LeafBinaryFunction semantics.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "MgjLLL (via GitHub)" <gi...@apache.org>.
MgjLLL commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3458049666
##########
paimon-python/pypaimon/catalog/catalog_environment.py:
##########
@@ -117,3 +117,12 @@ def empty() -> 'CatalogEnvironment':
catalog_loader=None,
supports_version_management=False
)
+
+ def table_query_auth(self, options, identifier):
+ if not options.query_auth_enabled or self.catalog_loader is None:
+ return None
+
+ def auth(select):
Review Comment:
Fixed. Replaced the nested closure with a top-level `_TableQueryAuthFn` class that is pickleable. Verified with pickle roundtrip test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3519595155
##########
paimon-python/pypaimon/read/table_scan.py:
##########
@@ -42,10 +42,23 @@ def __init__(
self.predicate = predicate
self.limit = limit
self.partition_predicate = partition_predicate
+ self._read_type = None
self.file_scanner = self._create_file_scanner()
def plan(self) -> Plan:
- return self.file_scanner.scan()
+ auth_result = self._auth_query()
+ plan = self.file_scanner.scan()
+ if auth_result is not None:
+ plan = auth_result.convert_plan(plan)
Review Comment:
This auth wrapping is only applied by `plan()`, but `scan_with_stats()` below still returns the raw scanner plan. `ReadBuilder.explain()` uses `scan_with_stats()`, and Daft explain derives routing from those split details; the actual Daft scan path sees `QueryAuthSplit` and falls back to pypaimon, while explain will still report native Parquet for the same auth-enabled scan. Please apply the same `_auth_query().convert_plan(...)` logic in `scan_with_stats()` while preserving the `ScanStats`, and add an explain regression for an auth-enabled table.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "MgjLLL (via GitHub)" <gi...@apache.org>.
MgjLLL commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3394104278
##########
paimon-python/pypaimon/catalog/table_query_auth.py:
##########
@@ -0,0 +1,87 @@
+################################################################################
+# 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.
+################################################################################
+
+from typing import Callable, Dict, List, Optional
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+from pypaimon.common.predicate_json_parser import (
+ extract_referenced_fields,
+ parse_predicate_to_batch_filter,
+)
+from pypaimon.schema.data_types import DataField
+
+
+class TableNoPermissionException(Exception):
+ MSG = "Table %s has no permission. Cause by %s."
+
+ def __init__(self, identifier, cause=None):
+ cause_msg = str(cause) if cause else ""
+ super().__init__(self.MSG % (identifier, cause_msg))
+ self.identifier = identifier
+ self.__cause__ = cause
+
+
+class TableQueryAuthResult:
+
+ def __init__(self, filter: Optional[List[str]], column_masking: Optional[Dict[str, str]]):
+ self.filter = filter
+ self.column_masking = column_masking
+
+ def convert_plan(self, plan):
+ from pypaimon.read.query_auth_split import QueryAuthSplit
+ from pypaimon.read.plan import Plan
+
+ if not self.filter and not self.column_masking:
+ return plan
+ auth_splits = [QueryAuthSplit(split, self) for split in plan.splits()]
+ return Plan(auth_splits)
Review Comment:
convert_plan now returns Plan(auth_splits, snapshot_id=plan.snapshot_id). Without this, query-auth tables planned from a non-empty snapshot lost row-id conflict / global-index update checks (check_from_snapshot=-1). New test TestConvertPlanPreservesSnapshotId covers both non-null and None cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "MgjLLL (via GitHub)" <gi...@apache.org>.
MgjLLL commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3394106225
##########
paimon-python/pypaimon/read/table_read.py:
##########
@@ -611,6 +611,66 @@ def _widen_to_top_level_for_merge(self) -> List[DataField]:
widened.append(field)
return widened
+ def _create_reader_for_split(self, split):
+ from pypaimon.read.query_auth_split import QueryAuthSplit
+
+ auth_result = None
+ if isinstance(split, QueryAuthSplit):
+ auth_result = split.auth_result
+ split = split.split
+
+ if auth_result is not None:
+ return self._authed_reader(split, auth_result)
+ else:
+ return self._create_split_read(split).create_reader()
+
+ def _authed_reader(self, split, auth_result):
+ from pypaimon.read.reader.auth_masking_reader import (
+ AuthFilterReader, AuthMaskingReader, ColumnProjectReader)
+
+ table_fields = self.table.fields
+ read_fields = self.read_type
+
+ extra_fields = auth_result.get_extra_fields_for_filter(read_fields, table_fields)
+ effective_read_type = read_fields
+ if extra_fields:
+ effective_read_type = read_fields + extra_fields
+
+ reader = self._create_split_read_with_read_type(split, effective_read_type).create_reader()
+
+ if not isinstance(reader, RecordBatchReader):
+ from pypaimon.read.reader.auth_masking_reader import RecordReaderToBatchAdapter
+ schema = PyarrowFieldParser.from_paimon_schema(effective_read_type)
+ reader = RecordReaderToBatchAdapter(reader, schema)
+
+ filter_fn = auth_result.extract_row_filter()
+ if filter_fn:
+ reader = AuthFilterReader(reader, filter_fn)
+
+ if auth_result.column_masking:
+ reader = AuthMaskingReader(reader, auth_result.column_masking, effective_read_type)
+
+ if extra_fields:
+ original_columns = [f.name for f in read_fields]
+ reader = ColumnProjectReader(reader, original_columns)
+
+ return reader
+
+ def _create_split_read_with_read_type(self, split, read_type):
Review Comment:
Removed the parallel method. _create_split_read now accepts an optional read_type, so the auth path goes through the same sequence.field injection + outer projection used by the normal PK read path. id,val projection over a PK table with sequence.field=ts no longer fails or merges by file sequence.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "MgjLLL (via GitHub)" <gi...@apache.org>.
MgjLLL commented on PR #8136:
URL: https://github.com/apache/paimon/pull/8136#issuecomment-4648661464
Fixes for issues raised by @JingsongLi, plus one additional issue found during analysis.
8 files changed, +188 -13 lines.
#### Fix 1: Ray read path bypasses auth (`split_provider.py`)
`_ensure_planned()` directly constructed `ReadBuilder(table)`, bypassing the `query_auth` injection from `FileStoreTable.new_read_builder()`.
**Fix:** Changed to `self._ensure_table().new_read_builder()`. Removed the direct `ReadBuilder` import.
#### Fix 2: Streaming read path skips auth entirely (`streaming_table_scan.py`, `stream_read_builder.py`)
`StreamReadBuilder` stored `_query_auth` but never passed it to `AsyncStreamingTableScan`. All three plan creation methods (initial/follow-up/catch-up) produced plans without auth wrapping.
**Fix:**
- Added `query_auth` parameter to `AsyncStreamingTableScan.__init__()`
- Added `_apply_auth(plan)` method that calls `query_auth(select)` → `convert_plan(plan)`
- Applied auth to `_create_initial_plan`, `_create_follow_up_plan`, `_create_catch_up_plan`
- Extracted `_create_initial_plan_raw()` to avoid double auth in catch-up path
- `StreamReadBuilder.new_streaming_scan()` now passes `query_auth`
#### Fix 3: Primary-key table merge reader incompatible with auth wrappers (`auth_masking_reader.py`, `table_read.py`)
`MergeFileSplitRead.create_reader()` returns a `RecordReader` (row-level), but `AuthFilterReader`/`AuthMaskingReader` require `RecordBatchReader` (batch-level with `read_arrow_batch()`). Java has a unified `RecordReader<InternalRow>` so this
mismatch doesn't exist in JVM.
**Fix:**
- Added `RecordReaderToBatchAdapter(RecordBatchReader)` that collects `OffsetRow` tuples from `read_batch()`/`next()` and converts them to `pa.RecordBatch`
- `_authed_reader()` now detects non-`RecordBatchReader` and wraps with the adapter before applying auth wrappers
#### Fix 4: Parallel read path bypasses auth + missing `raw_convertible` proxy (`table_read.py`, `query_auth_split.py`)
`_read_one_split_to_batches()` called `_create_split_read(split).create_reader()` directly, bypassing `QueryAuthSplit` detection. Additionally, `QueryAuthSplit` lacked a `raw_convertible` property proxy, causing `AttributeError` when accessed.
**Fix:**
- Changed to `self._create_reader_for_split(split)` which correctly handles `QueryAuthSplit`
- Added `raw_convertible` property proxy to `QueryAuthSplit`
### New Tests (for fixes)
- `TestQueryAuthSplitRawConvertible` (2 tests) — verifies `raw_convertible` proxy for true/false
- `TestRecordReaderToBatchAdapter` (5 tests) — basic conversion, multi-batch, empty reader, close delegation, integration with `AuthFilterReader`
All 90 auth-related tests pass. Full test suite: 318/324 pass (6 pre-existing failures unrelated to this PR).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3465167175
##########
paimon-python/pypaimon/table/file_store_table.py:
##########
@@ -407,11 +410,13 @@ def bucket_mode(self) -> BucketMode:
else:
return BucketMode.HASH_FIXED
- def new_read_builder(self) -> 'ReadBuilder':
- return ReadBuilder(self)
+ def new_read_builder(self, *, _skip_auth=False) -> 'ReadBuilder':
Review Comment:
Why adding `_skip_auth`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "MgjLLL (via GitHub)" <gi...@apache.org>.
MgjLLL commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3464269826
##########
paimon-python/pypaimon/catalog/table_query_auth.py:
##########
@@ -0,0 +1,77 @@
+################################################################################
+# 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.
+################################################################################
+
+from typing import Callable, Dict, List, Optional
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+from pypaimon.common.predicate_json_parser import (
+ extract_referenced_fields,
+ parse_predicate_to_batch_filter,
+)
+from pypaimon.schema.data_types import DataField
+
+
+class TableQueryAuthResult:
+
+ def __init__(self, filter: Optional[List[str]], column_masking: Optional[Dict[str, str]]):
+ self.filter = filter
+ self.column_masking = column_masking
+
+ def convert_plan(self, plan):
+ from pypaimon.read.query_auth_split import QueryAuthSplit
+ from pypaimon.read.plan import Plan
+
+ if not self.filter and not self.column_masking:
+ return plan
+ auth_splits = [QueryAuthSplit(split, self) for split in plan.splits()]
+ return Plan(auth_splits, snapshot_id=plan.snapshot_id)
+
+ 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]
Review Comment:
Fixed. Now filter empty/blank entries at the source — TableQueryAuthResult constructor strips falsy entries from the filter list (`[f for f in filter if f]`) and empty keys/values from column_masking dict. AuthMaskingReader also guards `if not tj: continue` before json.loads. Python truthiness check (`if f`) matches Java StringUtils.isEmpty semantics: skips null and "" but passes through whitespace strings. Added regression tests for mixed empty/valid filters and empty masking values.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "MgjLLL (via GitHub)" <gi...@apache.org>.
MgjLLL commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3465388651
##########
paimon-python/pypaimon/table/file_store_table.py:
##########
@@ -407,11 +410,13 @@ def bucket_mode(self) -> BucketMode:
else:
return BucketMode.HASH_FIXED
- def new_read_builder(self) -> 'ReadBuilder':
- return ReadBuilder(self)
+ def new_read_builder(self, *, _skip_auth=False) -> 'ReadBuilder':
Review Comment:
The Python write paths (e.g. `table_update.py`, `table_upsert_by_key.py`,
`file_store_write.py`) internally call `new_read_builder()` to read existing
data for operations like UPDATE, UPSERT, and MERGE INTO. This is a pre-existing
design that predates this PR.
Once query auth is injected into `new_read_builder()`, those internal reads
would also be subject to row filtering and column masking — returning a subset
of rows or masked column values to the write engine, which would corrupt the
write results.
`_skip_auth=True` bypasses auth for these internal, system-initiated reads,
matching Java's design: Java's write paths operate at a lower abstraction level
(e.g. `LocalTableQuery`, `SortedGlobalIndexBuilder`) that never passes through
`AbstractDataTableScan`'s auth injection point in the first place.
The leading underscore signals that this is an internal API not intended for
external callers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3458088218
##########
paimon-python/pypaimon/read/table_read.py:
##########
@@ -687,6 +720,51 @@ def _widen_to_top_level_for_merge(self) -> List[DataField]:
widened.append(field)
return widened
+ def _create_reader_for_split(self, split):
+ from pypaimon.read.query_auth_split import QueryAuthSplit
+
+ auth_result = None
+ if isinstance(split, QueryAuthSplit):
+ auth_result = split.auth_result
+ split = split.split
+
+ if auth_result is not None:
+ return self._authed_reader(split, auth_result)
+ else:
+ return self._create_split_read(split).create_reader()
+
+ def _authed_reader(self, split, auth_result):
+ from pypaimon.read.reader.auth_masking_reader import (
+ AuthFilterReader, AuthMaskingReader, ColumnProjectReader)
+
+ table_fields = self.table.fields
+ read_fields = self.read_type
+
+ extra_fields = auth_result.get_extra_fields_for_filter(read_fields, table_fields)
+ effective_read_type = read_fields
+ if extra_fields:
+ effective_read_type = read_fields + extra_fields
+
+ reader = self._create_split_read(split, read_type=effective_read_type).create_reader()
+
+ if not isinstance(reader, RecordBatchReader):
+ from pypaimon.read.reader.auth_masking_reader import RecordReaderToBatchAdapter
+ schema = PyarrowFieldParser.from_paimon_schema(effective_read_type)
+ reader = RecordReaderToBatchAdapter(reader, schema, include_row_kind=self.include_row_kind)
+
+ filter_fn = auth_result.extract_row_filter()
+ if filter_fn:
+ reader = AuthFilterReader(reader, filter_fn)
+
+ if auth_result.column_masking:
+ reader = AuthMaskingReader(reader, auth_result.column_masking, effective_read_type)
Review Comment:
`effective_read_type` includes hidden columns added only to evaluate row filters. Passing it as `read_fields` makes `AuthMaskingReader` treat those hidden columns as projected masking targets. For example, a user projection `id` plus an auth filter on `dept` adds `dept` to `effective_read_type`; if the server also returns a masking rule for `dept` that references `secret`, this constructor validates and fails because `secret` is absent even though `dept` will be dropped by `ColumnProjectReader`. Java filters masking rules against the final output row type, so this path should use the original projected `read_fields` to choose masking targets and validate references against the actual batch or effective type separately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3460502245
##########
paimon-python/pypaimon/catalog/table_query_auth.py:
##########
@@ -0,0 +1,77 @@
+################################################################################
+# 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.
+################################################################################
+
+from typing import Callable, Dict, List, Optional
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+from pypaimon.common.predicate_json_parser import (
+ extract_referenced_fields,
+ parse_predicate_to_batch_filter,
+)
+from pypaimon.schema.data_types import DataField
+
+
+class TableQueryAuthResult:
+
+ def __init__(self, filter: Optional[List[str]], column_masking: Optional[Dict[str, str]]):
+ self.filter = filter
+ self.column_masking = column_masking
+
+ def convert_plan(self, plan):
+ from pypaimon.read.query_auth_split import QueryAuthSplit
+ from pypaimon.read.plan import Plan
+
+ if not self.filter and not self.column_masking:
+ return plan
+ auth_splits = [QueryAuthSplit(split, self) for split in plan.splits()]
+ return Plan(auth_splits, snapshot_id=plan.snapshot_id)
+
+ 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]
Review Comment:
Java `TableQueryAuthResult.extractPredicate()` skips empty filter strings before deserializing them, but this path tries to parse every entry. If the REST server returns `filter=[""]` (or a mix of empty and valid filters), Python will fail the read with `JSONDecodeError` in both `extract_row_filter()` and `get_extra_fields_for_filter()`, while the JVM client treats the empty entry as no-op. Please filter out empty/blank JSON strings before wrapping the plan or before parsing, and add a regression test for empty filter entries.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "MgjLLL (via GitHub)" <gi...@apache.org>.
MgjLLL commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3458015698
##########
paimon-python/pypaimon/common/predicate_json_parser.py:
##########
@@ -0,0 +1,344 @@
+################################################################################
+# 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 re
+from typing import Callable
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+
+def parse_predicate_to_batch_filter(json_str: str) -> Callable[[pa.RecordBatch], pa.Array]:
+ data = json.loads(json_str)
+ return _build_filter(data)
+
+
+def _build_filter(data: dict) -> Callable[[pa.RecordBatch], pa.Array]:
+ kind = data["kind"]
+ if kind == "LEAF":
+ return _build_leaf_filter(data)
+ elif kind == "COMPOUND":
+ return _build_compound_filter(data)
+ raise ValueError(f"Unknown predicate kind: {kind}")
+
+
+def _build_leaf_filter(data: dict) -> Callable:
+ transform = data["transform"]
+ function = data["function"]
+ literals = data.get("literals", [])
+
+ def filter_fn(batch: pa.RecordBatch) -> pa.Array:
+ value_array = _apply_predicate_transform(transform, batch)
+ return _apply_leaf_function(function, value_array, literals, len(batch))
+
+ return filter_fn
+
+
+def _build_compound_filter(data: dict) -> Callable:
+ function = data["function"]
+ child_filters = [_build_filter(child) for child in data["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}")
+
+ return filter_fn
+
+
+def _apply_predicate_transform(transform: dict, batch: pa.RecordBatch) -> pa.Array:
+ name = transform["name"]
+
+ if name == "FIELD_REF":
+ return batch.column(transform["fieldRef"]["name"])
+
+ elif name == "CAST":
+ col = batch.column(transform["fieldRef"]["name"])
+ target_type = _paimon_type_to_arrow(transform["type"])
+ return pc.cast(col, target_type)
+
+ elif name == "UPPER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_upper(input_col)
+
+ elif name == "LOWER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_lower(input_col)
+
+ 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 pc.binary_join_element_wise(*resolved, "")
+
+ elif name == "CONCAT_WS":
+ sep = _resolve_transform_input(transform["inputs"][0], batch)
+ values = [_resolve_transform_input(inp, batch) for inp in transform["inputs"][1:]]
+ if not values:
+ return pa.nulls(len(batch), type=pa.string())
+ return pc.binary_join_element_wise(*values, sep, null_handling='skip')
+
+ elif name == "NULL":
+ return pa.nulls(len(batch), type=pa.bool_())
+
+ raise ValueError(f"Unknown transform type in predicate: {name}")
+
+
+def _resolve_transform_input(inp, batch: pa.RecordBatch) -> pa.Array:
+ if isinstance(inp, dict):
+ return batch.column(inp["name"])
+ elif isinstance(inp, str):
+ return pa.array([inp] * len(batch), type=pa.string())
+ elif inp is None:
+ return pa.nulls(len(batch), type=pa.string())
+ return pa.array([str(inp)] * len(batch), type=pa.string())
+
+
+def _apply_leaf_function(function: str, value_array: pa.Array, literals: list, batch_len: int) -> pa.Array:
+ """Null literal yields False to match Java LeafBinaryFunction/LeafTernaryFunction semantics."""
+ converted = [_convert_literal(lit, value_array.type) for lit in literals]
+
+ if function == "EQUAL":
+ return pc.equal(value_array, converted[0])
+ elif function == "NOT_EQUAL":
+ return pc.not_equal(value_array, converted[0])
+ elif function == "LESS_THAN":
+ return pc.less(value_array, converted[0])
+ elif function == "LESS_OR_EQUAL":
+ return pc.less_equal(value_array, converted[0])
+ elif function == "GREATER_THAN":
+ return pc.greater(value_array, converted[0])
+ elif function == "GREATER_OR_EQUAL":
+ return pc.greater_equal(value_array, converted[0])
+ elif function == "IS_NULL":
+ return pc.is_null(value_array)
+ elif function == "IS_NOT_NULL":
+ return pc.is_valid(value_array)
+ elif function == "IN":
+ return pc.is_in(value_array, pa.array(converted, type=value_array.type))
Review Comment:
Fixed. Now drop null literals from the value set and wrap with pc.if_else(pc.is_valid(value_array), in_mask, False). If all literals are null, returns all-false. Also added null literal guards for EQUAL/NOT_EQUAL/LT/LE/GT/GE to match Java's LeafBinaryFunction semantics.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3458071671
##########
paimon-python/pypaimon/common/predicate_json_parser.py:
##########
@@ -0,0 +1,365 @@
+################################################################################
+# 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 re
+from typing import Callable
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+
+def parse_predicate_to_batch_filter(json_str: str) -> Callable[[pa.RecordBatch], pa.Array]:
+ data = json.loads(json_str)
+ return _build_filter(data)
+
+
+def _build_filter(data: dict) -> Callable[[pa.RecordBatch], pa.Array]:
+ kind = data["kind"]
+ if kind == "LEAF":
+ return _build_leaf_filter(data)
+ elif kind == "COMPOUND":
+ return _build_compound_filter(data)
+ raise ValueError(f"Unknown predicate kind: {kind}")
+
+
+def _build_leaf_filter(data: dict) -> Callable:
+ transform = data["transform"]
+ function = data["function"]
+ literals = data.get("literals", [])
+
+ def filter_fn(batch: pa.RecordBatch) -> pa.Array:
+ value_array = _apply_predicate_transform(transform, batch)
+ return _apply_leaf_function(function, value_array, literals, len(batch))
+
+ return filter_fn
+
+
+def _build_compound_filter(data: dict) -> Callable:
+ function = data["function"]
+ child_filters = [_build_filter(child) for child in data["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}")
+
+ return filter_fn
+
+
+def _apply_predicate_transform(transform: dict, batch: pa.RecordBatch) -> pa.Array:
+ name = transform["name"]
+
+ if name == "FIELD_REF":
+ return batch.column(transform["fieldRef"]["name"])
+
+ elif name == "CAST":
+ col = batch.column(transform["fieldRef"]["name"])
+ target_type = _paimon_type_to_arrow(transform["type"])
+ return pc.cast(col, target_type)
+
+ elif name == "UPPER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_upper(input_col)
+
+ elif name == "LOWER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_lower(input_col)
+
+ 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 pc.binary_join_element_wise(*resolved, "")
+
+ elif name == "CONCAT_WS":
+ sep = _resolve_transform_input(transform["inputs"][0], batch)
+ values = [_resolve_transform_input(inp, batch) for inp in transform["inputs"][1:]]
+ if not values:
+ return pa.nulls(len(batch), type=pa.string())
+ result = pc.binary_join_element_wise(*values, sep, null_handling='skip')
+ if len(result) < len(batch):
+ padded = result.to_pylist() + [None] * (len(batch) - len(result))
Review Comment:
Padding the compacted Arrow result at the end misaligns rows. drops rows whose value inputs are all null, so for , , it returns ; this code pads that to , while Java should evaluate the rows as . In a row-filter predicate that can allow or deny the wrong rows. Please make the CONCAT_WS transform row-preserving, and add a mixed all-null/non-null regression test.
##########
paimon-python/pypaimon/read/reader/auth_masking_reader.py:
##########
@@ -0,0 +1,229 @@
+################################################################################
+# 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
+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,
+ _paimon_type_to_arrow,
+)
+from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
+
+
+class RecordReaderToBatchAdapter(RecordBatchReader):
+
+ def __init__(self, inner, schema: pa.Schema, chunk_size: int = 65536, include_row_kind: bool = False):
+ self._inner = inner
+ self._schema = schema
+ self._chunk_size = chunk_size
+ self._exhausted = False
+ self._pending_iterator = None
+ self._include_row_kind = include_row_kind
+
+ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
+ if self._exhausted:
+ return None
+ row_tuples = []
+ row_kinds = []
+ while len(row_tuples) < self._chunk_size:
+ if self._pending_iterator is not None:
+ row = self._pending_iterator.next()
+ while row is not None:
+ row_tuples.append(
+ row.row_tuple[row.offset:row.offset + row.arity])
+ if self._include_row_kind:
+ row_kinds.append(row.get_row_kind().to_string())
+ if len(row_tuples) >= self._chunk_size:
+ return self._flush(row_tuples, row_kinds)
+ row = self._pending_iterator.next()
+ self._pending_iterator = None
+
+ row_iterator = self._inner.read_batch()
+ if row_iterator is None:
+ self._exhausted = True
+ break
+ self._pending_iterator = row_iterator
+
+ if not row_tuples:
+ return None
+ return self._flush(row_tuples, row_kinds)
+
+ def _flush(self, row_tuples, row_kinds=None):
+ columns_data = list(zip(*row_tuples))
+ pydict = {
+ name: list(col)
+ for name, col in zip(self._schema.names, columns_data)
+ }
+ batch = pa.RecordBatch.from_pydict(pydict, schema=self._schema)
+ if row_kinds:
+ row_kind_array = pa.array(row_kinds, type=pa.string())
+ row_kind_field = pa.field("_row_kind", pa.string())
+ new_schema = pa.schema([row_kind_field] + list(batch.schema))
+ columns = [row_kind_array] + [batch.column(i) for i in range(batch.num_columns)]
+ batch = pa.RecordBatch.from_arrays(columns, schema=new_schema)
+ return batch
+
+ def close(self):
+ self._inner.close()
+
+
+class AuthFilterReader(RecordBatchReader):
+
+ def __init__(self, inner_reader: RecordBatchReader, filter_fn: Callable[[pa.RecordBatch], pa.Array]):
+ self._inner = inner_reader
+ self._filter_fn = filter_fn
+
+ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
+ batch = self._inner.read_arrow_batch()
+ if batch is None:
+ return None
+ mask = self._filter_fn(batch)
+ return batch.filter(mask)
+
+ def close(self):
+ self._inner.close()
+
+
+class AuthMaskingReader(RecordBatchReader):
+
+ def __init__(self, inner_reader: RecordBatchReader, masking_rules: Dict[str, str], read_fields: List):
+ self._inner = inner_reader
+ self._masking_rules = masking_rules
+ self._read_fields = read_fields
+ read_field_names = {f.name for f in read_fields}
+ # Filter to projected columns, then validate field references.
+ self._parsed_rules = {
+ col: json.loads(tj) for col, tj in masking_rules.items()
+ if col in read_field_names
+ }
+ for col_name, transform in self._parsed_rules.items():
+ for ref_name in _collect_all_field_refs_from_transform(transform):
+ if ref_name not in read_field_names:
+ raise RuntimeError(
+ f"Column masking refers to field '{ref_name}' which is not present "
+ f"in output row type. Available fields: {read_field_names}"
+ )
+
+ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
+ batch = self._inner.read_arrow_batch()
+ if batch is None:
+ return None
+ original_batch = batch
+ masked_columns = {}
+ for col_name, transform in self._parsed_rules.items():
+ if col_name in original_batch.schema.names:
+ col_idx = original_batch.schema.get_field_index(col_name)
+ target_col_type = original_batch.schema.field(col_idx).type
+ masked_columns[col_idx] = self._apply_transform(transform, original_batch, target_col_type)
+ for col_idx, masked_array in masked_columns.items():
+ original_field = original_batch.schema.field(col_idx)
+ if masked_array.type != original_field.type:
+ masked_array = pc.cast(masked_array, original_field.type)
+ batch = batch.set_column(
+ col_idx, pa.field(original_field.name, original_field.type, nullable=True), masked_array)
+ return batch
+
+ def close(self):
+ self._inner.close()
+
+ def _apply_transform(
+ self,
+ transform: dict,
+ original_batch: pa.RecordBatch,
+ target_col_type: pa.DataType,
+ ) -> pa.Array:
+ name = transform["name"]
+
+ if name == "NULL":
+ return pa.nulls(len(original_batch), type=target_col_type)
+
+ elif name == "FIELD_REF":
+ ref_name = transform["fieldRef"]["name"]
+ return original_batch.column(ref_name)
+
+ elif name == "CAST":
+ ref_name = transform["fieldRef"]["name"]
+ source_col = original_batch.column(ref_name)
+ target_type = _paimon_type_to_arrow(transform["type"])
+ return pc.cast(source_col, target_type)
+
+ elif name == "UPPER":
+ input_col = self._resolve_input(transform["inputs"][0], original_batch)
+ return pc.utf8_upper(input_col)
+
+ elif name == "LOWER":
+ input_col = self._resolve_input(transform["inputs"][0], original_batch)
+ return pc.utf8_lower(input_col)
+
+ elif name == "CONCAT":
+ return self._apply_concat(transform["inputs"], original_batch)
+
+ elif name == "CONCAT_WS":
+ return self._apply_concat_ws(transform["inputs"], original_batch)
+
+ raise ValueError(f"Unknown transform type: {name}")
+
+ def _resolve_input(self, inp, original_batch: pa.RecordBatch) -> pa.Array:
+ if isinstance(inp, dict):
+ return original_batch.column(inp["name"])
+ elif isinstance(inp, str):
+ return pa.array([inp] * len(original_batch), type=pa.string())
+ elif inp is None:
+ return pa.nulls(len(original_batch), type=pa.string())
+ return pa.array([str(inp)] * len(original_batch), type=pa.string())
+
+ def _apply_concat(self, inputs: list, original_batch: pa.RecordBatch) -> pa.Array:
+ resolved = [self._resolve_input(inp, original_batch) for inp in inputs]
+ if not resolved:
+ return pa.nulls(len(original_batch), type=pa.string())
+ return pc.binary_join_element_wise(*resolved, "")
+
+ def _apply_concat_ws(self, inputs: list, original_batch: pa.RecordBatch) -> pa.Array:
+ if len(inputs) < 2:
+ return pa.nulls(len(original_batch), type=pa.string())
+ sep = self._resolve_input(inputs[0], original_batch)
+ values = [self._resolve_input(inp, original_batch) for inp in inputs[1:]]
+ result = pc.binary_join_element_wise(*values, sep, null_handling='skip')
+ if len(result) < len(original_batch):
+ padded = result.to_pylist() + [None] * (len(original_batch) - len(result))
Review Comment:
This has the same row-alignment problem as the predicate CONCAT_WS path. Arrow returns a compacted array when some rows have all-null value inputs, so appending padding at the end shifts the later rows' masked values. For example + with separator becomes here, but Java would produce . That masks the wrong rows; please preserve original row positions and cover the mixed all-null/non-null case in tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "plusplusjiajia (via GitHub)" <gi...@apache.org>.
plusplusjiajia commented on PR #8136:
URL: https://github.com/apache/paimon/pull/8136#issuecomment-4891581770
Thanks for working on this — pypaimon needs query auth, and the plan()-fetch + read-time enforcement shape matches the Java client. Verified findings, in severity order:
1. Write paths read through auth → silent data corruption. table_update.py:248, table_delete.py:136, table_upsert_by_key.py:507, table_update_by_row_id.py:110 and file_store_write.py:300 all read existing data via new_read_builder().new_scan().plan(), and plan() now applies the row filter/masking unconditionally (there is
no _skip_auth or equivalent on this branch). With a row filter active, upsert treats hidden rows as absent (INSERT instead of UPDATE) and masked keys never match. Java avoids this structurally: auth lives in AbstractDataTableScan/Read (public read layer) while writes read via FileStoreRead below it.
2. LIMIT pushdown drops authorized rows. scanner/file_scanner.py:566 trims splits by raw merged_row_count before the auth filter is applied — _has_non_partition_filter() only checks the user predicate. E.g. with_limit(100) + auth filter region='cn', first 100-row split all us → 0 rows returned. This is exactly what
authHasNonPartitionFilter guards in Java's DataTableBatchScan (#8447);
3. predicate_json_parser crashes on common policy types (fail-closed, feature-blocking):
- LTZ: Java emits TIMESTAMP(p) WITH LOCAL TIME ZONE (LocalZonedTimestampType.FORMAT), but _paimon_type_to_arrow:259 can't match parens mid-string and the branch keys at :291 are spellings Paimon never emits; the unit test asserts the never-emitted format, masking this.
- Temporal literals: on the wire, DATE/TIME/TIMESTAMP literals are serialized from JSR-310 objects (LeafPredicate.serializeLiterals → convertToJavaObject returns LocalDate etc., and JsonSerdeUtil registers JavaTimeModule), but _convert_literal:233 only handles ISO strings. Suggest fixture tests using the exact JSON
strings from Java's PredicateJsonSerdeTest to lock wire parity.
- pc.cast at :83 defaults to safe=True and aborts the whole read on any truncating row (2.7 → INT), where Java's cast rules truncate/wrap and keep evaluating.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "plusplusjiajia (via GitHub)" <gi...@apache.org>.
plusplusjiajia commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3553031959
##########
paimon-python/pypaimon/read/table_scan.py:
##########
@@ -42,17 +42,94 @@ def __init__(
self.predicate = predicate
self.limit = limit
self.partition_predicate = partition_predicate
+ self._read_type = None
+ self._query_auth_fn = self.table.catalog_environment.table_query_auth(
+ self.table.options, self.table.identifier)
self.file_scanner = self._create_file_scanner()
def plan(self) -> Plan:
- return self.file_scanner.scan()
+ auth_result = self._auth_query()
+ if auth_result is not None:
+ self._apply_auth_to_scanner(auth_result)
+ plan = self.file_scanner.scan()
+ if auth_result is not None:
+ plan = auth_result.convert_plan(plan)
+ return plan
+
+ def _apply_auth_to_scanner(self, auth_result):
+ if not auth_result.filter:
+ return
+ partition_preds, has_non_partition = self._split_auth_filter(auth_result)
+ if partition_preds:
+ from pypaimon.common.predicate_builder import PredicateBuilder
+ combined = PredicateBuilder.and_predicates(partition_preds)
+ self.file_scanner.auth_partition_predicate = combined
+ if has_non_partition:
+ self.file_scanner.auth_has_non_partition_filter = True
+
+ def _split_auth_filter(self, auth_result):
+ partition_keys = set(self.table.partition_keys or [])
+ if not partition_keys:
+ return [], bool(auth_result.filter)
+
+ partition_preds = []
+ has_non_partition = False
+ field_index_map = {f.name: i for i, f in enumerate(self.table.fields)}
Review Comment:
When query auth returns a partition-column filter for a table whose partition column is not in the same position as the partition row, this uses full table field positions but later tests the predicate against entry.partition, which only contains partition fields. For common layouts like id, name, dtpart PARTITIONED BY (dtpart), the generated index is out of range or points to the wrong partition key, so scan planning can fail or prune visible splits before the auth reader runs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #8136:
URL: https://github.com/apache/paimon/pull/8136#issuecomment-4815851882
AI cannot solve these problems, please provide feedback in the form of a real person.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3418042647
##########
paimon-python/pypaimon/common/predicate_json_parser.py:
##########
@@ -0,0 +1,322 @@
+################################################################################
+# 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 re
+from typing import Callable
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+
+def parse_predicate_to_batch_filter(json_str: str) -> Callable[[pa.RecordBatch], pa.Array]:
+ data = json.loads(json_str)
+ return _build_filter(data)
+
+
+def _build_filter(data: dict) -> Callable[[pa.RecordBatch], pa.Array]:
+ kind = data["kind"]
+ if kind == "LEAF":
+ return _build_leaf_filter(data)
+ elif kind == "COMPOUND":
+ return _build_compound_filter(data)
+ raise ValueError(f"Unknown predicate kind: {kind}")
+
+
+def _build_leaf_filter(data: dict) -> Callable:
+ transform = data["transform"]
+ function = data["function"]
+ literals = data.get("literals", [])
+
+ def filter_fn(batch: pa.RecordBatch) -> pa.Array:
+ value_array = _apply_predicate_transform(transform, batch)
+ return _apply_leaf_function(function, value_array, literals, len(batch))
+
+ return filter_fn
+
+
+def _build_compound_filter(data: dict) -> Callable:
+ function = data["function"]
+ child_filters = [_build_filter(child) for child in data["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}")
+
+ return filter_fn
+
+
+def _apply_predicate_transform(transform: dict, batch: pa.RecordBatch) -> pa.Array:
+ name = transform["name"]
+
+ if name == "FIELD_REF":
+ return batch.column(transform["fieldRef"]["name"])
+
+ elif name == "CAST":
+ col = batch.column(transform["fieldRef"]["name"])
+ target_type = _paimon_type_to_arrow(transform["type"])
+ return pc.cast(col, target_type)
+
+ elif name == "UPPER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_upper(input_col)
+
+ elif name == "LOWER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_lower(input_col)
+
+ 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 pc.binary_join_element_wise(*resolved, "")
+
+ elif name == "CONCAT_WS":
+ sep = _resolve_transform_input(transform["inputs"][0], batch)
+ values = [_resolve_transform_input(inp, batch) for inp in transform["inputs"][1:]]
+ if not values:
+ return pa.nulls(len(batch), type=pa.string())
+ return pc.binary_join_element_wise(*values, sep, null_handling='skip')
+
+ elif name == "NULL":
+ return pa.nulls(len(batch), type=pa.bool_())
+
+ raise ValueError(f"Unknown transform type in predicate: {name}")
+
+
+def _resolve_transform_input(inp, batch: pa.RecordBatch) -> pa.Array:
+ if isinstance(inp, dict):
+ return batch.column(inp["name"])
+ elif isinstance(inp, str):
+ return pa.array([inp] * len(batch), type=pa.string())
+ elif inp is None:
+ return pa.nulls(len(batch), type=pa.string())
+ return pa.array([str(inp)] * len(batch), type=pa.string())
+
+
+def _apply_leaf_function(function: str, value_array: pa.Array, literals: list, batch_len: int) -> pa.Array:
+ converted = [_convert_literal(lit, value_array.type) for lit in literals]
+
+ if function == "EQUAL":
+ return pc.equal(value_array, converted[0])
+ elif function == "NOT_EQUAL":
+ return pc.not_equal(value_array, converted[0])
+ elif function == "LESS_THAN":
+ return pc.less(value_array, converted[0])
+ elif function == "LESS_OR_EQUAL":
+ return pc.less_equal(value_array, converted[0])
+ elif function == "GREATER_THAN":
+ return pc.greater(value_array, converted[0])
+ elif function == "GREATER_OR_EQUAL":
+ return pc.greater_equal(value_array, converted[0])
+ elif function == "IS_NULL":
+ return pc.is_null(value_array)
+ elif function == "IS_NOT_NULL":
+ return pc.is_valid(value_array)
+ elif function == "IN":
+ return pc.is_in(value_array, pa.array(converted, type=value_array.type))
+ elif function == "NOT_IN":
+ return pc.invert(pc.is_in(value_array, pa.array(converted, type=value_array.type)))
Review Comment:
Java Paimon's `NotIn` returns false when the input field is null, but `pc.is_in(null, ...)` is false and inverting it makes null rows pass. For a row-filter auth rule such as `dept NOT_IN ('blocked')`, Python would expose rows where `dept` is null. Please combine this with `pc.is_valid(value_array)` and also preserve Java's null-literal behavior, where any null literal makes `NOT_IN` false.
##########
paimon-python/pypaimon/common/predicate_json_parser.py:
##########
@@ -0,0 +1,322 @@
+################################################################################
+# 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 re
+from typing import Callable
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+
+def parse_predicate_to_batch_filter(json_str: str) -> Callable[[pa.RecordBatch], pa.Array]:
+ data = json.loads(json_str)
+ return _build_filter(data)
+
+
+def _build_filter(data: dict) -> Callable[[pa.RecordBatch], pa.Array]:
+ kind = data["kind"]
+ if kind == "LEAF":
+ return _build_leaf_filter(data)
+ elif kind == "COMPOUND":
+ return _build_compound_filter(data)
+ raise ValueError(f"Unknown predicate kind: {kind}")
+
+
+def _build_leaf_filter(data: dict) -> Callable:
+ transform = data["transform"]
+ function = data["function"]
+ literals = data.get("literals", [])
+
+ def filter_fn(batch: pa.RecordBatch) -> pa.Array:
+ value_array = _apply_predicate_transform(transform, batch)
+ return _apply_leaf_function(function, value_array, literals, len(batch))
+
+ return filter_fn
+
+
+def _build_compound_filter(data: dict) -> Callable:
+ function = data["function"]
+ child_filters = [_build_filter(child) for child in data["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}")
+
+ return filter_fn
+
+
+def _apply_predicate_transform(transform: dict, batch: pa.RecordBatch) -> pa.Array:
+ name = transform["name"]
+
+ if name == "FIELD_REF":
+ return batch.column(transform["fieldRef"]["name"])
+
+ elif name == "CAST":
+ col = batch.column(transform["fieldRef"]["name"])
+ target_type = _paimon_type_to_arrow(transform["type"])
+ return pc.cast(col, target_type)
+
+ elif name == "UPPER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_upper(input_col)
+
+ elif name == "LOWER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_lower(input_col)
+
+ 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 pc.binary_join_element_wise(*resolved, "")
+
+ elif name == "CONCAT_WS":
+ sep = _resolve_transform_input(transform["inputs"][0], batch)
+ values = [_resolve_transform_input(inp, batch) for inp in transform["inputs"][1:]]
+ if not values:
+ return pa.nulls(len(batch), type=pa.string())
+ return pc.binary_join_element_wise(*values, sep, null_handling='skip')
+
+ elif name == "NULL":
+ return pa.nulls(len(batch), type=pa.bool_())
+
+ raise ValueError(f"Unknown transform type in predicate: {name}")
+
+
+def _resolve_transform_input(inp, batch: pa.RecordBatch) -> pa.Array:
+ if isinstance(inp, dict):
+ return batch.column(inp["name"])
+ elif isinstance(inp, str):
+ return pa.array([inp] * len(batch), type=pa.string())
+ elif inp is None:
+ return pa.nulls(len(batch), type=pa.string())
+ return pa.array([str(inp)] * len(batch), type=pa.string())
+
+
+def _apply_leaf_function(function: str, value_array: pa.Array, literals: list, batch_len: int) -> pa.Array:
+ converted = [_convert_literal(lit, value_array.type) for lit in literals]
+
+ if function == "EQUAL":
+ return pc.equal(value_array, converted[0])
+ elif function == "NOT_EQUAL":
+ return pc.not_equal(value_array, converted[0])
+ elif function == "LESS_THAN":
+ return pc.less(value_array, converted[0])
+ elif function == "LESS_OR_EQUAL":
+ return pc.less_equal(value_array, converted[0])
+ elif function == "GREATER_THAN":
+ return pc.greater(value_array, converted[0])
+ elif function == "GREATER_OR_EQUAL":
+ return pc.greater_equal(value_array, converted[0])
+ elif function == "IS_NULL":
+ return pc.is_null(value_array)
+ elif function == "IS_NOT_NULL":
+ return pc.is_valid(value_array)
+ elif function == "IN":
+ return pc.is_in(value_array, pa.array(converted, type=value_array.type))
+ elif function == "NOT_IN":
+ return pc.invert(pc.is_in(value_array, pa.array(converted, type=value_array.type)))
+ elif function == "BETWEEN":
+ return pc.and_(pc.greater_equal(value_array, converted[0]),
+ pc.less_equal(value_array, converted[1]))
+ elif function == "NOT_BETWEEN":
+ return pc.or_(pc.less(value_array, converted[0]),
+ pc.greater(value_array, converted[1]))
+ elif function == "STARTS_WITH":
+ return pc.starts_with(value_array, converted[0])
+ elif function == "ENDS_WITH":
+ return pc.ends_with(value_array, converted[0])
+ elif function == "CONTAINS":
+ return pc.match_substring(value_array, converted[0])
+ elif function == "LIKE":
+ raw = literals[0]
+ pattern = _sql_to_regex_like(raw)
+ return pc.match_substring_regex(value_array, f"^{pattern}$")
+ elif function == "TRUE":
+ return pa.array([True] * batch_len, type=pa.bool_())
+ elif function == "FALSE":
+ return pa.array([False] * batch_len, type=pa.bool_())
+ raise ValueError(f"Unknown leaf function: {function}")
Review Comment:
`IS_NAN` is a valid Java Paimon predicate (`PredicateBuilder.isNaN` / `IsNaN.NAME`) and can be serialized in REST auth filters. With the current switch it fails every Python read with `Unknown leaf function: IS_NAN`; please add a branch using `pc.is_nan` for float/double arrays.
##########
paimon-python/pypaimon/read/reader/auth_masking_reader.py:
##########
@@ -0,0 +1,220 @@
+################################################################################
+# 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
+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,
+ _paimon_type_to_arrow,
+)
+from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
+
+
+class RecordReaderToBatchAdapter(RecordBatchReader):
+
+ def __init__(self, inner, schema: pa.Schema, chunk_size: int = 65536, include_row_kind: bool = False):
+ self._inner = inner
+ self._schema = schema
+ self._chunk_size = chunk_size
+ self._exhausted = False
+ self._pending_iterator = None
+ self._include_row_kind = include_row_kind
+
+ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
+ if self._exhausted:
+ return None
+ row_tuples = []
+ row_kinds = []
+ while len(row_tuples) < self._chunk_size:
+ if self._pending_iterator is not None:
+ row = self._pending_iterator.next()
+ while row is not None:
+ row_tuples.append(
+ row.row_tuple[row.offset:row.offset + row.arity])
+ if self._include_row_kind:
+ row_kinds.append(row.get_row_kind().to_string())
+ if len(row_tuples) >= self._chunk_size:
+ return self._flush(row_tuples, row_kinds)
+ row = self._pending_iterator.next()
+ self._pending_iterator = None
+
+ row_iterator = self._inner.read_batch()
+ if row_iterator is None:
+ self._exhausted = True
+ break
+ self._pending_iterator = row_iterator
+
+ if not row_tuples:
+ return None
+ return self._flush(row_tuples, row_kinds)
+
+ def _flush(self, row_tuples, row_kinds=None):
+ columns_data = list(zip(*row_tuples))
+ pydict = {
+ name: list(col)
+ for name, col in zip(self._schema.names, columns_data)
+ }
+ batch = pa.RecordBatch.from_pydict(pydict, schema=self._schema)
+ if row_kinds:
+ row_kind_array = pa.array(row_kinds, type=pa.string())
+ row_kind_field = pa.field("_row_kind", pa.string())
+ new_schema = pa.schema([row_kind_field] + list(batch.schema))
+ columns = [row_kind_array] + [batch.column(i) for i in range(batch.num_columns)]
+ batch = pa.RecordBatch.from_arrays(columns, schema=new_schema)
+ return batch
+
+ def close(self):
+ self._inner.close()
+
+
+class AuthFilterReader(RecordBatchReader):
+
+ def __init__(self, inner_reader: RecordBatchReader, filter_fn: Callable[[pa.RecordBatch], pa.Array]):
+ self._inner = inner_reader
+ self._filter_fn = filter_fn
+
+ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
+ batch = self._inner.read_arrow_batch()
+ if batch is None:
+ return None
+ mask = self._filter_fn(batch)
+ return batch.filter(mask)
+
+ def close(self):
+ self._inner.close()
+
+
+class AuthMaskingReader(RecordBatchReader):
+
+ def __init__(self, inner_reader: RecordBatchReader, masking_rules: Dict[str, str], read_fields: List):
+ self._inner = inner_reader
+ self._masking_rules = masking_rules
+ self._read_fields = read_fields
+ self._parsed_rules = {col: json.loads(tj) for col, tj in masking_rules.items()}
+ read_field_names = {f.name for f in read_fields}
+ for col_name, transform in self._parsed_rules.items():
+ for ref_name in _collect_all_field_refs_from_transform(transform):
Review Comment:
This validates references for every masking rule before checking whether the masked target column is actually projected. If REST returns a rule like `secret = FIELD_REF(email)` and the user reads only `id`, the Python reader raises because `email` is absent even though `secret` is not returned. Java skips masking rules whose target column is absent from the output row type before remapping inputs, so this should filter to projected target columns first.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "MgjLLL (via GitHub)" <gi...@apache.org>.
MgjLLL commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3458043036
##########
paimon-python/pypaimon/common/predicate_json_parser.py:
##########
@@ -0,0 +1,344 @@
+################################################################################
+# 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 re
+from typing import Callable
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+
+def parse_predicate_to_batch_filter(json_str: str) -> Callable[[pa.RecordBatch], pa.Array]:
+ data = json.loads(json_str)
+ return _build_filter(data)
+
+
+def _build_filter(data: dict) -> Callable[[pa.RecordBatch], pa.Array]:
+ kind = data["kind"]
+ if kind == "LEAF":
+ return _build_leaf_filter(data)
+ elif kind == "COMPOUND":
+ return _build_compound_filter(data)
+ raise ValueError(f"Unknown predicate kind: {kind}")
+
+
+def _build_leaf_filter(data: dict) -> Callable:
+ transform = data["transform"]
+ function = data["function"]
+ literals = data.get("literals", [])
+
+ def filter_fn(batch: pa.RecordBatch) -> pa.Array:
+ value_array = _apply_predicate_transform(transform, batch)
+ return _apply_leaf_function(function, value_array, literals, len(batch))
+
+ return filter_fn
+
+
+def _build_compound_filter(data: dict) -> Callable:
+ function = data["function"]
+ child_filters = [_build_filter(child) for child in data["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}")
+
+ return filter_fn
+
+
+def _apply_predicate_transform(transform: dict, batch: pa.RecordBatch) -> pa.Array:
+ name = transform["name"]
+
+ if name == "FIELD_REF":
+ return batch.column(transform["fieldRef"]["name"])
+
+ elif name == "CAST":
+ col = batch.column(transform["fieldRef"]["name"])
+ target_type = _paimon_type_to_arrow(transform["type"])
+ return pc.cast(col, target_type)
+
+ elif name == "UPPER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_upper(input_col)
+
+ elif name == "LOWER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_lower(input_col)
+
+ 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 pc.binary_join_element_wise(*resolved, "")
+
+ elif name == "CONCAT_WS":
+ sep = _resolve_transform_input(transform["inputs"][0], batch)
+ values = [_resolve_transform_input(inp, batch) for inp in transform["inputs"][1:]]
+ if not values:
+ return pa.nulls(len(batch), type=pa.string())
+ return pc.binary_join_element_wise(*values, sep, null_handling='skip')
+
+ elif name == "NULL":
+ return pa.nulls(len(batch), type=pa.bool_())
+
+ raise ValueError(f"Unknown transform type in predicate: {name}")
+
+
+def _resolve_transform_input(inp, batch: pa.RecordBatch) -> pa.Array:
+ if isinstance(inp, dict):
+ return batch.column(inp["name"])
+ elif isinstance(inp, str):
+ return pa.array([inp] * len(batch), type=pa.string())
+ elif inp is None:
+ return pa.nulls(len(batch), type=pa.string())
+ return pa.array([str(inp)] * len(batch), type=pa.string())
+
+
+def _apply_leaf_function(function: str, value_array: pa.Array, literals: list, batch_len: int) -> pa.Array:
+ """Null literal yields False to match Java LeafBinaryFunction/LeafTernaryFunction semantics."""
+ converted = [_convert_literal(lit, value_array.type) for lit in literals]
+
+ if function == "EQUAL":
+ return pc.equal(value_array, converted[0])
+ elif function == "NOT_EQUAL":
+ return pc.not_equal(value_array, converted[0])
+ elif function == "LESS_THAN":
+ return pc.less(value_array, converted[0])
+ elif function == "LESS_OR_EQUAL":
+ return pc.less_equal(value_array, converted[0])
+ elif function == "GREATER_THAN":
+ return pc.greater(value_array, converted[0])
+ elif function == "GREATER_OR_EQUAL":
+ return pc.greater_equal(value_array, converted[0])
+ elif function == "IS_NULL":
+ return pc.is_null(value_array)
+ elif function == "IS_NOT_NULL":
+ return pc.is_valid(value_array)
+ elif function == "IN":
+ return pc.is_in(value_array, pa.array(converted, type=value_array.type))
Review Comment:
Fixed. Now drop null literals from the value set and wrap with `pc.if_else(pc.is_valid(value_array), in_mask, False)`. If all literals are null, returns all-false. Also added null literal guards for EQUAL/NOT_EQUAL/LT/LE/GT/GE to match Java's `LeafBinaryFunction` semantics.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "plusplusjiajia (via GitHub)" <gi...@apache.org>.
plusplusjiajia commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3552656158
##########
paimon-python/pypaimon/write/table_update.py:
##########
@@ -245,7 +245,9 @@ def _update_by_predicate(
else:
read_builder.with_projection([SpecialFields.ROW_ID.name])
- splits = read_builder.new_scan().plan().splits()
+ scan = read_builder.new_scan()
+ scan._enable_auth = False
Review Comment:
@JingsongLi Sounds good — just assert both the row filter and column masking are empty.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #8136:
URL: https://github.com/apache/paimon/pull/8136
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "plusplusjiajia (via GitHub)" <gi...@apache.org>.
plusplusjiajia commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3552726103
##########
paimon-python/pypaimon/read/table_scan.py:
##########
@@ -42,17 +42,94 @@ def __init__(
self.predicate = predicate
self.limit = limit
self.partition_predicate = partition_predicate
+ self._read_type = None
+ self._query_auth_fn = self.table.catalog_environment.table_query_auth(
+ self.table.options, self.table.identifier)
self.file_scanner = self._create_file_scanner()
def plan(self) -> Plan:
- return self.file_scanner.scan()
+ auth_result = self._auth_query()
+ if auth_result is not None:
+ self._apply_auth_to_scanner(auth_result)
+ plan = self.file_scanner.scan()
+ if auth_result is not None:
+ plan = auth_result.convert_plan(plan)
+ return plan
+
+ def _apply_auth_to_scanner(self, auth_result):
+ if not auth_result.filter:
+ return
+ partition_preds, has_non_partition = self._split_auth_filter(auth_result)
+ if partition_preds:
+ from pypaimon.common.predicate_builder import PredicateBuilder
+ combined = PredicateBuilder.and_predicates(partition_preds)
+ self.file_scanner.auth_partition_predicate = combined
+ if has_non_partition:
+ self.file_scanner.auth_has_non_partition_filter = True
+
+ def _split_auth_filter(self, auth_result):
+ partition_keys = set(self.table.partition_keys or [])
+ if not partition_keys:
+ return [], bool(auth_result.filter)
+
+ partition_preds = []
+ has_non_partition = False
+ field_index_map = {f.name: i for i, f in enumerate(self.table.fields)}
+
+ for json_str in (auth_result.filter or []):
+ pred = self._try_parse_partition_predicate(json_str, partition_keys, field_index_map)
+ if pred is not None:
+ partition_preds.append(pred)
+ else:
+ has_non_partition = True
+ return partition_preds, has_non_partition
+
+ def _try_parse_partition_predicate(self, json_str, partition_keys, field_index_map):
+ import json as _json
+ data = _json.loads(json_str)
+ if data is None or data.get("kind") != "LEAF":
+ return None
+ transform = data.get("transform", {})
+ if transform.get("name") != "FIELD_REF":
+ return None
+ field_name = transform.get("fieldRef", {}).get("name")
+ if field_name is None or field_name not in partition_keys:
+ return None
+ field_index = field_index_map.get(field_name)
+ if field_index is None:
+ return None
+
+ from pypaimon.common.predicate import Predicate
+ function = data.get("function", "")
+ literals = data.get("literals", [])
+ method_map = {
+ "EQUAL": "equal", "NOT_EQUAL": "not_equal",
+ "LESS_THAN": "less_than", "LESS_OR_EQUAL": "less_or_equal",
+ "GREATER_THAN": "greater_than", "GREATER_OR_EQUAL": "greater_or_equal",
+ "IS_NULL": "is_null", "IS_NOT_NULL": "is_not_null",
+ "IN": "in", "NOT_IN": "not_in",
+ }
+ method = method_map.get(function)
+ if method is None:
+ return None
+ return Predicate(method=method, index=field_index, field=field_name, literals=literals)
+
+ def _auth_query(self):
+ fn = self._query_auth_fn
+ if fn is None:
+ return None
+ select = [f.name for f in self._read_type] if self._read_type else None
+ return fn(select)
def scan_with_stats(self) -> Tuple[Plan, ScanStats]:
"""Run :meth:`plan` while recording manifest / pruning counters.
Only used by :meth:`ReadBuilder.explain`; the regular read path
keeps going through :meth:`plan`.
"""
+ auth_result = self._auth_query()
+ if auth_result is not None:
+ self._apply_auth_to_scanner(auth_result)
return self.file_scanner.scan_with_stats()
Review Comment:
scan_with_stats() still returns file_scanner.scan_with_stats() without convert_plan — looks like it's still raw splits, not QueryAuthSplit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #8136:
URL: https://github.com/apache/paimon/pull/8136#issuecomment-4975545104
Write operations determine authorization restrictions solely based on whether a `QueryAuthSplit` exists in the plan; if the auth filter filters out all data or an empty table results in zero splits, restricted write operations will bypass the rejection logic and may update data based on the erroneous assumption that “no hidden rows exist.”
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "plusplusjiajia (via GitHub)" <gi...@apache.org>.
plusplusjiajia commented on PR #8136:
URL: https://github.com/apache/paimon/pull/8136#issuecomment-4911872410
@MgjLLL @JingsongLi flagging you both since this ties back to your read_builder / public-API note — following up the _skip_auth thread: as the branch stands now (flag removed), the write-corruption @MgjLLL described on Jun 24 is back. Write-path read-modify-write reads existing data through auth (table_update L248-249
(https://github.com/apache/paimon/blob/e774825510a7610bff2a1d88488ff2c7bafd3afb/paimon-python/pypaimon/write/table_update.py#L248-L249)), so column masking is written back over real values (L260
(https://github.com/apache/paimon/blob/e774825510a7610bff2a1d88488ff2c7bafd3afb/paimon-python/pypaimon/write/table_update.py#L260)), independent of write permission.
The real tension seems to be _skip_auth vs. "don't touch the public read_builder." One way to satisfy both: inject the auth provider into new_read_builder() at construction — public signature unchanged, no flag on the API — and route write paths through an auth-free internal builder. Writes then sit below the auth layer
by construction, same as the Java client.
# table_scan.py
def _auth_query(self):
# Skip query-auth for internal write-path reads (read-modify-write); otherwise
# row filtering / column masking would corrupt the data written back.
if not self._enable_auth:
return None
fn = self.table.catalog_environment.table_query_auth(
self.table.options, self.table.identifier)
...
_enable_auth is threaded from the ReadBuilder: new_read_builder() defaults to True, write paths use an internal builder that passes False.
Is there a better way to handle this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "plusplusjiajia (via GitHub)" <gi...@apache.org>.
plusplusjiajia commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3553016282
##########
paimon-python/pypaimon/read/table_scan.py:
##########
@@ -42,17 +42,94 @@ def __init__(
self.predicate = predicate
self.limit = limit
self.partition_predicate = partition_predicate
+ self._read_type = None
+ self._query_auth_fn = self.table.catalog_environment.table_query_auth(
+ self.table.options, self.table.identifier)
self.file_scanner = self._create_file_scanner()
def plan(self) -> Plan:
- return self.file_scanner.scan()
+ auth_result = self._auth_query()
+ if auth_result is not None:
+ self._apply_auth_to_scanner(auth_result)
+ plan = self.file_scanner.scan()
+ if auth_result is not None:
+ plan = auth_result.convert_plan(plan)
+ return plan
+
+ def _apply_auth_to_scanner(self, auth_result):
+ if not auth_result.filter:
+ return
+ partition_preds, has_non_partition = self._split_auth_filter(auth_result)
+ if partition_preds:
+ from pypaimon.common.predicate_builder import PredicateBuilder
+ combined = PredicateBuilder.and_predicates(partition_preds)
+ self.file_scanner.auth_partition_predicate = combined
+ if has_non_partition:
+ self.file_scanner.auth_has_non_partition_filter = True
+
+ def _split_auth_filter(self, auth_result):
+ partition_keys = set(self.table.partition_keys or [])
+ if not partition_keys:
+ return [], bool(auth_result.filter)
+
+ partition_preds = []
+ has_non_partition = False
+ field_index_map = {f.name: i for i, f in enumerate(self.table.fields)}
+
+ for json_str in (auth_result.filter or []):
+ pred = self._try_parse_partition_predicate(json_str, partition_keys, field_index_map)
+ if pred is not None:
+ partition_preds.append(pred)
+ else:
+ has_non_partition = True
+ return partition_preds, has_non_partition
+
+ def _try_parse_partition_predicate(self, json_str, partition_keys, field_index_map):
+ import json as _json
+ data = _json.loads(json_str)
+ if data is None or data.get("kind") != "LEAF":
+ return None
+ transform = data.get("transform", {})
+ if transform.get("name") != "FIELD_REF":
+ return None
+ field_name = transform.get("fieldRef", {}).get("name")
+ if field_name is None or field_name not in partition_keys:
+ return None
+ field_index = field_index_map.get(field_name)
+ if field_index is None:
+ return None
+
+ from pypaimon.common.predicate import Predicate
+ function = data.get("function", "")
+ literals = data.get("literals", [])
+ method_map = {
+ "EQUAL": "equal", "NOT_EQUAL": "not_equal",
+ "LESS_THAN": "less_than", "LESS_OR_EQUAL": "less_or_equal",
+ "GREATER_THAN": "greater_than", "GREATER_OR_EQUAL": "greater_or_equal",
+ "IS_NULL": "is_null", "IS_NOT_NULL": "is_not_null",
+ "IN": "in", "NOT_IN": "not_in",
+ }
Review Comment:
For partition auth filters using anything other than EQUAL/IN, these mappings create method names such as less_than, is_null, and not_in, but Predicate.test() is registered with camelCase names like lessThan, isNull, and notIn. Once such a rule is installed as auth_partition_predicate, scans fail with an unsupported predicate method instead of falling back to the row-level auth filter.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "MgjLLL (via GitHub)" <gi...@apache.org>.
MgjLLL commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3458057844
##########
paimon-python/pypaimon/catalog/table_query_auth.py:
##########
@@ -0,0 +1,87 @@
+################################################################################
+# 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.
+################################################################################
+
+from typing import Callable, Dict, List, Optional
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+from pypaimon.common.predicate_json_parser import (
+ extract_referenced_fields,
+ parse_predicate_to_batch_filter,
+)
+from pypaimon.schema.data_types import DataField
+
+
+class TableNoPermissionException(Exception):
Review Comment:
Fixed. Removed the duplicate class from `table_query_auth.py`. Updated `catalog_exception.TableNoPermissionException` to accept an optional `cause` parameter and handle both `Identifier` objects and plain strings. All callers now use the unified exception.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3458071679
##########
paimon-python/pypaimon/read/reader/auth_masking_reader.py:
##########
@@ -0,0 +1,229 @@
+################################################################################
+# 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
+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,
+ _paimon_type_to_arrow,
+)
+from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
+
+
+class RecordReaderToBatchAdapter(RecordBatchReader):
+
+ def __init__(self, inner, schema: pa.Schema, chunk_size: int = 65536, include_row_kind: bool = False):
+ self._inner = inner
+ self._schema = schema
+ self._chunk_size = chunk_size
+ self._exhausted = False
+ self._pending_iterator = None
+ self._include_row_kind = include_row_kind
+
+ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
+ if self._exhausted:
+ return None
+ row_tuples = []
+ row_kinds = []
+ while len(row_tuples) < self._chunk_size:
+ if self._pending_iterator is not None:
+ row = self._pending_iterator.next()
+ while row is not None:
+ row_tuples.append(
+ row.row_tuple[row.offset:row.offset + row.arity])
+ if self._include_row_kind:
+ row_kinds.append(row.get_row_kind().to_string())
+ if len(row_tuples) >= self._chunk_size:
+ return self._flush(row_tuples, row_kinds)
+ row = self._pending_iterator.next()
+ self._pending_iterator = None
+
+ row_iterator = self._inner.read_batch()
+ if row_iterator is None:
+ self._exhausted = True
+ break
+ self._pending_iterator = row_iterator
+
+ if not row_tuples:
+ return None
+ return self._flush(row_tuples, row_kinds)
+
+ def _flush(self, row_tuples, row_kinds=None):
+ columns_data = list(zip(*row_tuples))
+ pydict = {
+ name: list(col)
+ for name, col in zip(self._schema.names, columns_data)
+ }
+ batch = pa.RecordBatch.from_pydict(pydict, schema=self._schema)
+ if row_kinds:
+ row_kind_array = pa.array(row_kinds, type=pa.string())
+ row_kind_field = pa.field("_row_kind", pa.string())
+ new_schema = pa.schema([row_kind_field] + list(batch.schema))
+ columns = [row_kind_array] + [batch.column(i) for i in range(batch.num_columns)]
+ batch = pa.RecordBatch.from_arrays(columns, schema=new_schema)
+ return batch
+
+ def close(self):
+ self._inner.close()
+
+
+class AuthFilterReader(RecordBatchReader):
+
+ def __init__(self, inner_reader: RecordBatchReader, filter_fn: Callable[[pa.RecordBatch], pa.Array]):
+ self._inner = inner_reader
+ self._filter_fn = filter_fn
+
+ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
+ batch = self._inner.read_arrow_batch()
+ if batch is None:
+ return None
+ mask = self._filter_fn(batch)
+ return batch.filter(mask)
+
+ def close(self):
+ self._inner.close()
+
+
+class AuthMaskingReader(RecordBatchReader):
+
+ def __init__(self, inner_reader: RecordBatchReader, masking_rules: Dict[str, str], read_fields: List):
+ self._inner = inner_reader
+ self._masking_rules = masking_rules
+ self._read_fields = read_fields
+ read_field_names = {f.name for f in read_fields}
+ # Filter to projected columns, then validate field references.
+ self._parsed_rules = {
+ col: json.loads(tj) for col, tj in masking_rules.items()
+ if col in read_field_names
+ }
+ for col_name, transform in self._parsed_rules.items():
+ for ref_name in _collect_all_field_refs_from_transform(transform):
+ if ref_name not in read_field_names:
+ raise RuntimeError(
+ f"Column masking refers to field '{ref_name}' which is not present "
+ f"in output row type. Available fields: {read_field_names}"
+ )
+
+ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
+ batch = self._inner.read_arrow_batch()
+ if batch is None:
+ return None
+ original_batch = batch
+ masked_columns = {}
+ for col_name, transform in self._parsed_rules.items():
+ if col_name in original_batch.schema.names:
+ col_idx = original_batch.schema.get_field_index(col_name)
+ target_col_type = original_batch.schema.field(col_idx).type
+ masked_columns[col_idx] = self._apply_transform(transform, original_batch, target_col_type)
+ for col_idx, masked_array in masked_columns.items():
+ original_field = original_batch.schema.field(col_idx)
+ if masked_array.type != original_field.type:
+ masked_array = pc.cast(masked_array, original_field.type)
+ batch = batch.set_column(
+ col_idx, pa.field(original_field.name, original_field.type, nullable=True), masked_array)
+ return batch
+
+ def close(self):
+ self._inner.close()
+
+ def _apply_transform(
+ self,
+ transform: dict,
+ original_batch: pa.RecordBatch,
+ target_col_type: pa.DataType,
+ ) -> pa.Array:
+ name = transform["name"]
+
+ if name == "NULL":
+ return pa.nulls(len(original_batch), type=target_col_type)
+
+ elif name == "FIELD_REF":
+ ref_name = transform["fieldRef"]["name"]
+ return original_batch.column(ref_name)
+
+ elif name == "CAST":
+ ref_name = transform["fieldRef"]["name"]
+ source_col = original_batch.column(ref_name)
+ target_type = _paimon_type_to_arrow(transform["type"])
+ return pc.cast(source_col, target_type)
+
+ elif name == "UPPER":
+ input_col = self._resolve_input(transform["inputs"][0], original_batch)
+ return pc.utf8_upper(input_col)
+
+ elif name == "LOWER":
+ input_col = self._resolve_input(transform["inputs"][0], original_batch)
+ return pc.utf8_lower(input_col)
+
+ elif name == "CONCAT":
+ return self._apply_concat(transform["inputs"], original_batch)
+
+ elif name == "CONCAT_WS":
+ return self._apply_concat_ws(transform["inputs"], original_batch)
+
+ raise ValueError(f"Unknown transform type: {name}")
+
+ def _resolve_input(self, inp, original_batch: pa.RecordBatch) -> pa.Array:
+ if isinstance(inp, dict):
+ return original_batch.column(inp["name"])
+ elif isinstance(inp, str):
+ return pa.array([inp] * len(original_batch), type=pa.string())
+ elif inp is None:
+ return pa.nulls(len(original_batch), type=pa.string())
+ return pa.array([str(inp)] * len(original_batch), type=pa.string())
+
+ def _apply_concat(self, inputs: list, original_batch: pa.RecordBatch) -> pa.Array:
+ resolved = [self._resolve_input(inp, original_batch) for inp in inputs]
+ if not resolved:
+ return pa.nulls(len(original_batch), type=pa.string())
+ return pc.binary_join_element_wise(*resolved, "")
+
+ def _apply_concat_ws(self, inputs: list, original_batch: pa.RecordBatch) -> pa.Array:
+ if len(inputs) < 2:
+ return pa.nulls(len(original_batch), type=pa.string())
+ sep = self._resolve_input(inputs[0], original_batch)
+ values = [self._resolve_input(inp, original_batch) for inp in inputs[1:]]
+ result = pc.binary_join_element_wise(*values, sep, null_handling='skip')
+ if len(result) < len(original_batch):
+ padded = result.to_pylist() + [None] * (len(original_batch) - len(result))
Review Comment:
This has the same row-alignment problem as the predicate CONCAT_WS path. Arrow returns a compacted array when some rows have all-null value inputs, so appending padding at the end shifts the later rows masked values. For example `a=[None, "x", None]`, `b=[None, "y", "z"]` with `sep="-"` becomes `["x-y", "z", ""]` here, but Java `BinaryString.concatWs` would produce `["", "x-y", "z"]`. That masks the wrong rows; please preserve original row positions and cover the mixed all-null/non-null case in tests.
##########
paimon-python/pypaimon/common/predicate_json_parser.py:
##########
@@ -0,0 +1,365 @@
+################################################################################
+# 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 re
+from typing import Callable
+
+import pyarrow as pa
+import pyarrow.compute as pc
+
+
+def parse_predicate_to_batch_filter(json_str: str) -> Callable[[pa.RecordBatch], pa.Array]:
+ data = json.loads(json_str)
+ return _build_filter(data)
+
+
+def _build_filter(data: dict) -> Callable[[pa.RecordBatch], pa.Array]:
+ kind = data["kind"]
+ if kind == "LEAF":
+ return _build_leaf_filter(data)
+ elif kind == "COMPOUND":
+ return _build_compound_filter(data)
+ raise ValueError(f"Unknown predicate kind: {kind}")
+
+
+def _build_leaf_filter(data: dict) -> Callable:
+ transform = data["transform"]
+ function = data["function"]
+ literals = data.get("literals", [])
+
+ def filter_fn(batch: pa.RecordBatch) -> pa.Array:
+ value_array = _apply_predicate_transform(transform, batch)
+ return _apply_leaf_function(function, value_array, literals, len(batch))
+
+ return filter_fn
+
+
+def _build_compound_filter(data: dict) -> Callable:
+ function = data["function"]
+ child_filters = [_build_filter(child) for child in data["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}")
+
+ return filter_fn
+
+
+def _apply_predicate_transform(transform: dict, batch: pa.RecordBatch) -> pa.Array:
+ name = transform["name"]
+
+ if name == "FIELD_REF":
+ return batch.column(transform["fieldRef"]["name"])
+
+ elif name == "CAST":
+ col = batch.column(transform["fieldRef"]["name"])
+ target_type = _paimon_type_to_arrow(transform["type"])
+ return pc.cast(col, target_type)
+
+ elif name == "UPPER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_upper(input_col)
+
+ elif name == "LOWER":
+ input_col = _resolve_transform_input(transform["inputs"][0], batch)
+ return pc.utf8_lower(input_col)
+
+ 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 pc.binary_join_element_wise(*resolved, "")
+
+ elif name == "CONCAT_WS":
+ sep = _resolve_transform_input(transform["inputs"][0], batch)
+ values = [_resolve_transform_input(inp, batch) for inp in transform["inputs"][1:]]
+ if not values:
+ return pa.nulls(len(batch), type=pa.string())
+ result = pc.binary_join_element_wise(*values, sep, null_handling='skip')
+ if len(result) < len(batch):
+ padded = result.to_pylist() + [None] * (len(batch) - len(result))
Review Comment:
Padding the compacted Arrow result at the end misaligns rows. `pc.binary_join_element_wise(..., null_handling="skip")` drops rows whose value inputs are all null, so for `a=[None, "x", None]`, `b=[None, "y", "z"]`, `sep="-"` it returns `["x-y", "z"]`; this code pads that to `["x-y", "z", ""]`, while Java `BinaryString.concatWs` should evaluate the rows as `["", "x-y", "z"]`. In a row-filter predicate that can allow or deny the wrong rows. Please make the CONCAT_WS transform row-preserving, and add a mixed all-null/non-null regression test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3458088218
##########
paimon-python/pypaimon/read/table_read.py:
##########
@@ -687,6 +720,51 @@ def _widen_to_top_level_for_merge(self) -> List[DataField]:
widened.append(field)
return widened
+ def _create_reader_for_split(self, split):
+ from pypaimon.read.query_auth_split import QueryAuthSplit
+
+ auth_result = None
+ if isinstance(split, QueryAuthSplit):
+ auth_result = split.auth_result
+ split = split.split
+
+ if auth_result is not None:
+ return self._authed_reader(split, auth_result)
+ else:
+ return self._create_split_read(split).create_reader()
+
+ def _authed_reader(self, split, auth_result):
+ from pypaimon.read.reader.auth_masking_reader import (
+ AuthFilterReader, AuthMaskingReader, ColumnProjectReader)
+
+ table_fields = self.table.fields
+ read_fields = self.read_type
+
+ extra_fields = auth_result.get_extra_fields_for_filter(read_fields, table_fields)
+ effective_read_type = read_fields
+ if extra_fields:
+ effective_read_type = read_fields + extra_fields
+
+ reader = self._create_split_read(split, read_type=effective_read_type).create_reader()
+
+ if not isinstance(reader, RecordBatchReader):
+ from pypaimon.read.reader.auth_masking_reader import RecordReaderToBatchAdapter
+ schema = PyarrowFieldParser.from_paimon_schema(effective_read_type)
+ reader = RecordReaderToBatchAdapter(reader, schema, include_row_kind=self.include_row_kind)
+
+ filter_fn = auth_result.extract_row_filter()
+ if filter_fn:
+ reader = AuthFilterReader(reader, filter_fn)
+
+ if auth_result.column_masking:
+ reader = AuthMaskingReader(reader, auth_result.column_masking, effective_read_type)
Review Comment:
`effective_read_type` includes hidden columns added only to evaluate row filters. Passing it as `read_fields` makes `AuthMaskingReader` treat those hidden columns as projected masking targets. For example, a user projection `id` plus an auth filter on `dept` adds `dept` to `effective_read_type`; if the server also returns a masking rule for `dept` that references `secret`, this constructor validates and fails because `secret` is absent even though `dept` will be dropped by `ColumnProjectReader`. Java filters masking rules against the final output row type, so this path should use the original projected `read_fields` to choose masking targets and validate references against the actual batch or effective type separately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support query auth (row filter & column masking) for REST catalog [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8136:
URL: https://github.com/apache/paimon/pull/8136#discussion_r3458124425
##########
paimon-python/pypaimon/read/reader/auth_masking_reader.py:
##########
@@ -0,0 +1,229 @@
+################################################################################
+# 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
+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,
+ _paimon_type_to_arrow,
+)
+from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
+
+
+class RecordReaderToBatchAdapter(RecordBatchReader):
+
+ def __init__(self, inner, schema: pa.Schema, chunk_size: int = 65536, include_row_kind: bool = False):
+ self._inner = inner
+ self._schema = schema
+ self._chunk_size = chunk_size
+ self._exhausted = False
+ self._pending_iterator = None
+ self._include_row_kind = include_row_kind
+
+ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
+ if self._exhausted:
+ return None
+ row_tuples = []
+ row_kinds = []
+ while len(row_tuples) < self._chunk_size:
+ if self._pending_iterator is not None:
+ row = self._pending_iterator.next()
+ while row is not None:
+ row_tuples.append(
+ row.row_tuple[row.offset:row.offset + row.arity])
+ if self._include_row_kind:
+ row_kinds.append(row.get_row_kind().to_string())
+ if len(row_tuples) >= self._chunk_size:
+ return self._flush(row_tuples, row_kinds)
+ row = self._pending_iterator.next()
+ self._pending_iterator = None
+
+ row_iterator = self._inner.read_batch()
+ if row_iterator is None:
+ self._exhausted = True
+ break
+ self._pending_iterator = row_iterator
+
+ if not row_tuples:
+ return None
+ return self._flush(row_tuples, row_kinds)
+
+ def _flush(self, row_tuples, row_kinds=None):
+ columns_data = list(zip(*row_tuples))
+ pydict = {
+ name: list(col)
+ for name, col in zip(self._schema.names, columns_data)
+ }
+ batch = pa.RecordBatch.from_pydict(pydict, schema=self._schema)
+ if row_kinds:
+ row_kind_array = pa.array(row_kinds, type=pa.string())
+ row_kind_field = pa.field("_row_kind", pa.string())
+ new_schema = pa.schema([row_kind_field] + list(batch.schema))
+ columns = [row_kind_array] + [batch.column(i) for i in range(batch.num_columns)]
+ batch = pa.RecordBatch.from_arrays(columns, schema=new_schema)
+ return batch
+
+ def close(self):
+ self._inner.close()
+
+
+class AuthFilterReader(RecordBatchReader):
+
+ def __init__(self, inner_reader: RecordBatchReader, filter_fn: Callable[[pa.RecordBatch], pa.Array]):
+ self._inner = inner_reader
+ self._filter_fn = filter_fn
+
+ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
+ batch = self._inner.read_arrow_batch()
+ if batch is None:
+ return None
+ mask = self._filter_fn(batch)
+ return batch.filter(mask)
+
+ def close(self):
+ self._inner.close()
+
+
+class AuthMaskingReader(RecordBatchReader):
+
+ def __init__(self, inner_reader: RecordBatchReader, masking_rules: Dict[str, str], read_fields: List):
+ self._inner = inner_reader
+ self._masking_rules = masking_rules
+ self._read_fields = read_fields
+ read_field_names = {f.name for f in read_fields}
+ # Filter to projected columns, then validate field references.
+ self._parsed_rules = {
+ col: json.loads(tj) for col, tj in masking_rules.items()
+ if col in read_field_names
+ }
+ for col_name, transform in self._parsed_rules.items():
+ for ref_name in _collect_all_field_refs_from_transform(transform):
+ if ref_name not in read_field_names:
+ raise RuntimeError(
+ f"Column masking refers to field '{ref_name}' which is not present "
+ f"in output row type. Available fields: {read_field_names}"
+ )
+
+ def read_arrow_batch(self) -> Optional[pa.RecordBatch]:
+ batch = self._inner.read_arrow_batch()
+ if batch is None:
+ return None
+ original_batch = batch
+ masked_columns = {}
+ for col_name, transform in self._parsed_rules.items():
+ if col_name in original_batch.schema.names:
+ col_idx = original_batch.schema.get_field_index(col_name)
+ target_col_type = original_batch.schema.field(col_idx).type
+ masked_columns[col_idx] = self._apply_transform(transform, original_batch, target_col_type)
+ for col_idx, masked_array in masked_columns.items():
+ original_field = original_batch.schema.field(col_idx)
+ if masked_array.type != original_field.type:
+ masked_array = pc.cast(masked_array, original_field.type)
Review Comment:
Column masking should not force the transformed value back to the original column type. Java `TableQueryAuthResult.transform` writes `transform.transform(row)` directly; `RESTCatalogTest.testColumnMaskingApplyOnRead` masks an `INT` column with `CastTransform(..., STRING)` and then reads `"100"` with `getString(2)`. Here the same `{"name":"CAST","type":"STRING"}` mask on an int column is cast back to `int32`, so the mask is effectively undone or can fail for non-castable masking expressions. Please replace the column using the masked array type instead, and update the schema-preservation test accordingly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org