Replace CallContext with RealmConfig in CatalogEntity

polaris-core/src/main/java/org/apache/polaris/core/entity/CatalogEntity.java

@@ -40,8 +40,8 @@
 import org.apache.polaris.core.admin.model.PolarisCatalog;
 import org.apache.polaris.core.admin.model.StorageConfigInfo;
 import org.apache.polaris.core.config.BehaviorChangeConfiguration;
+import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
-import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.secrets.UserSecretReference;
 import org.apache.polaris.core.storage.FileStorageConfigurationInfo;
 import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
@@ -79,7 +79,7 @@ public static CatalogEntity of(PolarisBaseEntity sourceEntity) {
     return null;
   }
 
-  public static CatalogEntity fromCatalog(CallContext callContext, Catalog catalog) {
+  public static CatalogEntity fromCatalog(RealmConfig realmConfig, Catalog catalog) {
     Builder builder =
         new Builder()
             .setName(catalog.getName())
@@ -89,7 +89,7 @@ public static CatalogEntity fromCatalog(CallContext callContext, Catalog catalog
     internalProperties.put(CATALOG_TYPE_PROPERTY, catalog.getType().name());
     builder.setInternalProperties(internalProperties);
     builder.setStorageConfigurationInfo(
-        callContext, catalog.getStorageConfigInfo(), getBaseLocation(catalog));
+        realmConfig, catalog.getStorageConfigInfo(), getBaseLocation(catalog));
     return builder.build();
   }
 
@@ -250,7 +250,7 @@ public Builder setReplaceNewLocationPrefixWithCatalogDefault(String value) {
     }
 
     public Builder setStorageConfigurationInfo(
-        CallContext callContext, StorageConfigInfo storageConfigModel, String defaultBaseLocation) {
+        RealmConfig realmConfig, StorageConfigInfo storageConfigModel, String defaultBaseLocation) {
       if (storageConfigModel != null) {
         PolarisStorageConfigurationInfo config;
         Set<String> allowedLocations = new HashSet<>(storageConfigModel.getAllowedLocations());
@@ -264,7 +264,7 @@ public Builder setStorageConfigurationInfo(
           throw new BadRequestException("Must specify default base location");
         }
         allowedLocations.add(defaultBaseLocation);
-        validateMaxAllowedLocations(callContext, allowedLocations);
+        validateMaxAllowedLocations(realmConfig, allowedLocations);
         switch (storageConfigModel.getStorageType()) {
           case S3:
             AwsStorageConfigInfo awsConfigModel = (AwsStorageConfigInfo) storageConfigModel;
@@ -315,11 +315,9 @@ public Builder setStorageConfigurationInfo(
 
     /** Validate the number of allowed locations not exceeding the max value. */
     private void validateMaxAllowedLocations(
-        CallContext callContext, Collection<String> allowedLocations) {
+        RealmConfig realmConfig, Collection<String> allowedLocations) {
       int maxAllowedLocations =
-          callContext
-              .getRealmConfig()
-              .getConfig(BehaviorChangeConfiguration.STORAGE_CONFIGURATION_MAX_LOCATIONS);
+          realmConfig.getConfig(BehaviorChangeConfiguration.STORAGE_CONFIGURATION_MAX_LOCATIONS);
       if (maxAllowedLocations != -1 && allowedLocations.size() > maxAllowedLocations) {
         throw new IllegalArgumentException(
             String.format(

polaris-core/src/main/java/org/apache/polaris/core/storage/PolarisStorageConfigurationInfo.java

@@ -34,9 +34,9 @@
 import java.util.Locale;
 import java.util.Optional;
 import java.util.Set;
-import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.admin.model.Catalog;
 import org.apache.polaris.core.config.FeatureConfiguration;
+import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.entity.CatalogEntity;
 import org.apache.polaris.core.entity.PolarisEntity;
 import org.apache.polaris.core.entity.PolarisEntityConstants;
@@ -117,7 +117,7 @@ public static PolarisStorageConfigurationInfo deserialize(final @Nonnull String
   }
 
   public static Optional<PolarisStorageConfigurationInfo> forEntityPath(
-      PolarisCallContext callContext, List<PolarisEntity> entityPath) {
+      RealmConfig realmConfig, List<PolarisEntity> entityPath) {
     return findStorageInfoFromHierarchy(entityPath)
         .map(
             storageInfo ->
@@ -142,9 +142,8 @@ public static Optional<PolarisStorageConfigurationInfo> forEntityPath(
                       .orElse(null);
               CatalogEntity catalog = CatalogEntity.of(entityPath.get(0));
               boolean allowEscape =
-                  callContext
-                      .getRealmConfig()
-                      .getConfig(FeatureConfiguration.ALLOW_UNSTRUCTURED_TABLE_LOCATION, catalog);
+                  realmConfig.getConfig(
+                      FeatureConfiguration.ALLOW_UNSTRUCTURED_TABLE_LOCATION, catalog);
               if (!allowEscape
                   && catalog.getCatalogType() != Catalog.TypeEnum.EXTERNAL
                   && baseLocation != null) {

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

@@ -755,7 +755,8 @@ public PolarisEntity createCatalog(CreateCatalogRequest catalogRequest) {
     PolarisAuthorizableOperation op = PolarisAuthorizableOperation.CREATE_CATALOG;
     authorizeBasicRootOperationOrThrow(op);
 
-    CatalogEntity entity = CatalogEntity.fromCatalog(callContext, catalogRequest.getCatalog());
+    CatalogEntity entity =
+        CatalogEntity.fromCatalog(callContext.getRealmConfig(), catalogRequest.getCatalog());
 
     checkArgument(entity.getId() == -1, "Entity to be created must have no ID assigned");
 
@@ -960,7 +961,7 @@ private void validateUpdateCatalogDiffOrThrow(
     }
     if (updateRequest.getStorageConfigInfo() != null) {
       updateBuilder.setStorageConfigurationInfo(
-          callContext, updateRequest.getStorageConfigInfo(), defaultBaseLocation);
+          callContext.getRealmConfig(), updateRequest.getStorageConfigInfo(), defaultBaseLocation);
     }
     CatalogEntity updatedEntity = updateBuilder.build();
 

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalog.java

@@ -1007,7 +1007,7 @@ private void validateLocationsForTableLike(
       PolarisResolvedPathWrapper resolvedStorageEntity) {
     Optional<PolarisStorageConfigurationInfo> optStorageConfiguration =
         PolarisStorageConfigurationInfo.forEntityPath(
-            callContext.getPolarisCallContext(), resolvedStorageEntity.getRawFullPath());
+            callContext.getRealmConfig(), resolvedStorageEntity.getRawFullPath());
 
     optStorageConfiguration.ifPresentOrElse(
         storageConfigInfo -> {

runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAuthzTestBase.java

@@ -55,6 +55,7 @@
 import org.apache.polaris.core.auth.PolarisAuthorizer;
 import org.apache.polaris.core.auth.PolarisAuthorizerImpl;
 import org.apache.polaris.core.config.PolarisConfigurationStore;
+import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.CatalogEntity;
@@ -199,6 +200,7 @@ public Map<String, String> getConfigOverrides() {
   protected PolarisBaseEntity catalogEntity;
   protected PrincipalEntity principalEntity;
   protected CallContext callContext;
+  protected RealmConfig realmConfig;
   protected AuthenticatedPolarisPrincipal authenticatedRoot;
   protected PolarisAuthorizer polarisAuthorizer;
 
@@ -234,6 +236,7 @@ public void before(TestInfo testInfo) {
             configurationStore);
 
     callContext = polarisContext;
+    realmConfig = polarisContext.getRealmConfig();
 
     PrincipalEntity rootPrincipal =
         metaStoreManager.findRootPrincipal(polarisContext).orElseThrow();
@@ -262,7 +265,7 @@ public void before(TestInfo testInfo) {
                     .setName(CATALOG_NAME)
                     .setCatalogType("INTERNAL")
                     .setDefaultBaseLocation(storageLocation)
-                    .setStorageConfigurationInfo(callContext, storageConfigModel, storageLocation)
+                    .setStorageConfigurationInfo(realmConfig, storageConfigModel, storageLocation)
                     .build()
                     .asCatalog()));
 

runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractIcebergCatalogTest.java

@@ -235,6 +235,7 @@ public Map<String, String> getConfigOverrides() {
   private PolarisMetaStoreManager metaStoreManager;
   private UserSecretsManager userSecretsManager;
   private PolarisCallContext polarisContext;
+  private RealmConfig realmConfig;
   private PolarisAdminService adminService;
   private ResolverFactory resolverFactory;
   private ResolutionManifestFactory resolutionManifestFactory;
@@ -279,8 +280,9 @@ public void before(TestInfo testInfo) {
             metaStoreManagerFactory.getOrCreateSession(realmContext),
             diagServices,
             configurationStore);
+    realmConfig = polarisContext.getRealmConfig();
 
-    EntityCache entityCache = createEntityCache(polarisContext.getRealmConfig(), metaStoreManager);
+    EntityCache entityCache = createEntityCache(realmConfig, metaStoreManager);
     resolverFactory =
         (callContext, securityContext, referenceCatalogName) ->
             new Resolver(
@@ -337,8 +339,7 @@ public void before(TestInfo testInfo) {
                         "true")
                     .addProperty(
                         FeatureConfiguration.DROP_WITH_PURGE_ENABLED.catalogConfig(), "true")
-                    .setStorageConfigurationInfo(
-                        polarisContext, storageConfigModel, storageLocation)
+                    .setStorageConfigurationInfo(realmConfig, storageConfigModel, storageLocation)
                     .build()
                     .asCatalog()));
 
@@ -1327,8 +1328,7 @@ public void testUpdateNotificationCreateTableWithLocalFilePrefix() {
         metadataLocation,
         TableMetadataParser.toJson(createSampleTableMetadata(metadataLocation)).getBytes(UTF_8));
 
-    if (!polarisContext
-        .getRealmConfig()
+    if (!realmConfig
         .getConfig(FeatureConfiguration.SUPPORTED_CATALOG_STORAGE_TYPES)
         .contains("FILE")) {
       Assertions.assertThatThrownBy(() -> catalog.sendNotification(table, request))
@@ -1380,8 +1380,7 @@ public void testUpdateNotificationCreateTableWithHttpPrefix() {
         metadataLocation,
         TableMetadataParser.toJson(createSampleTableMetadata(metadataLocation)).getBytes(UTF_8));
 
-    if (!polarisContext
-        .getRealmConfig()
+    if (!realmConfig
         .getConfig(FeatureConfiguration.SUPPORTED_CATALOG_STORAGE_TYPES)
         .contains("FILE")) {
       Assertions.assertThatThrownBy(() -> catalog.sendNotification(table, request))
@@ -1871,7 +1870,7 @@ public void testDropTableWithPurgeDisabled() {
                     FeatureConfiguration.ALLOW_UNSTRUCTURED_TABLE_LOCATION.catalogConfig(), "true")
                 .addProperty(FeatureConfiguration.DROP_WITH_PURGE_ENABLED.catalogConfig(), "false")
                 .setStorageConfigurationInfo(
-                    polarisContext, noPurgeStorageConfigModel, storageLocation)
+                    realmConfig, noPurgeStorageConfigModel, storageLocation)
                 .build()
                 .asCatalog()));
     IcebergCatalog noPurgeCatalog =

runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractIcebergCatalogViewTest.java

@@ -44,6 +44,7 @@
 import org.apache.polaris.core.auth.PolarisAuthorizerImpl;
 import org.apache.polaris.core.config.FeatureConfiguration;
 import org.apache.polaris.core.config.PolarisConfigurationStore;
+import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.CatalogEntity;
 import org.apache.polaris.core.entity.PrincipalEntity;
@@ -120,6 +121,7 @@ public Map<String, String> getConfigOverrides() {
   private PolarisMetaStoreManager metaStoreManager;
   private UserSecretsManager userSecretsManager;
   private PolarisCallContext polarisContext;
+  private RealmConfig realmConfig;
 
   private TestPolarisEventListener testPolarisEventListener;
 
@@ -160,6 +162,7 @@ public void before(TestInfo testInfo) {
             metaStoreManagerFactory.getOrCreateSession(realmContext),
             diagServices,
             configurationStore);
+    realmConfig = polarisContext.getRealmConfig();
 
     PrincipalEntity rootPrincipal =
         metaStoreManager.findRootPrincipal(polarisContext).orElseThrow();
@@ -192,7 +195,7 @@ public void before(TestInfo testInfo) {
                 .addProperty(FeatureConfiguration.DROP_WITH_PURGE_ENABLED.catalogConfig(), "true")
                 .setDefaultBaseLocation("file://tmp")
                 .setStorageConfigurationInfo(
-                    polarisContext,
+                    realmConfig,
                     new FileStorageConfigInfo(
                         StorageConfigInfo.StorageTypeEnum.FILE, List.of("file://", "/", "*")),
                     "file://tmp")

runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractPolarisGenericTableCatalogTest.java

@@ -45,6 +45,7 @@
 import org.apache.polaris.core.auth.PolarisAuthorizerImpl;
 import org.apache.polaris.core.config.FeatureConfiguration;
 import org.apache.polaris.core.config.PolarisConfigurationStore;
+import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.CatalogEntity;
 import org.apache.polaris.core.entity.PolarisEntity;
@@ -110,6 +111,7 @@ public abstract class AbstractPolarisGenericTableCatalogTest {
   private PolarisMetaStoreManager metaStoreManager;
   private UserSecretsManager userSecretsManager;
   private PolarisCallContext polarisContext;
+  private RealmConfig realmConfig;
   private PolarisAdminService adminService;
   private FileIOFactory fileIOFactory;
   private AuthenticatedPolarisPrincipal authenticatedRoot;
@@ -152,6 +154,7 @@ public void before(TestInfo testInfo) {
             metaStoreManagerFactory.getOrCreateSession(realmContext),
             diagServices,
             configurationStore);
+    realmConfig = polarisContext.getRealmConfig();
 
     PrincipalEntity rootPrincipal =
         metaStoreManager.findRootPrincipal(polarisContext).orElseThrow();
@@ -196,8 +199,7 @@ public void before(TestInfo testInfo) {
                         "true")
                     .addProperty(
                         FeatureConfiguration.DROP_WITH_PURGE_ENABLED.catalogConfig(), "true")
-                    .setStorageConfigurationInfo(
-                        polarisContext, storageConfigModel, storageLocation)
+                    .setStorageConfigurationInfo(realmConfig, storageConfigModel, storageLocation)
                     .build()
                     .asCatalog()));
 

runtime/service/src/test/java/org/apache/polaris/service/catalog/AbstractPolicyCatalogTest.java

@@ -52,7 +52,7 @@
 import org.apache.polaris.core.auth.PolarisAuthorizerImpl;
 import org.apache.polaris.core.config.FeatureConfiguration;
 import org.apache.polaris.core.config.PolarisConfigurationStore;
-import org.apache.polaris.core.context.CallContext;
+import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.CatalogEntity;
 import org.apache.polaris.core.entity.PolarisEntity;
@@ -132,12 +132,12 @@ public abstract class AbstractPolicyCatalogTest {
 
   private PolicyCatalog policyCatalog;
   private IcebergCatalog icebergCatalog;
-  private CallContext callContext;
   private AwsStorageConfigInfo storageConfigModel;
   private String realmName;
   private PolarisMetaStoreManager metaStoreManager;
   private UserSecretsManager userSecretsManager;
   private PolarisCallContext polarisContext;
+  private RealmConfig realmConfig;
   private PolarisAdminService adminService;
   private FileIOFactory fileIOFactory;
   private AuthenticatedPolarisPrincipal authenticatedRoot;
@@ -175,8 +175,7 @@ public void before(TestInfo testInfo) {
             metaStoreManagerFactory.getOrCreateSession(realmContext),
             diagServices,
             configurationStore);
-
-    callContext = polarisContext;
+    realmConfig = polarisContext.getRealmConfig();
 
     PrincipalEntity rootPrincipal =
         metaStoreManager.findRootPrincipal(polarisContext).orElseThrow();
@@ -190,7 +189,7 @@ public void before(TestInfo testInfo) {
 
     adminService =
         new PolarisAdminService(
-            callContext,
+            polarisContext,
             resolutionManifestFactory,
             metaStoreManager,
             userSecretsManager,
@@ -219,14 +218,13 @@ public void before(TestInfo testInfo) {
                     .addProperty(
                         FeatureConfiguration.ALLOW_UNSTRUCTURED_TABLE_LOCATION.catalogConfig(),
                         "true")
-                    .setStorageConfigurationInfo(
-                        polarisContext, storageConfigModel, storageLocation)
+                    .setStorageConfigurationInfo(realmConfig, storageConfigModel, storageLocation)
                     .build()
                     .asCatalog()));
 
     PolarisPassthroughResolutionView passthroughView =
         new PolarisPassthroughResolutionView(
-            callContext, resolutionManifestFactory, securityContext, CATALOG_NAME);
+            polarisContext, resolutionManifestFactory, securityContext, CATALOG_NAME);
     TaskExecutor taskExecutor = Mockito.mock();
     this.fileIOFactory = new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory);
 
@@ -250,13 +248,13 @@ public void before(TestInfo testInfo) {
             isA(AwsStorageConfigurationInfo.class)))
         .thenReturn((PolarisStorageIntegration) storageIntegration);
 
-    this.policyCatalog = new PolicyCatalog(metaStoreManager, callContext, passthroughView);
+    this.policyCatalog = new PolicyCatalog(metaStoreManager, polarisContext, passthroughView);
     this.icebergCatalog =
         new IcebergCatalog(
             storageCredentialCache,
             resolverFactory,
             metaStoreManager,
-            callContext,
+            polarisContext,
             passthroughView,
             securityContext,
             taskExecutor,

runtime/service/src/test/java/org/apache/polaris/service/catalog/IcebergCatalogHandlerAuthzTest.java

@@ -1722,7 +1722,7 @@ public void testSendNotificationSufficientPrivileges() {
             new CatalogEntity.Builder()
                 .setName(externalCatalog)
                 .setDefaultBaseLocation(storageLocation)
-                .setStorageConfigurationInfo(callContext, storageConfigModel, storageLocation)
+                .setStorageConfigurationInfo(realmConfig, storageConfigModel, storageLocation)
                 .setCatalogType("EXTERNAL")
                 .build()
                 .asCatalog()));