Support persisting TableMetadata in the metastore

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

@@ -19,6 +19,7 @@
 package org.apache.polaris.core.config;
 
 import java.util.Optional;
+import java.util.function.Function;
 
 /**
  * Internal configuration flags for non-feature behavior changes in Polaris. These flags control
@@ -37,8 +38,9 @@ protected BehaviorChangeConfiguration(
       String description,
       T defaultValue,
       Optional<String> catalogConfig,
-      Optional<String> catalogConfigUnsafe) {
-    super(key, description, defaultValue, catalogConfig, catalogConfigUnsafe);
+      Optional<String> catalogConfigUnsafe,
+      Optional<Function<T, Boolean>> validation) {
+    super(key, description, defaultValue, catalogConfig, catalogConfigUnsafe, validation);
   }
 
   public static final BehaviorChangeConfiguration<Boolean> VALIDATE_VIEW_LOCATION_OVERLAP =

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

@@ -20,6 +20,7 @@
 
 import java.util.List;
 import java.util.Optional;
+import java.util.function.Function;
 import org.apache.polaris.core.admin.model.StorageConfigInfo;
 import org.apache.polaris.core.connection.ConnectionType;
 import org.apache.polaris.core.context.CallContext;
@@ -38,8 +39,14 @@ protected FeatureConfiguration(
       String description,
       T defaultValue,
       Optional<String> catalogConfig,
-      Optional<String> catalogConfigUnsafe) {
-    super(key, description, defaultValue, catalogConfig, catalogConfigUnsafe);
+      Optional<String> catalogConfigUnsafe,
+      Optional<Function<T, Boolean>> validation) {
+    super(key, description, defaultValue, catalogConfig, catalogConfigUnsafe, validation);
+  }
+
+  public static final class Constants {
+    public static final int METADATA_CACHE_MAX_BYTES_NO_CACHING = 0;
+    public static final int METADATA_CACHE_MAX_BYTES_INFINITE_CACHING = -1;
   }
 
   /**
@@ -287,4 +294,25 @@ public static void enforceFeatureEnabledOrThrow(
                   + "This should only be set to 'true' for tests!")
           .defaultValue(false)
           .buildFeatureConfiguration();
+
+  public static final PolarisConfiguration<Integer> METADATA_CACHE_MAX_BYTES =
+      PolarisConfiguration.<Integer>builder()
+          .key("METADATA_CACHE_MAX_BYTES")
+          .catalogConfig("polaris.config.metadata-cache-max-bytes")
+          .description(
+              "If nonzero, the approximate max size a table's metadata can be in order to be cached in the persistence"
+                  + " layer. If zero, no metadata will be cached or served from the cache. If -1, all metadata"
+                  + " will be cached.")
+          .defaultValue(Constants.METADATA_CACHE_MAX_BYTES_NO_CACHING)
+          .validation(value -> value >= -1)
+          .buildFeatureConfiguration();
+
+  public static final PolarisConfiguration<Boolean> ALWAYS_FILTER_SNAPSHOTS =
+      PolarisConfiguration.<Boolean>builder()
+          .key("ALWAYS_FILTER_SNAPSHOTS")
+          .description(
+              "If set, Polaris will always attempt to filter snapshots from a LoadTableResponse even when "
+                  + "doing so requires additional serialization of the TableMetadata")
+          .defaultValue(true)
+          .buildFeatureConfiguration();
 }

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

@@ -46,6 +46,7 @@ public abstract class PolarisConfiguration<T> {
   private final Optional<String> catalogConfigImpl;
   private final Optional<String> catalogConfigUnsafeImpl;
   private final Class<T> typ;
+  private final Optional<Function<T, Boolean>> validation;
 
   /** catalog configs are expected to start with this prefix */
   private static final String SAFE_CATALOG_CONFIG_PREFIX = "polaris.config.";
@@ -90,13 +91,18 @@ protected PolarisConfiguration(
       String description,
       T defaultValue,
       Optional<String> catalogConfig,
-      Optional<String> catalogConfigUnsafe) {
+      Optional<String> catalogConfigUnsafe,
+      Optional<Function<T, Boolean>> validation) {
     this.key = key;
     this.description = description;
     this.defaultValue = defaultValue;
     this.catalogConfigImpl = catalogConfig;
     this.catalogConfigUnsafeImpl = catalogConfigUnsafe;
     this.typ = (Class<T>) defaultValue.getClass();
+    this.validation = validation;
+
+    // Force validation:
+    apply(defaultValue);
   }
 
   public boolean hasCatalogConfig() {
@@ -110,6 +116,22 @@ public String catalogConfig() {
                 "Attempted to read a catalog config key from a configuration that doesn't have one."));
   }
 
+  T apply(Object value) {
+    T result = this.typ.cast(value);
+    validate(result);
+    return result;
+  }
+
+  private void validate(T value) {
+    this.validation.ifPresent(
+        v -> {
+          if (!v.apply(value)) {
+            throw new IllegalArgumentException(
+                String.format("Configuration %s has invalid value %s", key, defaultValue));
+          }
+        });
+  }
+
   public boolean hasCatalogConfigUnsafe() {
     return catalogConfigUnsafeImpl.isPresent();
   }
