(Based on PR#2223)Support Namespace/Table level RBAC for external passthrough catalogs

polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java

@@ -271,6 +271,15 @@ public static void enforceFeatureEnabledOrThrow(
           .defaultValue(false)
           .buildFeatureConfiguration();
 
+  public static final FeatureConfiguration<Boolean> ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS =
+      PolarisConfiguration.<Boolean>builder()
+          .key("ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS")
+          .description(
+              "When enabled, allows RBAC operations to create synthetic entities for"
+                  + " entities in federated catalogs that don't exist in the local metastore.")
+          .defaultValue(false)
+          .buildFeatureConfiguration();
+
   public static final FeatureConfiguration<Boolean> ENABLE_POLICY_STORE =
       PolarisConfiguration.<Boolean>builder()
           .key("ENABLE_POLICY_STORE")

runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisAdminService.java

@@ -102,6 +102,7 @@
 import org.apache.polaris.core.exceptions.CommitConflictException;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper;
+import org.apache.polaris.core.persistence.dao.entity.BaseResult;
 import org.apache.polaris.core.persistence.dao.entity.CreateCatalogResult;
 import org.apache.polaris.core.persistence.dao.entity.CreatePrincipalResult;
 import org.apache.polaris.core.persistence.dao.entity.DropEntityResult;
@@ -446,7 +447,7 @@ private void authorizeGrantOnNamespaceOperationOrThrow(
     resolutionManifest =
         resolutionManifestFactory.createResolutionManifest(
             callContext, securityContext, catalogName);
-    resolutionManifest.addPath(
+    resolutionManifest.addPassthroughPath(
         new ResolverPath(Arrays.asList(namespace.levels()), PolarisEntityType.NAMESPACE),
         namespace);
     resolutionManifest.addPath(
@@ -487,7 +488,11 @@ private void authorizeGrantOnTableLikeOperationOrThrow(
     resolutionManifest =
         resolutionManifestFactory.createResolutionManifest(
             callContext, securityContext, catalogName);
-    resolutionManifest.addPath(
+    resolutionManifest.addPassthroughPath(
+        new ResolverPath(
+            Arrays.asList(identifier.namespace().levels()), PolarisEntityType.NAMESPACE),
+        identifier.namespace());
+    resolutionManifest.addPassthroughPath(
         new ResolverPath(
             PolarisCatalogHelpers.tableIdentifierToList(identifier), PolarisEntityType.TABLE_LIKE),
         identifier);
@@ -509,7 +514,12 @@ private void authorizeGrantOnTableLikeOperationOrThrow(
     PolarisResolvedPathWrapper tableLikeWrapper =
         resolutionManifest.getResolvedPath(
             identifier, PolarisEntityType.TABLE_LIKE, PolarisEntitySubType.ANY_SUBTYPE, true);
-    if (!subTypes.contains(tableLikeWrapper.getRawLeafEntity().getSubType())) {
+    boolean rbacForFederatedCatalogsEnabled =
+        getCurrentPolarisContext()
+            .getRealmConfig()
+            .getConfig(FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS);
+    if (!(resolutionManifest.getIsPassthroughFacade() && rbacForFederatedCatalogsEnabled)
+        && !subTypes.contains(tableLikeWrapper.getRawLeafEntity().getSubType())) {
       CatalogHandler.throwNotFoundExceptionForTableLikeEntity(identifier, subTypes);
     }
 
@@ -1687,14 +1697,34 @@ public PrivilegeResult grantPrivilegeOnNamespaceToRole(
         PolarisAuthorizableOperation.ADD_NAMESPACE_GRANT_TO_CATALOG_ROLE;
     authorizeGrantOnNamespaceOperationOrThrow(op, catalogName, namespace, catalogRoleName);
 
+    CatalogEntity catalogEntity =
+        findCatalogByName(catalogName)
+            .orElseThrow(() -> new NotFoundException("Parent catalog %s not found", catalogName));
     PolarisEntity catalogRoleEntity =
         findCatalogRoleByName(catalogName, catalogRoleName)
             .orElseThrow(() -> new NotFoundException("CatalogRole %s not found", catalogRoleName));
 
     PolarisResolvedPathWrapper resolvedPathWrapper = resolutionManifest.getResolvedPath(namespace);
     if (resolvedPathWrapper == null
         || !resolvedPathWrapper.isFullyResolvedNamespace(catalogName, namespace)) {
-      throw new NotFoundException("Namespace %s not found", namespace);
+      boolean rbacForFederatedCatalogsEnabled =
+          getCurrentPolarisContext()
+              .getRealmConfig()
+              .getConfig(FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS);
+      if (resolutionManifest.getIsPassthroughFacade() && rbacForFederatedCatalogsEnabled) {
+        resolvedPathWrapper =
+            createSyntheticNamespaceEntities(catalogEntity, namespace, resolvedPathWrapper);
+        if (resolvedPathWrapper == null
+            || !resolvedPathWrapper.isFullyResolvedNamespace(catalogName, namespace)) {
+          // TODO: update the exception thrown as we refine the possible retry scenarios
+          throw new RuntimeException(
+              String.format(
+                  "Failed to create synthetic namespace entities for namespace %s in catalog %s",
+                  namespace, catalogName));
+        }
+      } else {
+        throw new NotFoundException("Namespace %s not found", namespace);
+      }
     }
     List<PolarisEntity> catalogPath = resolvedPathWrapper.getRawParentPath();
     PolarisEntity namespaceEntity = resolvedPathWrapper.getRawLeafEntity();
@@ -1734,6 +1764,86 @@ public PrivilegeResult revokePrivilegeOnNamespaceFromRole(
         privilege);
   }
 
+  /**
+   * Creates and persists the missing synthetic namespace entities for external catalogs.
+   *
+   * @param catalogEntity the external passthrough facade catalog entity.
+   * @param namespace the expected fully resolved namespace to be created.
+   * @param existingPath the partially resolved path currently stored in the metastore.
+   * @return the fully resolved path wrapper.
+   */
+  private PolarisResolvedPathWrapper createSyntheticNamespaceEntities(
+      CatalogEntity catalogEntity, Namespace namespace, PolarisResolvedPathWrapper existingPath) {
+
+    if (existingPath == null) {
+      throw new IllegalStateException(
+          String.format("Catalog entity %s does not exist.", catalogEntity.getName()));
+    }
+
+    List<PolarisEntity> completePath = new ArrayList<>(existingPath.getRawFullPath());
+    PolarisEntity currentParent = existingPath.getRawLeafEntity();
+
+    String[] allNamespaceLevels = namespace.levels();
+    int numMatchingLevels = 0;
+    // Find parts of the complete path that match the namespace levels.
+    // We skip index 0 because it is the CatalogEntity.
+    for (PolarisEntity entity : completePath.subList(1, completePath.size())) {
+      if (!entity.getName().equals(allNamespaceLevels[numMatchingLevels])) {
+        break;
+      }
+      numMatchingLevels++;
+    }
+
+    for (int i = numMatchingLevels; i < allNamespaceLevels.length; i++) {
+      String[] namespacePart = Arrays.copyOfRange(allNamespaceLevels, 0, i + 1);
+      String leafNamespace = namespacePart[namespacePart.length - 1];
+      Namespace currentNamespace = Namespace.of(namespacePart);
+
+      // TODO: Instead of creating synthetic entitties, rely on external catalog mediated backfill.
+      PolarisEntity syntheticNamespace =
+          new NamespaceEntity.Builder(currentNamespace)
+              .setId(metaStoreManager.generateNewEntityId(getCurrentPolarisContext()).getId())
+              .setCatalogId(catalogEntity.getId())
+              .setParentId(currentParent.getId())
+              .setCreateTimestamp(System.currentTimeMillis())
+              .build();
+
+      EntityResult result =
+          metaStoreManager.createEntityIfNotExists(
+              getCurrentPolarisContext(),
+              PolarisEntity.toCoreList(completePath),
+              syntheticNamespace);
+
+      if (result.isSuccess()) {
+        syntheticNamespace = PolarisEntity.of(result.getEntity());
+      } else if (result.getReturnStatus() == BaseResult.ReturnStatus.ENTITY_ALREADY_EXISTS) {
+        PolarisResolvedPathWrapper partialPath =
+            resolutionManifest.getPassthroughResolvedPath(namespace);
+        PolarisEntity partialLeafEntity =
+            partialPath != null ? partialPath.getRawLeafEntity() : null;
+        if (partialLeafEntity == null
+            || !(partialLeafEntity.getName().equals(leafNamespace)
+                && partialLeafEntity.getType() == PolarisEntityType.NAMESPACE)) {
+          throw new RuntimeException(
+              String.format(
+                  "Failed to create or find namespace entity '%s' in federated catalog '%s'",
+                  leafNamespace, catalogEntity.getName()));
+        }
+        syntheticNamespace = partialLeafEntity;
+      } else {
+        throw new RuntimeException(
+            String.format(
+                "Failed to create or find namespace entity '%s' in federated catalog '%s'",
+                leafNamespace, catalogEntity.getName()));
+      }
+      completePath.add(syntheticNamespace);
+      currentParent = syntheticNamespace;
+    }
+    PolarisResolvedPathWrapper resolvedPathWrapper =
+        resolutionManifest.getPassthroughResolvedPath(namespace);
+    return resolvedPathWrapper;
+  }
+
   public PrivilegeResult grantPrivilegeOnTableToRole(
       String catalogName,
       String catalogRoleName,
@@ -2011,9 +2121,9 @@ private PrivilegeResult grantPrivilegeOnTableLikeToRole(
       TableIdentifier identifier,
       List<PolarisEntitySubType> subTypes,
       PolarisPrivilege privilege) {
-    if (findCatalogByName(catalogName).isEmpty()) {
-      throw new NotFoundException("Parent catalog %s not found", catalogName);
-    }
+    CatalogEntity catalogEntity =
+        findCatalogByName(catalogName)
+            .orElseThrow(() -> new NotFoundException("Parent catalog %s not found", catalogName));
     PolarisEntity catalogRoleEntity =
         findCatalogRoleByName(catalogName, catalogRoleName)
             .orElseThrow(() -> new NotFoundException("CatalogRole %s not found", catalogRoleName));
@@ -2023,7 +2133,25 @@ private PrivilegeResult grantPrivilegeOnTableLikeToRole(
             identifier, PolarisEntityType.TABLE_LIKE, PolarisEntitySubType.ANY_SUBTYPE);
     if (resolvedPathWrapper == null
         || !subTypes.contains(resolvedPathWrapper.getRawLeafEntity().getSubType())) {
-      CatalogHandler.throwNotFoundExceptionForTableLikeEntity(identifier, subTypes);
+      boolean rbacForFederatedCatalogsEnabled =
+          getCurrentPolarisContext()
+              .getRealmConfig()
+              .getConfig(FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS);
+      if (resolutionManifest.getIsPassthroughFacade() && rbacForFederatedCatalogsEnabled) {
+        resolvedPathWrapper =
+            createSyntheticTableLikeEntities(
+                catalogEntity, identifier, subTypes, resolvedPathWrapper);
+        if (resolvedPathWrapper == null
+            || !subTypes.contains(resolvedPathWrapper.getRawLeafEntity().getSubType())) {
+          // TODO: update the exception thrown as we refine the possible retry scenarios
+          throw new RuntimeException(
+              String.format(
+                  "Failed to create synthetic table-like entity for table %s in catalog %s",
+                  identifier.name(), catalogEntity.getName()));
+        }
+      } else {
+        CatalogHandler.throwNotFoundExceptionForTableLikeEntity(identifier, subTypes);
+      }
     }
     List<PolarisEntity> catalogPath = resolvedPathWrapper.getRawParentPath();
     PolarisEntity tableLikeEntity = resolvedPathWrapper.getRawLeafEntity();
@@ -2036,6 +2164,77 @@ private PrivilegeResult grantPrivilegeOnTableLikeToRole(
         privilege);
   }
 
+  /**
+   * Creates and persists the missing synthetic table-like entity and its parent namespace entities
+   * for external catalogs.
+   *
+   * @param catalogEntity the external passthrough facade catalog entity.
+   * @param identifier the path of the table-like entity(including the namespace).
+   * @param subTypes the expected subtypes of the table-like entity
+   * @param existingPathWrapper the partially resolved path currently stored in the metastore.
+   * @return the resolved path wrapper
+   */
+  private PolarisResolvedPathWrapper createSyntheticTableLikeEntities(
+      CatalogEntity catalogEntity,
+      TableIdentifier identifier,
+      List<PolarisEntitySubType> subTypes,
+      PolarisResolvedPathWrapper existingPathWrapper) {
+
+    Namespace namespace = identifier.namespace();
+    PolarisResolvedPathWrapper resolvedNamespacePathWrapper =
+        !namespace.isEmpty()
+            ? createSyntheticNamespaceEntities(catalogEntity, namespace, existingPathWrapper)
+            : existingPathWrapper;
+
+    if (resolvedNamespacePathWrapper == null
+        || (!namespace.isEmpty()
+            && !resolvedNamespacePathWrapper.isFullyResolvedNamespace(
+                catalogEntity.getName(), namespace))) {
+      throw new RuntimeException(
+          String.format(
+              "Failed to create synthetic namespace entities for namespace %s in catalog %s",
+              namespace.toString(), catalogEntity.getName()));
+    }
+
+    PolarisEntity parentNamespaceEntity = resolvedNamespacePathWrapper.getRawLeafEntity();
+
+    // TODO: Once we support GENERIC_TABLE federation, select the intended type depending on the
+    // callsite; if it is instantiated via an Iceberg RESTCatalog factory or a different factory
+    // for GenericCatalogs.
+    PolarisEntitySubType syntheticEntitySubType = selectEntitySubType(subTypes);
+
+    // TODO: Instead of creating a synthetic table-like entity, rely on external catalog mediated
+    // backfill and use the metadata location from the external catalog.
+    PolarisEntity syntheticTableEntity =
+        new IcebergTableLikeEntity.Builder(syntheticEntitySubType, identifier, "")
+            .setParentId(parentNamespaceEntity.getId())
+            .setId(metaStoreManager.generateNewEntityId(getCurrentPolarisContext()).getId())
+            .setCatalogId(parentNamespaceEntity.getCatalogId())
+            .setCreateTimestamp(System.currentTimeMillis())
+            .build();
+    // We will re-resolve later anyway, so
+    metaStoreManager.createEntityIfNotExists(
+        getCurrentPolarisContext(),
+        PolarisEntity.toCoreList(resolvedNamespacePathWrapper.getRawFullPath()),
+        syntheticTableEntity);
+
+    PolarisResolvedPathWrapper completePathWrapper =
+        resolutionManifest.getPassthroughResolvedPath(identifier);
+    PolarisEntity leafEntity =
+        completePathWrapper != null ? completePathWrapper.getRawLeafEntity() : null;
+    if (completePathWrapper == null
+        || leafEntity == null
+        || !(leafEntity.getType() == PolarisEntityType.TABLE_LIKE
+            && leafEntity.getSubType() == PolarisEntitySubType.ICEBERG_TABLE
+            && Objects.equals(leafEntity.getName(), identifier.name()))) {
+      throw new RuntimeException(
+          String.format(
+              "Failed to create or find table entity '%s' in federated catalog '%s'",
+              identifier.name(), catalogEntity.getName()));
+    }
+    return completePathWrapper;
+  }
+
   /**
    * Removes a table-level or view-level grant on {@code identifier} from {@code catalogRoleName}.
    */
@@ -2125,4 +2324,25 @@ private PrivilegeResult revokePrivilegeOnPolicyEntityFromRole(
         policyEntity,
         privilege);
   }
+
+  /**
+   * Selects the appropriate entity subtype for synthetic entities in external catalogs.
+   *
+   * @param subTypes list of candidate subtypes
+   * @return the selected subtype for the synthetic entity
+   * @throws IllegalStateException if no supported subtype is found
+   */
+  private static PolarisEntitySubType selectEntitySubType(List<PolarisEntitySubType> subTypes) {
+    if (subTypes.contains(PolarisEntitySubType.ICEBERG_TABLE)) {
+      return PolarisEntitySubType.ICEBERG_TABLE;
+    } else if (subTypes.contains(PolarisEntitySubType.ICEBERG_VIEW)) {
+      return PolarisEntitySubType.ICEBERG_VIEW;
+    } else {
+      throw new IllegalStateException(
+          String.format(
+              "No supported subtype found in %s. Only ICEBERG_TABLE and ICEBERG_VIEW are"
+                  + " supported for synthetic entities in external catalogs.",
+              subTypes));
+    }
+  }
 }