[rest] Add view penetration endpoint for table access via view

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

@@ -740,6 +740,69 @@ paths:
           $ref: '#/components/responses/TableNotExistErrorResponse'
         "500":
           $ref: '#/components/responses/ServerErrorResponse'
+  /v1/{prefix}/databases/{database}/tables/{table}/via/{viaDatabase}/{viaObject}:
+    post:
+      tags:
+        - table
+      summary: Get table via view (view penetration)
+      description: >
+        Get the table metadata accessed via a view. If the caller has permission
+        on the view, they can access the underlying table referenced by the view.
+        This API can only be called by trusted engines. The server must
+        authenticate whether the caller is a trusted engine.
+      operationId: getTableVia
+      parameters:
+        - name: prefix
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: database
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: table
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: viaDatabase
+          in: path
+          required: true
+          schema:
+            type: string
+          description: Database name of the view through which access is granted
+        - name: viaObject
+          in: path
+          required: true
+          schema:
+            type: string
+          description: Name of the view through which access is granted
+      responses:
+        "200":
+          description: Table metadata accessed via the view
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/GetTableResponse'
+        "401":
+          $ref: '#/components/responses/UnauthorizedErrorResponse'
+        "403":
+          $ref: '#/components/responses/ForbiddenErrorResponse'
+        404:
+          description:
+            Not Found
+            - TableNotExistException, table does not exist
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+              examples:
+                TableNotExist:
+                  $ref: '#/components/examples/TableNotExistError'
+        "500":
+          $ref: '#/components/responses/ServerErrorResponse'
   /v1/{prefix}/databases/{database}/tables/{table}/auth:
     post:
       tags:

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

@@ -519,6 +519,32 @@ public GetTableResponse getTableById(String tableId) {
         return client.get(resourcePaths.table(tableId), GetTableResponse.class, restAuthFunction);
     }
 
+    /**
+     * Get table via a view (view penetration). If the caller has permission on the view identified
+     * by {@code via}, they can access the underlying table referenced by the view.
+     *
+     * <p>This API can only be called by trusted engines. The server must authenticate whether the
+     * caller is a trusted engine.
+     *
+     * @param table database name and table name of the target table.
+     * @param via database name and object name of the view through which access is granted.
+     * @return {@link GetTableResponse}
+     * @throws NoSuchResourceException Exception thrown on HTTP 404 means the table or view not
+     *     exists
+     * @throws ForbiddenException Exception thrown on HTTP 403 means don't have the permission
+     */
+    public GetTableResponse getTableVia(Identifier table, Identifier via) {
+        return client.post(
+                resourcePaths.tableVia(
+                        table.getDatabaseName(),
+                        table.getObjectName(),
+                        via.getDatabaseName(),
+                        via.getObjectName()),
+                new ForwardBranchRequest(),
+                GetTableResponse.class,
+                restAuthFunction);
+    }
+
     /**
      * Load latest snapshot for table.
      *

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

@@ -43,6 +43,7 @@ public class ResourcePaths {
     protected static final String FUNCTIONS = "functions";
     protected static final String FUNCTION_DETAILS = "function-details";
     protected static final String ID = "id";
+    protected static final String VIA = "via";
 
     private static final Joiner SLASH = Joiner.on("/").skipNulls();
 
@@ -94,6 +95,19 @@ public String table(String databaseName, String objectName) {
                 encodeString(objectName));
     }
 
+    public String tableVia(String databaseName, String objectName, String viaDb, String viaObject) {
+        return SLASH.join(
+                V1,
+                prefix,
+                DATABASES,
+                encodeString(databaseName),
+                TABLES,
+                encodeString(objectName),
+                VIA,
+                encodeString(viaDb),
+                encodeString(viaObject));
+    }
+
     public String renameTable() {
         return SLASH.join(V1, prefix, TABLES, "rename");
     }

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

@@ -157,6 +157,22 @@ void alterDatabase(String name, List<PropertyChange> changes, boolean ignoreIfNo
      */
     Table getTable(Identifier identifier) throws TableNotExistException;
 
