[core][test] Add REST management query authorization support

docs/docs/concepts/rest/management-api.md

@@ -170,10 +170,11 @@ are intersected. If any applicable range rejects a selected column, the query fa
 silently dropping that column.
 
 Schema evolution keeps the assignment attached to the stable table identity. Renaming a referenced
-column updates its stored name. Dropping a referenced column removes it from the range; if that
-would leave the stored list empty, the assignment is removed. An allowlist denies columns added
-later, while a denylist allows them, so allowlists are safer when new columns may contain sensitive
-data.
+column updates its stored name. Dropping a referenced column removes it from the range. The server
+must reject a schema change that would leave an allowlist empty because removing that assignment
+would widen access; an empty denylist is equivalent to no column restriction, so that assignment is
+removed. An allowlist denies columns added later, while a denylist allows them, so allowlists are
+safer when new columns may contain sensitive data.
 
 `expireTime`, when present, is an exclusive upper bound evaluated against the REST server clock.
 At `now >= expireTime`, the assignment must not authorize access. Expired direct assignments may

docs/static/rest-management-open-api.yaml

@@ -638,7 +638,9 @@ components:
  names must exist when granted. An allowlist denies columns added later, while a denylist
  allows columns added later. All applicable column ranges are intersected, and selecting any
  column outside the effective range fails the query. The target table must enforce query
- authorization before the grant becomes visible.
+ authorization before the grant becomes visible. Schema evolution must reject removal of
+ every column in an allowlist because deleting the resulting empty assignment would widen
+ access; an empty denylist assignment may be removed.
  properties:
  columnNames:
  type: array

paimon-core/src/main/java/org/apache/paimon/catalog/TableQueryAuthResult.java

