[WIP] support row filter in rest catalog

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

@@ -2916,8 +2916,14 @@ components:
       properties:
         filter:
           type: array
+          description: Additional row-level filter as JSON strings. Each element should be a JSON object containing a 'filter' field.
           items:
             type: string
+        columnMasking:
+          type: object
+          description: Column masking rules as a map from column name to predicate entry JSON string (TransformPredicate).
+          additionalProperties:
+            type: string
     AlterDatabaseRequest:
       type: object
       properties:

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

@@ -671,21 +671,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 auth result including additional row-level filter and optional 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;
 
     @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;
     }
+
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    @JsonProperty(FIELD_COLUMN_MASKING)
+    private final Map<String, String> columnMasking;
+
+    @JsonGetter(FIELD_COLUMN_MASKING)
+    public Map<String, String> columnMasking() {
+        return columnMasking;
+    }
 }

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

@@ -62,6 +62,14 @@ public Transform transform() {
         return transform;
     }
 
+    public LeafFunction function() {
+        return function;
+    }
+
+    public List<Object> literals() {
+        return literals;
+    }
+
     public TransformPredicate copyWithNewInputs(List<Object> newInputs) {
         return TransformPredicate.of(transform.copyWithNewInputs(newInputs), function, literals);
     }

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

@@ -574,7 +574,8 @@ public boolean supportsVersionManagement() {
     }
 
     @Override