+    /**
+     * Return a {@link Table} identified by the given {@link Identifier}, accessed via a view (view
+     * penetration). If the caller has permission on the view, they can access the underlying table.
+     *
+     * <p>This API can only be called by trusted engines. The server must authenticate whether the
+     * caller is a trusted engine.
+     *
+     * @param table Path of the target table
+     * @param via Path of the view through which access is granted
+     * @return The requested table
+     * @throws TableNotExistException if the target does not exist
+     */
+    default Table getTableVia(Identifier table, Identifier via) throws TableNotExistException {
+        return getTable(table);
+    }
+
     /**
      * Return a {@link Table} identified by the given tableId.
      *

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

@@ -375,6 +375,11 @@ public Table getTable(Identifier identifier) throws TableNotExistException {
         return wrapped.getTable(identifier);
     }
 
+    @Override
+    public Table getTableVia(Identifier table, Identifier via) throws TableNotExistException {
+        return wrapped.getTableVia(table, via);
+    }
+
     @Override
     public View getView(Identifier identifier) throws ViewNotExistException {
         return wrapped.getView(identifier);

paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedCatalog.java

@@ -158,6 +158,17 @@ public Table getTable(Identifier identifier) throws TableNotExistException {
         }
     }
 
+    @Override
+    public Table getTableVia(Identifier table, Identifier via) throws TableNotExistException {
+        Table result = wrapped.getTableVia(table, via);
+        if (result instanceof FileStoreTable) {
+            return PrivilegedFileStoreTable.wrap(
+                    (FileStoreTable) result, privilegeManager.getPrivilegeChecker(), table);
+        } else {
+            return result;
+        }
+    }
+
     @Override
     public void markDonePartitions(Identifier identifier, List<Map<String, String>> partitions)
             throws TableNotExistException {

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

@@ -319,6 +319,20 @@ public Table getTable(Identifier identifier) throws TableNotExistException {
                 true);
     }
 
+    @Override
+    public Table getTableVia(Identifier table, Identifier via) throws TableNotExistException {
+        return CatalogUtils.loadTable(
+                this,
+                table,
+                path -> fileIOForData(path, table),
+                this::fileIOFromOptions,
+                i -> loadTableMetadataVia(i, via),
+                null,
+                null,
+                context,
+                true);
+    }
+
     @Override
     public Optional<TableSnapshot> loadSnapshot(Identifier identifier)
             throws TableNotExistException {
@@ -480,6 +494,25 @@ private TableMetadata loadTableMetadata(Identifier identifier) throws TableNotEx
         return toTableMetadata(identifier.getDatabaseName(), response);
     }
 
+    /**
+     * Load table metadata via a view identifier (view penetration).
+     *
+     * <p>This API can only be called by trusted engines. The server must authenticate whether the
+     * caller is a trusted engine.
+     */
+    public TableMetadata loadTableMetadataVia(Identifier table, Identifier via)
+            throws TableNotExistException {
+        GetTableResponse response;
+        try {
+            response = api.getTableVia(table, via);
+        } catch (NoSuchResourceException e) {
+            throw new TableNotExistException(table);
+        } catch (ForbiddenException e) {
+            throw new TableNoPermissionException(table, e);
+        }
+        return toTableMetadata(table.getDatabaseName(), response);
+    }
+
     private TableMetadata toTableMetadata(String db, GetTableResponse response) {
         TableSchema schema = TableSchema.create(response.getSchemaId(), response.getSchema());
         Map<String, String> options = new HashMap<>(schema.options());

paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java

@@ -413,6 +413,10 @@ && isTableByIdRequest(request.getPath())) {
                                 resources.length == 5
                                         && ResourcePaths.TABLES.equals(resources[1])
                                         && ResourcePaths.SNAPSHOTS.equals(resources[3]);
+                        boolean isTableVia =
+                                resources.length == 6
+                                        && ResourcePaths.TABLES.equals(resources[1])
+                                        && ResourcePaths.VIA.equals(resources[3]);
                         boolean isTableAuth =
                                 resources.length == 4
                                         && ResourcePaths.TABLES.equals(resources[1])
@@ -529,6 +533,8 @@ && isTableByIdRequest(request.getPath())) {
                             return resetConsumer(identifier, restAuthParameter.data());
                         } else if (isLoadSnapshot) {
                             return loadSnapshot(identifier, resources[4]);
+                        } else if (isTableVia) {
+                            return tableViaHandle(identifier);
                         } else if (isTableAuth) {
                             return authTable(identifier, restAuthParameter.data());
                         } else if (isCommitSnapshot) {
@@ -1701,6 +1707,32 @@ private MockResponse tableByIdHandle(String requestPath) throws Exception {
                 404);
     }
 
+    // This API can only be called by trusted engines. The server must authenticate
+    // whether the caller is a trusted engine.
+    private MockResponse tableViaHandle(Identifier identifier) throws Exception {
+        if (noPermissionTables.contains(identifier.getFullName())) {
+            throw new Catalog.TableNoPermissionException(identifier);
+        }
+        TableMetadata tableMetadata = tableMetadataStore.get(identifier.getFullName());
+        Schema schema = tableMetadata.schema().toSchema();
+        String path = schema.options().remove(PATH.key());
+        RESTResponse response =
+                new GetTableResponse(
+                        tableMetadata.uuid(),
+                        identifier.getDatabaseName(),
+                        identifier.getObjectName(),
+                        path,
+                        tableMetadata.isExternal(),
+                        tableMetadata.schema().id(),
+                        schema,
+                        "owner",
+                        1L,
+                        "created",
+                        1L,
+                        "updated");
+        return mockResponse(response, 200);
+    }
+
     private MockResponse tableHandle(String method, String data, Identifier identifier)
             throws Exception {
         RESTResponse response;

paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java

@@ -406,6 +406,36 @@ void testGetTableById() throws Exception {
                 () -> restCatalog.getTableById("missing_table_id"));
     }
 
