[rest] support row filter & column masking in rest catalog

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

@@ -2921,8 +2921,14 @@ components:
       properties:
         filter:
           type: array
+          description: Additional row-level filter as Predicate entry strings.
           items:
             type: string
+        columnMasking:
+          type: object
+          description: Column masking rules as a map from column name to Transform entry JSON string.
+          additionalProperties:
+            type: string
     AlterDatabaseRequest:
       type: object
       properties:

paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java

@@ -672,21 +672,19 @@ public void alterTable(Identifier identifier, List<SchemaChange> changes) {
      *
      * @param identifier database name and table name.
      * @param select select columns, null if select all
-     * @return additional filter for row level access control
+     * @return additional row-level filter and column masking
      * @throws NoSuchResourceException Exception thrown on HTTP 404 means the table not exists
      * @throws ForbiddenException Exception thrown on HTTP 403 means don't have the permission for
      *     this table
      */
-    public List<String> authTableQuery(Identifier identifier, @Nullable List<String> select) {
+    public AuthTableQueryResponse authTableQuery(
+            Identifier identifier, @Nullable List<String> select) {
         AuthTableQueryRequest request = new AuthTableQueryRequest(select);
-        AuthTableQueryResponse response =
-                client.post(
-                        resourcePaths.authTable(
-                                identifier.getDatabaseName(), identifier.getObjectName()),
-                        request,
-                        AuthTableQueryResponse.class,
-                        restAuthFunction);
-        return response.filter();
+        return client.post(
+                resourcePaths.authTable(identifier.getDatabaseName(), identifier.getObjectName()),
+                request,
+                AuthTableQueryResponse.class,
+                restAuthFunction);
     }
 
     /**

paimon-api/src/main/java/org/apache/paimon/rest/responses/AuthTableQueryResponse.java

@@ -27,24 +27,38 @@
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 import java.util.List;
+import java.util.Map;
 
 /** Response for auth table query. */
 @JsonIgnoreProperties(ignoreUnknown = true)
 public class AuthTableQueryResponse implements RESTResponse {
 
     private static final String FIELD_FILTER = "filter";
+    private static final String FIELD_COLUMN_MASKING = "columnMasking";
 
     @JsonInclude(JsonInclude.Include.NON_NULL)
     @JsonProperty(FIELD_FILTER)
     private final List<String> filter;
 
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    @JsonProperty(FIELD_COLUMN_MASKING)
+    private final Map<String, String> columnMasking;
+
     @JsonCreator
-    public AuthTableQueryResponse(@JsonProperty(FIELD_FILTER) List<String> filter) {
+    public AuthTableQueryResponse(
+            @JsonProperty(FIELD_FILTER) List<String> filter,
+            @JsonProperty(FIELD_COLUMN_MASKING) Map<String, String> columnMasking) {
         this.filter = filter;
+        this.columnMasking = columnMasking;
     }
 
     @JsonGetter(FIELD_FILTER)
     public List<String> filter() {
         return filter;
     }
+
+    @JsonGetter(FIELD_COLUMN_MASKING)
+    public Map<String, String> columnMasking() {
+        return columnMasking;
+    }
 }

paimon-common/src/main/java/org/apache/paimon/predicate/DefaultValueTransform.java

@@ -0,0 +1,92 @@
+/*
+ * 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.predicate;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.utils.DefaultValueUtils;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * A {@link Transform} which always returns the default value of the input field's {@link DataType}.
+ */
+public class DefaultValueTransform implements Transform {
+
+    private static final long serialVersionUID = 1L;
+
+    private final FieldRef fieldRef;
+
+    public DefaultValueTransform(FieldRef fieldRef) {
+        this.fieldRef = Objects.requireNonNull(fieldRef, "fieldRef must not be null");
+    }
+
+    public FieldRef fieldRef() {
+        return fieldRef;
+    }
+
+    @Override
+    public List<Object> inputs() {
+        return Collections.singletonList(fieldRef);
+    }
+
+    @Override
+    public DataType outputType() {
+        return fieldRef.type();
+    }
+
+    @Override
+    public Object transform(InternalRow row) {
+        return DefaultValueUtils.defaultValue(fieldRef.type());
+    }
+
+    @Override
+    public Transform copyWithNewInputs(List<Object> inputs) {
+        List<Object> nonNullInputs =
+                Objects.requireNonNull(inputs, "DefaultValueTransform expects non-null inputs");
+        checkArgument(nonNullInputs.size() == 1, "DefaultValueTransform expects 1 input");
+        checkArgument(
+                nonNullInputs.get(0) instanceof FieldRef,
+                "DefaultValueTransform input must be FieldRef");
+        return new DefaultValueTransform((FieldRef) nonNullInputs.get(0));
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        DefaultValueTransform that = (DefaultValueTransform) o;
+        return Objects.equals(fieldRef, that.fieldRef);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(fieldRef);
+    }
+
+    @Override
+    public String toString() {
+        return "DefaultValueTransform{" + "fieldRef=" + fieldRef + '}';
+    }
+}

paimon-common/src/main/java/org/apache/paimon/predicate/HashMaskTransform.java

@@ -0,0 +1,168 @@
+/*
+ * 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.predicate;
+
+import org.apache.paimon.data.BinaryString;
+
+import org.apache.paimon.shade.guava30.com.google.common.hash.HashCode;
+
+import javax.annotation.Nullable;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.Objects;
+
+import static org.apache.paimon.types.DataTypeFamily.CHARACTER_STRING;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * A {@link Transform} which masks a string column by hashing it.
+ *
+ * <p>Output is a hex string. Default algorithm is {@code SHA-256}.
+ */
+public class HashMaskTransform extends StringTransform {
+
+    private static final long serialVersionUID = 1L;
+
+    private final String algorithm;
+
+    @Nullable private final BinaryString salt;
+
+    private transient MessageDigest digest;
+
+    public HashMaskTransform(FieldRef fieldRef) {
+        this(fieldRef, null, null);
+    }
+
+    public HashMaskTransform(
+            FieldRef fieldRef, @Nullable String algorithm, @Nullable BinaryString salt) {
+        super(Arrays.asList(Objects.requireNonNull(fieldRef, "fieldRef must not be null")));
+        checkArgument(fieldRef.type().is(CHARACTER_STRING), "fieldRef must be a string type");
+        this.algorithm = resolveAlgorithm(algorithm);
+        this.salt = salt;
+        ensureDigest();
+    }
+
+    public FieldRef fieldRef() {
+        return (FieldRef) inputs().get(0);
+    }
+
+    public String algorithm() {
+        return algorithm;
+    }
+
+    @Nullable
+    public BinaryString salt() {
+        return salt;
+    }
+
+    @Nullable
+    @Override
+    protected BinaryString transform(List<BinaryString> inputs) {
+        BinaryString s = inputs.get(0);
+        if (s == null) {
+            return null;
+        }
+
+        MessageDigest md = ensureDigest();
+        md.reset();
+
+        if (salt != null) {
+            md.update(salt.toString().getBytes(StandardCharsets.UTF_8));
+        }
+        md.update(s.toString().getBytes(StandardCharsets.UTF_8));
+
+        return BinaryString.fromString(HashCode.fromBytes(md.digest()).toString());
+    }
+
+    @Override
+    public Transform copyWithNewInputs(List<Object> inputs) {
+        List<Object> nonNullInputs =
+                Objects.requireNonNull(inputs, "HashMaskTransform expects non-null inputs");
+        checkArgument(nonNullInputs.size() == 1, "HashMaskTransform expects 1 input");
+        checkArgument(
+                nonNullInputs.get(0) instanceof FieldRef,
+                "HashMaskTransform input must be FieldRef");
+        return new HashMaskTransform((FieldRef) nonNullInputs.get(0), algorithm, salt);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        HashMaskTransform that = (HashMaskTransform) o;
+        return Objects.equals(fieldRef(), that.fieldRef())
+                && Objects.equals(algorithm, that.algorithm)
+                && Objects.equals(salt, that.salt);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(fieldRef(), algorithm, salt);
+    }
+
+    @Override
+    public String toString() {
+        return "HashMaskTransform{"
+                + "fieldRef="
+                + fieldRef()
+                + ", algorithm='"
+                + algorithm
+                + '\''
+                + ", salt="
+                + salt
+                + '}';
+    }
+
+    private MessageDigest ensureDigest() {
+        if (digest == null) {
+            try {
+                digest = MessageDigest.getInstance(algorithm);
+            } catch (NoSuchAlgorithmException e) {
+                throw new IllegalArgumentException("Unsupported hash algorithm: " + algorithm, e);
+            }
+        }
+        return digest;
+    }
+
+    private static String resolveAlgorithm(@Nullable String algorithm) {
+        if (algorithm == null || algorithm.trim().isEmpty()) {
+            return "SHA-256";
+        }
+        String a = algorithm.trim();
+        String normalized = a.replace("-", "").toLowerCase(Locale.ROOT);
+        switch (normalized) {
+            case "md5":
+                return "MD5";
+            case "sha1":
+                return "SHA-1";
+            case "sha256":
+                return "SHA-256";
+            case "sha512":
+                return "SHA-512";
+            default:
+                return a;
+        }
+    }
+}

paimon-common/src/main/java/org/apache/paimon/predicate/LeafPredicate.java

@@ -56,10 +56,6 @@ public LeafPredicate(
         super(fieldTransform, function, literals);
     }
 
-    public LeafFunction function() {
-        return function;
-    }
-
     public DataType type() {
         return fieldRef().type();
     }
@@ -80,10 +76,6 @@ public FieldRef fieldRef() {
         return ((FieldTransform) transform).fieldRef();
     }
 
-    public List<Object> literals() {
-        return literals;
-    }
-
     public LeafPredicate copyWithNewIndex(int fieldIndex) {
         return new LeafPredicate(function, type(), fieldIndex, fieldName(), literals);
     }