-    public List<String> authTableQuery(Identifier identifier, List<String> select) {
+    public TableQueryAuthResult authTableQuery(
+            Identifier identifier, @Nullable List<String> select) {
         throw new UnsupportedOperationException();
     }
 

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

@@ -1028,14 +1028,15 @@ void alterFunction(
     // ==================== Table Auth ==========================
 
     /**
-     * Auth table query select and get the filter for row level access control.
+     * Auth table query select and get the row-level filter and column masking rules.
      *
      * @param identifier path of the table to alter partitions
      * @param select selected fields, null if select all
-     * @return additional filter for row level access control
+     * @return auth result including additional filter for row level access control and column
+     *     masking
      * @throws TableNotExistException if the table does not exist
      */
-    List<String> authTableQuery(Identifier identifier, @Nullable List<String> select)
+    TableQueryAuthResult authTableQuery(Identifier identifier, @Nullable List<String> select)
             throws TableNotExistException;
 
     // ==================== Catalog Information ==========================

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

@@ -406,7 +406,7 @@ public PagedList<Partition> listPartitionsPaged(
     }
 
     @Override
-    public List<String> authTableQuery(Identifier identifier, @Nullable List<String> select)
+    public TableQueryAuthResult authTableQuery(Identifier identifier, @Nullable List<String> select)
             throws TableNotExistException {
         return wrapped.authTableQuery(identifier, select);
     }

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

@@ -0,0 +1,53 @@
+/*
+ * 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.predicate.TransformPredicate;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.Map;
+
+/** Auth result for table query, including row-level filter and optional column masking rules. */
+public class TableQueryAuthResult {
+
+    @Nullable private final Predicate rowFilter;
+    private final Map<String, TransformPredicate> columnMasking;
+
+    public TableQueryAuthResult(
+            @Nullable Predicate rowFilter, Map<String, TransformPredicate> columnMasking) {
+        this.rowFilter = rowFilter;
+        this.columnMasking = columnMasking == null ? Collections.emptyMap() : columnMasking;
+    }
+
+    public static TableQueryAuthResult empty() {
+        return new TableQueryAuthResult(null, Collections.emptyMap());
+    }
+
+    @Nullable
+    public Predicate rowFilter() {
+        return rowFilter;
+    }
+
+    public Map<String, TransformPredicate> columnMasking() {
+        return columnMasking;
+    }
+}

paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java

@@ -30,13 +30,18 @@
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.catalog.PropertyChange;
 import org.apache.paimon.catalog.TableMetadata;
+import org.apache.paimon.catalog.TableQueryAuthResult;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.function.Function;
 import org.apache.paimon.function.FunctionChange;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.partition.Partition;
 import org.apache.paimon.partition.PartitionStatistics;
+import org.apache.paimon.predicate.And;
+import org.apache.paimon.predicate.CompoundPredicate;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.TransformPredicate;
 import org.apache.paimon.rest.exceptions.AlreadyExistsException;
 import org.apache.paimon.rest.exceptions.BadRequestException;
 import org.apache.paimon.rest.exceptions.ForbiddenException;
@@ -59,12 +64,14 @@
 import org.apache.paimon.table.sink.BatchTableCommit;
 import org.apache.paimon.table.system.SystemTableLoader;
 import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.PredicateJsonSerde;
 import org.apache.paimon.utils.SnapshotNotExistException;
 import org.apache.paimon.view.View;
 import org.apache.paimon.view.ViewChange;
 import org.apache.paimon.view.ViewImpl;
 import org.apache.paimon.view.ViewSchema;
 
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
 import org.apache.paimon.shade.org.apache.commons.lang3.StringUtils;
 
 import javax.annotation.Nullable;
@@ -79,6 +86,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.TreeMap;
 import java.util.stream.Collectors;
 
 import static org.apache.paimon.CoreOptions.BRANCH;
@@ -524,11 +532,61 @@ public void alterTable(
     }
 
     @Override
-    public List<String> authTableQuery(Identifier identifier, @Nullable List<String> select)
+    public TableQueryAuthResult authTableQuery(Identifier identifier, @Nullable List<String> select)
             throws TableNotExistException {
         checkNotSystemTable(identifier, "authTable");
         try {
-            return api.authTableQuery(identifier, select);
+            org.apache.paimon.rest.responses.AuthTableQueryResponse response =
+                    api.authTableQuery(identifier, select);
+
+            List<String> predicateJsons = response == null ? null : response.filter();
+            Predicate rowFilter = null;
+            if (predicateJsons != null && !predicateJsons.isEmpty()) {
+                List<Predicate> predicates = new ArrayList<>();
+                for (String json : predicateJsons) {
+                    if (json == null || json.trim().isEmpty()) {
+                        continue;
+                    }
+                    Predicate predicate = PredicateJsonSerde.parse(json);
+                    if (predicate != null) {
+                        predicates.add(predicate);
+                    }
+                }
+                if (predicates.size() == 1) {
+                    rowFilter = predicates.get(0);
+                } else if (!predicates.isEmpty()) {
+                    rowFilter = new CompoundPredicate(And.INSTANCE, predicates);
+                }
+            }
+
+            Map<String, TransformPredicate> columnMasking = new TreeMap<>();
+            Map<String, String> maskingJsons = response == null ? null : response.columnMasking();
+            if (maskingJsons != null && !maskingJsons.isEmpty()) {
+                for (Map.Entry<String, String> e : maskingJsons.entrySet()) {
+                    String column = e.getKey();
+                    String json = e.getValue();
+                    if (column == null
+                            || column.trim().isEmpty()
+                            || json == null
+                            || json.trim().isEmpty()) {
+                        continue;
+                    }
+                    Predicate predicate = PredicateJsonSerde.parse(json);
+                    if (predicate == null) {
+                        continue;
+                    }
+                    if (!(predicate instanceof TransformPredicate)) {
+                        throw new IllegalArgumentException(
+                                "Column masking must be a TransformPredicate, but got "
+                                        + predicate.getClass().getName()
+                                        + " for column "
+                                        + column);
+                    }
+                    columnMasking.put(column, (TransformPredicate) predicate);
+                }
+            }
+
+            return new TableQueryAuthResult(rowFilter, columnMasking);
         } catch (NoSuchResourceException e) {
             throw new TableNotExistException(identifier);
         } catch (ForbiddenException e) {
@@ -539,6 +597,8 @@ public List<String> authTableQuery(Identifier identifier, @Nullable List<String>
             throw new UnsupportedOperationException(e.getMessage());
         } catch (BadRequestException e) {
             throw new RuntimeException(new IllegalArgumentException(e.getMessage()));
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
         }
     }
 

paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java

@@ -28,6 +28,7 @@
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.catalog.RenamingSnapshotCommit;
 import org.apache.paimon.catalog.SnapshotCommit;
+import org.apache.paimon.catalog.TableQueryAuthResult;
 import org.apache.paimon.operation.Lock;
 import org.apache.paimon.table.source.TableQueryAuth;
 import org.apache.paimon.tag.SnapshotLoaderImpl;
@@ -37,9 +38,10 @@
 import javax.annotation.Nullable;
 
 import java.io.Serializable;
-import java.util.Collections;
 import java.util.Optional;
 
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
 /** Catalog environment in table which contains log factory, metastore client factory. */
 public class CatalogEnvironment implements Serializable {
 
@@ -154,10 +156,11 @@ public CatalogEnvironment copy(Identifier identifier) {
 
     public TableQueryAuth tableQueryAuth(CoreOptions options) {
         if (!options.queryAuthEnabled() || catalogLoader == null) {
-            return select -> Collections.emptyList();
+            return select -> TableQueryAuthResult.empty();
         }
+        final CatalogLoader loader = checkNotNull(catalogLoader);
         return select -> {
-            try (Catalog catalog = catalogLoader.load()) {
+            try (Catalog catalog = loader.load()) {
                 return catalog.authTableQuery(identifier, select);
             } catch (Exception e) {
                 throw new RuntimeException(e);

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

@@ -169,8 +169,12 @@ protected void authQuery() {
         if (!options.queryAuthEnabled()) {
             return;
         }
-        queryAuth.auth(readType == null ? null : readType.getFieldNames());
-        // TODO add support for row level access control
+        org.apache.paimon.catalog.TableQueryAuthResult authResult =
+                queryAuth.auth(readType == null ? null : readType.getFieldNames());
+        Predicate rowFilter = authResult == null ? null : authResult.rowFilter();
+        if (rowFilter != null) {
+            snapshotReader.withFilter(rowFilter);
+        }
     }
 
     @Override