Make *StorageConfigurationInfo types immutable (#2236)

This change eventually enables usage of the `*StorageConfigurationInfo` in the `StorageCredentialCacheKey` due to the then memoized hash-code values, to eliminate a couple of JSON re-serializations.

Author: snazy

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

@@ -22,7 +22,6 @@
 
 import jakarta.annotation.Nonnull;
 import jakarta.annotation.Nullable;
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -269,37 +268,40 @@ public Builder setStorageConfigurationInfo(
           case S3:
             AwsStorageConfigInfo awsConfigModel = (AwsStorageConfigInfo) storageConfigModel;
             AwsStorageConfigurationInfo awsConfig =
-                new AwsStorageConfigurationInfo(
-                    PolarisStorageConfigurationInfo.StorageType.S3,
-                    new ArrayList<>(allowedLocations),
-                    awsConfigModel.getRoleArn(),
-                    awsConfigModel.getExternalId(),
-                    awsConfigModel.getRegion(),
-                    awsConfigModel.getEndpoint(),
-                    awsConfigModel.getStsEndpoint(),
-                    awsConfigModel.getPathStyleAccess(),
-                    awsConfigModel.getEndpointInternal());
+                AwsStorageConfigurationInfo.builder()
+                    .allowedLocations(allowedLocations)
+                    .roleARN(awsConfigModel.getRoleArn())
+                    .externalId(awsConfigModel.getExternalId())
+                    .region(awsConfigModel.getRegion())
+                    .endpoint(awsConfigModel.getEndpoint())
+                    .stsEndpoint(awsConfigModel.getStsEndpoint())
+                    .pathStyleAccess(awsConfigModel.getPathStyleAccess())
+                    .endpointInternal(awsConfigModel.getEndpointInternal())
+                    .build();
             awsConfig.validateArn(awsConfigModel.getRoleArn());
             config = awsConfig;
             break;
           case AZURE:
             AzureStorageConfigInfo azureConfigModel = (AzureStorageConfigInfo) storageConfigModel;
-            AzureStorageConfigurationInfo azureConfigInfo =
-                new AzureStorageConfigurationInfo(
-                    new ArrayList<>(allowedLocations), azureConfigModel.getTenantId());
-            azureConfigInfo.setMultiTenantAppName(azureConfigModel.getMultiTenantAppName());
-            azureConfigInfo.setConsentUrl(azureConfigModel.getConsentUrl());
-            config = azureConfigInfo;
+            config =
+                AzureStorageConfigurationInfo.builder()
+                    .allowedLocations(allowedLocations)
+                    .tenantId(azureConfigModel.getTenantId())
+                    .multiTenantAppName(azureConfigModel.getMultiTenantAppName())
+                    .consentUrl(azureConfigModel.getConsentUrl())
+                    .build();
             break;
           case GCS:
-            GcpStorageConfigurationInfo gcpConfig =
-                new GcpStorageConfigurationInfo(new ArrayList<>(allowedLocations));
-            gcpConfig.setGcpServiceAccount(
-                ((GcpStorageConfigInfo) storageConfigModel).getGcsServiceAccount());
-            config = gcpConfig;
+            config =
+                GcpStorageConfigurationInfo.builder()
+                    .allowedLocations(allowedLocations)
+                    .gcpServiceAccount(
+                        ((GcpStorageConfigInfo) storageConfigModel).getGcsServiceAccount())
+                    .build();
             break;
           case FILE:
-            config = new FileStorageConfigurationInfo(new ArrayList<>(allowedLocations));
+            config =
+                FileStorageConfigurationInfo.builder().allowedLocations(allowedLocations).build();
             break;
           default:
             throw new IllegalStateException(

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

@@ -18,29 +18,37 @@
  */
 package org.apache.polaris.core.storage;
 
-import com.fasterxml.jackson.annotation.JsonProperty;
-import jakarta.annotation.Nonnull;
-import java.util.List;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import java.util.Locale;
+import org.apache.polaris.immutables.PolarisImmutable;
 
 /**
  * Support for file:// URLs in storage configuration. This is pretty-much only used for testing.
  * Supports URLs that start with file:// or /, but also supports wildcard (*) to support certain
  * test cases.
  */
-public class FileStorageConfigurationInfo extends PolarisStorageConfigurationInfo {
+@PolarisImmutable
+@JsonSerialize(as = ImmutableFileStorageConfigurationInfo.class)
+@JsonDeserialize(as = ImmutableFileStorageConfigurationInfo.class)
+@JsonTypeName("FileStorageConfigurationInfo")
+public abstract class FileStorageConfigurationInfo extends PolarisStorageConfigurationInfo {
 
-  public FileStorageConfigurationInfo(
-      @JsonProperty(value = "allowedLocations", required = true) @Nonnull
-          List<String> allowedLocations) {
-    super(StorageType.FILE, allowedLocations);
+  public static ImmutableFileStorageConfigurationInfo.Builder builder() {
+    return ImmutableFileStorageConfigurationInfo.builder();
   }
 
   @Override
   public String getFileIoImplClassName() {
     return "org.apache.iceberg.hadoop.HadoopFileIO";
   }
 
+  @Override
+  public StorageType getStorageType() {
+    return StorageType.FILE;
+  }
+
   @Override
   public void validatePrefixForStorageType(String loc) {
     if (getStorageType().getPrefixes().stream()

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

@@ -18,8 +18,9 @@
  */
 package org.apache.polaris.core.storage;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonInclude;
-import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import com.fasterxml.jackson.core.JsonProcessingException;
@@ -46,6 +47,7 @@
 import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
 import org.apache.polaris.core.storage.azure.AzureStorageConfigurationInfo;
 import org.apache.polaris.core.storage.gcp.GcpStorageConfigurationInfo;
+import org.immutables.value.Value;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -65,40 +67,28 @@
   @JsonSubTypes.Type(value = GcpStorageConfigurationInfo.class),
   @JsonSubTypes.Type(value = FileStorageConfigurationInfo.class),
 })
+@JsonIgnoreProperties(ignoreUnknown = true)
 public abstract class PolarisStorageConfigurationInfo {
 
   private static final Logger LOGGER =
       LoggerFactory.getLogger(PolarisStorageConfigurationInfo.class);
 
-  // a list of allowed locations
-  private final List<String> allowedLocations;
-
-  // storage type
-  private final StorageType storageType;
-
-  public PolarisStorageConfigurationInfo(
-      @JsonProperty(value = "storageType", required = true) @Nonnull StorageType storageType,
-      @JsonProperty(value = "allowedLocations", required = true) @Nonnull
-          List<String> allowedLocations) {
-    this(storageType, allowedLocations, true);
-  }
-
-  protected PolarisStorageConfigurationInfo(
-      StorageType storageType, List<String> allowedLocations, boolean validatePrefix) {
-    this.allowedLocations = allowedLocations;
-    this.storageType = storageType;
-    if (validatePrefix) {
-      allowedLocations.forEach(this::validatePrefixForStorageType);
+  @Value.Check
+  protected void check() {
+    if (validatePrefix()) {
+      getAllowedLocations().forEach(this::validatePrefixForStorageType);
     }
   }
 
-  public List<String> getAllowedLocations() {
-    return allowedLocations;
+  @JsonIgnore
+  @Value.Auxiliary
+  public boolean validatePrefix() {
+    return true;
   }
 
-  public StorageType getStorageType() {
-    return storageType;
-  }
+  public abstract List<String> getAllowedLocations();
+
+  public abstract StorageType getStorageType();
 
   private static final ObjectMapper DEFAULT_MAPPER;
 
@@ -213,11 +203,12 @@ private static List<String> userSpecifiedWriteLocations(Map<String, String> prop
 
   /** Validate if the provided allowed locations are valid for the storage type */
   protected void validatePrefixForStorageType(String loc) {
-    if (storageType.prefixes.stream().noneMatch(p -> loc.toLowerCase(Locale.ROOT).startsWith(p))) {
+    if (getStorageType().prefixes.stream()
+        .noneMatch(p -> loc.toLowerCase(Locale.ROOT).startsWith(p))) {
       throw new IllegalArgumentException(
           String.format(
               "Location prefix not allowed: '%s', expected prefixes: '%s'",
-              loc, String.join(",", storageType.prefixes)));
+              loc, String.join(",", getStorageType().prefixes)));
     }
   }
 

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

@@ -30,20 +30,36 @@
 public class StorageConfigurationOverride extends PolarisStorageConfigurationInfo {
 
   private final PolarisStorageConfigurationInfo parentStorageConfiguration;
+  private final List<String> allowedLocations;
 
   public StorageConfigurationOverride(
       @Nonnull PolarisStorageConfigurationInfo parentStorageConfiguration,
       List<String> allowedLocations) {
-    super(parentStorageConfiguration.getStorageType(), allowedLocations, false);
     this.parentStorageConfiguration = parentStorageConfiguration;
+    this.allowedLocations = List.copyOf(allowedLocations);
     allowedLocations.forEach(this::validatePrefixForStorageType);
   }
 
+  @Override
+  public List<String> getAllowedLocations() {
+    return allowedLocations;
+  }
+
+  @Override
+  public StorageType getStorageType() {
+    return parentStorageConfiguration.getStorageType();
+  }
+
   @Override
   public String getFileIoImplClassName() {
     return parentStorageConfiguration.getFileIoImplClassName();
   }
 
+  @Override
+  public boolean validatePrefix() {
+    return false;
+  }
+
   // delegate to the wrapped class in case they override the parent behavior
   @Override
   protected void validatePrefixForStorageType(String loc) {