@@ -121,16 +143,13 @@ public String catalogConfigUnsafe() {
                 "Attempted to read an unsafe catalog config key from a configuration that doesn't have one."));
   }
 
-  T cast(Object value) {
-    return this.typ.cast(value);
-  }
-
   public static class Builder<T> {
     private String key;
     private String description;
     private T defaultValue;
     private Optional<String> catalogConfig = Optional.empty();
     private Optional<String> catalogConfigUnsafe = Optional.empty();
+    private Optional<Function<T, Boolean>> validation = Optional.empty();
 
     public Builder<T> key(String key) {
       this.key = key;
@@ -162,6 +181,11 @@ public Builder<T> catalogConfig(String catalogConfig) {
       return this;
     }
 
+    public Builder<T> validation(Function<T, Boolean> validation) {
+      this.validation = Optional.of(validation);
+      return this;
+    }
+
     /**
      * Used to support backwards compatability before there were reserved properties. Usage of this
      * method should be removed over time.
@@ -177,7 +201,7 @@ public Builder<T> catalogConfigUnsafe(String catalogConfig) {
       return this;
     }
 
-    private void validateOrThrow() {
+    private void validateFieldsOrThrow() {
       if (key == null || description == null || defaultValue == null) {
         throw new IllegalArgumentException("key, description, and defaultValue are required");
       }
@@ -187,23 +211,23 @@ private void validateOrThrow() {
     }
 
     public FeatureConfiguration<T> buildFeatureConfiguration() {
-      validateOrThrow();
+      validateFieldsOrThrow();
       FeatureConfiguration<T> config =
           new FeatureConfiguration<>(
-              key, description, defaultValue, catalogConfig, catalogConfigUnsafe);
+              key, description, defaultValue, catalogConfig, catalogConfigUnsafe, validation);
       PolarisConfiguration.registerConfiguration(config);
       return config;
     }
 
     public BehaviorChangeConfiguration<T> buildBehaviorChangeConfiguration() {
-      validateOrThrow();
+      validateFieldsOrThrow();
       if (catalogConfig.isPresent() || catalogConfigUnsafe.isPresent()) {
         throw new IllegalArgumentException(
             "catalog configs are not valid for behavior change configs");
       }
       BehaviorChangeConfiguration<T> config =
           new BehaviorChangeConfiguration<>(
-              key, description, defaultValue, catalogConfig, catalogConfigUnsafe);
+              key, description, defaultValue, catalogConfig, catalogConfigUnsafe, validation);
       PolarisConfiguration.registerConfiguration(config);
       return config;
     }

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

@@ -75,17 +75,17 @@ public interface PolarisConfigurationStore {
     }
 
     if (config.defaultValue instanceof Boolean) {
-      return config.cast(Boolean.valueOf(String.valueOf(value)));
+      return config.apply(Boolean.valueOf(String.valueOf(value)));
     } else if (config.defaultValue instanceof Integer) {
-      return config.cast(Integer.valueOf(String.valueOf(value)));
+      return config.apply(Integer.valueOf(String.valueOf(value)));
     } else if (config.defaultValue instanceof Long) {
-      return config.cast(Long.valueOf(String.valueOf(value)));
+      return config.apply(Long.valueOf(String.valueOf(value)));
     } else if (config.defaultValue instanceof Double) {
-      return config.cast(Double.valueOf(String.valueOf(value)));
+      return config.apply(Double.valueOf(String.valueOf(value)));
     } else if (config.defaultValue instanceof List<?>) {
-      return config.cast(new ArrayList<>((List<?>) value));
+      return config.apply(new ArrayList<>((List<?>) value));
     } else {
-      return config.cast(value);
+      return config.apply(value);
     }
   }
 

polaris-core/src/main/java/org/apache/polaris/core/entity/table/IcebergTableLikeEntity.java

@@ -38,6 +38,15 @@ public class IcebergTableLikeEntity extends TableLikeEntity {
   // of the internalProperties JSON file.
   public static final String METADATA_LOCATION_KEY = "metadata-location";
 
+  // For applicable types, this key on the "internalProperties" map will return the content of the
+  // metadata.json file located at `METADATA_CACHE_LOCATION_KEY`
+  public static final String METADATA_CACHE_CONTENT_KEY = "metadata-cache-content";
+
+  // For applicable types, this key on the "internalProperties" map will return the location of the
+  // `metadata.json` that is cached in `METADATA_CACHE_CONTENT_KEY`. This will often match the
+  // current metadata location in `METADATA_LOCATION_KEY`; if it does not the cache is invalid
+  public static final String METADATA_CACHE_LOCATION_KEY = "metadata-cache-location";
+
   public static final String USER_SPECIFIED_WRITE_DATA_LOCATION_KEY = "write.data.path";
   public static final String USER_SPECIFIED_WRITE_METADATA_LOCATION_KEY = "write.metadata.path";
 
@@ -60,6 +69,16 @@ public String getMetadataLocation() {
     return getInternalPropertiesAsMap().get(METADATA_LOCATION_KEY);
   }
 
+  @JsonIgnore
+  public String getMetadataCacheContent() {
+    return getInternalPropertiesAsMap().get(METADATA_CACHE_CONTENT_KEY);
+  }
+
+  @JsonIgnore
+  public String getMetadataCacheLocation() {
+    return getInternalPropertiesAsMap().get(METADATA_CACHE_LOCATION_KEY);
+  }
+
   @JsonIgnore
   public Optional<Long> getLastAdmittedNotificationTimestamp() {
     return Optional.ofNullable(
@@ -114,6 +133,12 @@ public Builder setMetadataLocation(String location) {
       return this;
     }
 
+    public Builder setMetadataContent(String location, String content) {
+      internalProperties.put(METADATA_CACHE_LOCATION_KEY, location);
+      internalProperties.put(METADATA_CACHE_CONTENT_KEY, content);
+      return this;
+    }
+
     public Builder setLastNotificationTimestamp(long timestamp) {
       internalProperties.put(LAST_ADMITTED_NOTIFICATION_TIMESTAMP_KEY, String.valueOf(timestamp));
       return this;

polaris-core/src/main/java/org/apache/polaris/core/persistence/cache/EntityWeigher.java

@@ -35,7 +35,7 @@ public class EntityWeigher implements Weigher<Long, ResolvedPolarisEntity> {
   private static final int APPROXIMATE_ENTITY_OVERHEAD = 1000;
 
   /* Represents the amount of bytes that a character is expected to take up */
-  private static final int APPROXIMATE_BYTES_PER_CHAR = 3;
+  public static final int APPROXIMATE_BYTES_PER_CHAR = 3;
 
   /** Singleton instance */
   private static final EntityWeigher instance = new EntityWeigher();

polaris-core/src/test/java/org/apache/polaris/core/persistence/cache/InMemoryEntityCacheTest.java

@@ -94,7 +94,7 @@ public InMemoryEntityCacheTest() {
     callCtx = new PolarisCallContext(metaStore, diagServices);
     metaStoreManager = new TransactionalMetaStoreManagerImpl();
 
-    // bootstrap the mata store with our test schema
+    // bootstrap the metastore with our test schema
     tm = new PolarisTestMetaStoreManager(metaStoreManager, callCtx);
     tm.testCreateTestCatalog();
   }

polaris-core/src/test/java/org/apache/polaris/service/storage/PolarisConfigurationStoreTest.java

@@ -72,7 +72,7 @@ public void testConfigsCanBeCastedFromString() {
   }
 
   @Test
-  public void testInvalidCastThrowsException() {
+  public void testInvalidApplyThrowsException() {
     // Bool not included because Boolean.valueOf turns non-boolean strings to false
     List<PolarisConfiguration<?>> configs =
         List.of(buildConfig("int2", 12), buildConfig("long2", 34L), buildConfig("double2", 5.6D));

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

@@ -39,6 +39,7 @@
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.rest.RESTResponse;
 import org.apache.iceberg.rest.requests.CommitTransactionRequest;
 import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
@@ -48,6 +49,7 @@
 import org.apache.iceberg.rest.requests.RenameTableRequest;
 import org.apache.iceberg.rest.requests.UpdateNamespacePropertiesRequest;
 import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
 import org.apache.iceberg.view.ImmutableSQLViewRepresentation;
 import org.apache.iceberg.view.ImmutableViewVersion;
 import org.apache.polaris.core.admin.model.CreateCatalogRequest;
@@ -757,6 +759,11 @@ public void testCreateTableStagedWithWriteDelegationInsufficientPermissions() {
         });
   }
 
+  private static String getMetadataLocation(RESTResponse resp) {
+    final String metadataLocation = ((LoadTableResponse) resp).metadataLocation();
+    return metadataLocation;
+  }
+
   @Test
   public void testRegisterTableAllSufficientPrivileges() {
     Assertions.assertThat(
@@ -770,7 +777,7 @@ public void testRegisterTableAllSufficientPrivileges() {
 
     // To get a handy metadata file we can use one from another table.
     // to avoid overlapping directories, drop the original table and recreate it via registerTable
-    final String metadataLocation = newWrapper().loadTable(TABLE_NS1_1, "all").metadataLocation();
+    final String metadataLocation = getMetadataLocation(newWrapper().loadTable(TABLE_NS1_1, "all"));
     newWrapper(Set.of(PRINCIPAL_ROLE2)).dropTableWithoutPurge(TABLE_NS1_1);
 
     final RegisterTableRequest registerRequest =
@@ -808,7 +815,7 @@ public void testRegisterTableInsufficientPermissions() {
         .isTrue();
 
     // To get a handy metadata file we can use one from another table.
-    final String metadataLocation = newWrapper().loadTable(TABLE_NS1_1, "all").metadataLocation();
+    final String metadataLocation = getMetadataLocation(newWrapper().loadTable(TABLE_NS1_1, "all"));
 
     final RegisterTableRequest registerRequest =
         new RegisterTableRequest() {