+    @Test
+    void testGetTableVia() throws Exception {
+        Identifier tableIdentifier = Identifier.create("test_table_db", "via_table");
+        Identifier viewIdentifier = Identifier.create("test_table_db", "via_view");
+        createTable(tableIdentifier, Maps.newHashMap(), Lists.newArrayList());
+        Table table = restCatalog.getTableVia(tableIdentifier, viewIdentifier);
+        assertThat(table).isNotNull();
+        assertThat(table.name()).isEqualTo("via_table");
+    }
+
+    @Test
+    void testGetTableViaWhenTableNotExist() {
+        Identifier tableIdentifier = Identifier.create("test_table_db", "non_exist_table");
+        Identifier viewIdentifier = Identifier.create("test_table_db", "via_view");
+        assertThrows(
+                Catalog.TableNotExistException.class,
+                () -> restCatalog.getTableVia(tableIdentifier, viewIdentifier));
+    }
+
+    @Test
+    void testGetTableViaWhenTableNoPermission() throws Exception {
+        Identifier tableIdentifier = Identifier.create("test_table_db", "no_perm_via_table");
+        Identifier viewIdentifier = Identifier.create("test_table_db", "via_view");
+        createTable(tableIdentifier, Maps.newHashMap(), Lists.newArrayList());
+        revokeTablePermission(tableIdentifier);
+        assertThrows(
+                Catalog.TableNoPermissionException.class,
+                () -> restCatalog.getTableVia(tableIdentifier, viewIdentifier));
+    }
+
     @Test
     void renameWhenTargetTableExist() throws Exception {
         Identifier identifier = Identifier.create("test_table_db", "rename_table");

paimon-python/pypaimon/api/resource_paths.py

@@ -34,6 +34,7 @@ class ResourcePaths:
     BRANCHES = "branches"
     RENAME = "rename"
     FORWARD = "forward"
+    VIA = "via"
 
     def __init__(self, prefix: str):
         self.base_path = "/{}/{}".format(self.V1, prefix).rstrip("/")
@@ -71,6 +72,12 @@ def table_token(self, database_name: str, table_name: str) -> str:
         return ("{}/{}/{}/{}/{}/token".format(self.base_path, self.DATABASES, RESTUtil.encode_string(database_name),
                 self.TABLES, RESTUtil.encode_string(table_name)))
 
+    def table_via(self, database_name: str, table_name: str, via_db: str, via_object: str) -> str:
+        return "{}/{}/{}/{}/{}/{}/{}/{}".format(
+            self.base_path, self.DATABASES, RESTUtil.encode_string(database_name),
+            self.TABLES, RESTUtil.encode_string(table_name), self.VIA,
+            RESTUtil.encode_string(via_db), RESTUtil.encode_string(via_object))
+
     def rename_table(self) -> str:
         return "{}/{}/rename".format(self.base_path, self.TABLES)
 

paimon-python/pypaimon/api/rest_api.py

@@ -293,6 +293,22 @@ def get_table(self, identifier: Identifier) -> GetTableResponse:
             self.rest_auth_function,
         )
 
+    def get_table_via(self, table: Identifier, via: Identifier) -> GetTableResponse:
+        """Get table via a view (view penetration).
+
+        This API can only be called by trusted engines. The server must authenticate
+        whether the caller is a trusted engine.
+        """
+        table_db, table_name = self.__validate_identifier(table)
+        via_db, via_object = self.__validate_identifier(via)
+
+        return self.client.post_with_response_type(
+            self.resource_paths.table_via(table_db, table_name, via_db, via_object),
+            ForwardBranchRequest(),
+            GetTableResponse,
+            self.rest_auth_function,
+        )
+
     def drop_table(self, identifier: Identifier) -> GetTableResponse:
         database_name, table_name = self.__validate_identifier(identifier)
 

paimon-python/pypaimon/catalog/catalog.py

@@ -89,6 +89,15 @@ def alter_database(self, name: str, changes: list):
     def get_table(self, identifier: Union[str, Identifier]) -> 'Table':
         """Get paimon table identified by the given Identifier."""
 
+    def get_table_via(self, table: Union[str, Identifier], via: Union[str, Identifier]) -> 'Table':
+        """Get table via a view (view penetration).
+
+        If the caller has permission on the view, they can access the underlying table.
+        This API can only be called by trusted engines. The server must authenticate
+        whether the caller is a trusted engine.
+        """
+        return self.get_table(table)
+
     @abstractmethod
     def create_table(self, identifier: Union[str, Identifier], schema: Schema, ignore_if_exists: bool):
         """Create table with schema."""