@@ -48,6 +48,8 @@
 import java.util.TreeMap;
 import java.util.stream.Collectors;
 
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
 /** Auth result for table query, including row level filter and optional column masking rules. */
 public class TableQueryAuthResult implements Serializable {
 
@@ -89,13 +91,10 @@ public Predicate extractPredicate() {
  if (filter != null && !filter.isEmpty()) {
  List<Predicate> predicates = new ArrayList<>();
  for (String json : filter) {
- if (StringUtils.isEmpty(json)) {
- continue;
- }
+ checkArgument(!StringUtils.isEmpty(json), "Row filter cannot be empty.");
  Predicate predicate = JsonSerdeUtil.fromJson(json, Predicate.class);
- if (predicate != null) {
- predicates.add(predicate);
- }
+ checkArgument(predicate != null, "Row filter cannot be JSON null.");
+ predicates.add(predicate);
  }
  if (predicates.size() == 1) {
  rowFilter = predicates.get(0);
@@ -122,13 +121,10 @@ public Map<String, Transform> extractColumnMasking() {
  for (Map.Entry<String, String> e : columnMasking.entrySet()) {
  String column = e.getKey();
  String json = e.getValue();
- if (StringUtils.isEmpty(column) || StringUtils.isEmpty(json)) {
- continue;
- }
+ checkArgument(!StringUtils.isEmpty(column), "Column mask target cannot be empty.");
+ checkArgument(!StringUtils.isEmpty(json), "Column mask transform cannot be empty.");
  Transform transform = JsonSerdeUtil.fromJson(json, Transform.class);
- if (transform == null) {
- continue;
- }
+ checkArgument(transform != null, "Column mask transform cannot be JSON null.");
  result.put(column, transform);
  }
  }
@@ -137,18 +133,25 @@ public Map<String, Transform> extractColumnMasking() {
 
  public RecordReader<InternalRow> doAuth(
  RecordReader<InternalRow> reader, RowType outputRowType) {
- Predicate rowFilter = extractPredicate();
+ return doAuth(reader, outputRowType, extractPredicate(), extractColumnMasking());
+ }
+
+ /** Applies already decoded query-authorization definitions to a physical read projection. */
+ public RecordReader<InternalRow> doAuth(
+ RecordReader<InternalRow> reader,
+ RowType outputRowType,
+ @Nullable Predicate rowFilter,
+ Map<String, Transform> selectedColumnMasking) {
  if (rowFilter != null) {
  Predicate remappedFilter = remapPredicate(rowFilter, outputRowType);
  if (remappedFilter != null) {
  reader = reader.filter(remappedFilter::test);
  }
  }
 
- Map<String, Transform> columnMasking = extractColumnMasking();
- if (columnMasking != null && !columnMasking.isEmpty()) {
+ if (!selectedColumnMasking.isEmpty()) {
  Map<Integer, Transform> remappedMasking =
- transformRemapping(outputRowType, columnMasking);
+ transformRemapping(outputRowType, selectedColumnMasking);
  if (!remappedMasking.isEmpty()) {
  reader = reader.transform(row -> transform(outputRowType, remappedMasking, row));
  }
@@ -184,14 +187,15 @@ private static Map<Integer, Transform> transformRemapping(
  for (Map.Entry<String, Transform> e : masking.entrySet()) {
  String targetColumn = e.getKey();
  Transform transform = e.getValue();
- if (targetColumn == null || transform == null) {
- continue;
- }
+ checkArgument(targetColumn != null, "Column mask target cannot be null.");
+ checkArgument(transform != null, "Column mask transform cannot be null.");
 
  int targetIndex = outputRowType.getFieldIndex(targetColumn);
- if (targetIndex < 0) {
- continue;
- }
+ checkArgument(
+ targetIndex >= 0,
+ "Column mask target '%s' is not present in output row type %s.",
+ targetColumn,
+ outputRowType);
 
  List<Object> newInputs = new ArrayList<>();
  for (Object input : transform.inputs()) {
@@ -234,7 +238,7 @@ public Predicate visit(LeafPredicate predicate) {
  String fieldName = ref.name();
  int newIndex = outputRowType.getFieldIndex(fieldName);
  if (newIndex < 0) {
- throw new RuntimeException(
+ throw new IllegalArgumentException(
  String.format(
  "Unable to read data without column %s when row filter enabled.",
  fieldName));
@@ -250,15 +254,20 @@ public Predicate visit(LeafPredicate predicate) {
 
  @Override
  public Predicate visit(CompoundPredicate predicate) {
+ checkArgument(
+ predicate.function() != null, "Compound row filter function cannot be null.");
+ checkArgument(
+ predicate.children() != null, "Compound row filter children cannot be null.");
  List<Predicate> remappedChildren = new ArrayList<>();
  for (Predicate child : predicate.children()) {
+ checkArgument(child != null, "Compound row filter child cannot be null.");
  Predicate remapped = child.visit(this);
  if (remapped != null) {
  remappedChildren.add(remapped);
  }
  }
  if (remappedChildren.isEmpty()) {
-return null;
+throw new IllegalArgumentException("Compound row filter must contain a predicate.");
  }
  if (remappedChildren.size() == 1) {
  return remappedChildren.get(0);

paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableRead.java

@@ -21,20 +21,24 @@
 import org.apache.paimon.catalog.TableQueryAuthResult;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.predicate.FieldRef;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.predicate.PredicateProjectionConverter;
+import org.apache.paimon.predicate.Transform;
 import org.apache.paimon.reader.RecordReader;
 import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.DataField;
 import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.ListUtils;
 import org.apache.paimon.utils.ProjectedRow;
 
 import javax.annotation.Nullable;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 
@@ -119,6 +123,13 @@ protected final QueryAuthContext unwrapQueryAuthSplit(Split split) {
 
  protected final RecordReader<InternalRow> createDataReader(
  Split split, @Nullable TableQueryAuthResult authResult) throws IOException {
+ // A TableRead can be reused for multiple splits. Authentication may have expanded an
+ // explicitly configured physical projection for the previous split, so restore it before
+ // applying the current split's authorization dependencies. Without an explicit projection,
+ // the underlying reader must retain its own default read type.
+ if (readType != null) {
+ applyReadType(readType);
+ }
  RecordReader<InternalRow> reader;
  if (authResult == null) {
  reader = reader(split);
@@ -138,24 +149,42 @@ private RecordReader<InternalRow> authedReader(Split split, TableQueryAuthResult
  RowType tableType = schema.logicalRowType();
  RowType readType = this.readType == null ? tableType : this.readType;
  Predicate authPredicate = authResult.extractPredicate();
+ Map<String, Transform> columnMasking = authResult.extractColumnMasking();
  ProjectedRow backRow = null;
+ List<String> readFields = readType.getFieldNames();
+ Set<String> readFieldSet = new HashSet<>(readFields);
+ Map<String, Transform> selectedColumnMasking = new HashMap<>();
+ for (Map.Entry<String, Transform> mask : columnMasking.entrySet()) {
+ if (readFieldSet.contains(mask.getKey())) {
+ selectedColumnMasking.put(mask.getKey(), mask.getValue());
+ }
+ }
+ Set<String> authFields = new HashSet<>();
  if (authPredicate != null) {
- Set<String> authFields = collectFieldNames(authPredicate);
- List<String> readFields = readType.getFieldNames();
- List<String> authAddNames = new ArrayList<>();
- Set<String> readFieldSet = new HashSet<>(readFields);
- for (String field : tableType.getFieldNames()) {
- if (authFields.contains(field) && !readFieldSet.contains(field)) {
- authAddNames.add(field);
+ authFields.addAll(collectFieldNames(authPredicate));
+ }
+ for (Map.Entry<String, Transform> mask : selectedColumnMasking.entrySet()) {
+ authFields.add(mask.getKey());
+ for (Object input : mask.getValue().inputs()) {
+ if (input instanceof FieldRef) {
+ authFields.add(((FieldRef) input).name());
+ }
+ }
+ }
+ if (!authFields.isEmpty()) {
+ List<DataField> expandedFields = new ArrayList<>(readType.getFields());
+ for (DataField field : tableType.getFields()) {
+ if (authFields.contains(field.name()) && !readFieldSet.contains(field.name())) {
+ expandedFields.add(field);
  }
  }
- if (!authAddNames.isEmpty()) {
- readType = tableType.project(ListUtils.union(readFields, authAddNames));
-withReadType(readType);
+ if (expandedFields.size() > readType.getFieldCount()) {
+ readType = readType.copy(expandedFields);
+applyReadType(readType);
  backRow = ProjectedRow.from(readType.projectIndexes(readFields));
  }
  }
- reader = authResult.doAuth(reader(split), readType);
+ reader = authResult.doAuth(reader(split), readType, authPredicate, selectedColumnMasking);
  if (backRow != null) {
  reader = reader.transform(backRow::replaceRow);
  }

paimon-core/src/test/java/org/apache/paimon/catalog/TableQueryAuthResultTest.java

@@ -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.
+ */
+
+package org.apache.paimon.catalog;
+
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.JsonSerdeUtil;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests that malformed query-authorization definitions cannot be silently ignored. */
+public class TableQueryAuthResultTest {
+
+ @Test
+ void testInvalidRowFilterFailsClosed() {
+ assertThatThrownBy(
+ () ->
+ new TableQueryAuthResult(Collections.singletonList(""), null)
+ .extractPredicate())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("cannot be empty");
+ assertThatThrownBy(
+ () ->
+ new TableQueryAuthResult(Collections.singletonList("null"), null)
+ .extractPredicate())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("JSON null");
+
+ Predicate emptyCompound =
+ JsonSerdeUtil.fromJson(
+ "{\"kind\":\"COMPOUND\",\"function\":\"AND\",\"children\":[]}",
+ Predicate.class);
+ assertThatThrownBy(() -> TableQueryAuthResult.remapPredicate(emptyCompound, RowType.of()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must contain a predicate");
+
+ Predicate missingFunction =
+ JsonSerdeUtil.fromJson(
+ "{\"kind\":\"COMPOUND\",\"function\":null,\"children\":["
+ + "{\"kind\":\"LEAF\",\"transform\":{\"name\":\"NULL\"},"
+ + "\"function\":\"TRUE\",\"literals\":[]},"
+ + "{\"kind\":\"LEAF\",\"transform\":{\"name\":\"NULL\"},"
+ + "\"function\":\"TRUE\",\"literals\":[]}]}",
+ Predicate.class);
+ assertThatThrownBy(() -> TableQueryAuthResult.remapPredicate(missingFunction, RowType.of()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("function cannot be null");
+ }
+
+ @Test
+ void testInvalidColumnMaskFailsClosed() {
+ assertThatThrownBy(
+ () ->
+ new TableQueryAuthResult(
+ null, Collections.singletonMap("email", ""))
+ .extractColumnMasking())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("cannot be empty");
+ assertThatThrownBy(
+ () ->
+ new TableQueryAuthResult(
+ null, Collections.singletonMap("email", "null"))
+ .extractColumnMasking())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("JSON null");
+ }
+}