Support persisting TableMetadata in the metastore

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

@@ -21,6 +21,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
+import java.util.function.Function;
 import org.apache.polaris.core.admin.model.StorageConfigInfo;
 
 public class PolarisConfiguration<T> {
@@ -30,15 +31,24 @@ public class PolarisConfiguration<T> {
   public final T defaultValue;
   private final Optional<String> catalogConfigImpl;
   private final Class<T> typ;
+  private final Optional<Function<T, Boolean>> validation;
 
   @SuppressWarnings("unchecked")
   public PolarisConfiguration(
-      String key, String description, T defaultValue, Optional<String> catalogConfig) {
+      String key,
+      String description,
+      T defaultValue,
+      Optional<String> catalogConfig,
+      Optional<Function<T, Boolean>> validation) {
     this.key = key;
     this.description = description;
     this.defaultValue = defaultValue;
     this.catalogConfigImpl = catalogConfig;
     this.typ = (Class<T>) defaultValue.getClass();
+    this.validation = validation;
+
+    // Force validation:
+    cast(defaultValue);
   }
 
   public boolean hasCatalogConfig() {
@@ -53,14 +63,27 @@ public String catalogConfig() {
   }
 
   T cast(Object value) {
-    return this.typ.cast(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 static class Builder<T> {
     private String key;
     private String description;
     private T defaultValue;
     private Optional<String> catalogConfig = Optional.empty();
+    private Optional<Function<T, Boolean>> validation = Optional.empty();
 
     public Builder<T> key(String key) {
       this.key = key;
@@ -88,11 +111,16 @@ public Builder<T> catalogConfig(String catalogConfig) {
       return this;
     }
 
+    public Builder<T> validation(Function<T, Boolean> validation) {
+      this.validation = Optional.of(validation);
+      return this;
+    }
+
     public PolarisConfiguration<T> build() {
       if (key == null || description == null || defaultValue == null) {
         throw new IllegalArgumentException("key, description, and defaultValue are required");
       }
-      return new PolarisConfiguration<>(key, description, defaultValue, catalogConfig);
+      return new PolarisConfiguration<>(key, description, defaultValue, catalogConfig, validation);
     }
   }
 
@@ -206,4 +234,18 @@ public static <T> Builder<T> builder() {
               "If set to true, allows tables to be dropped with the purge parameter set to true.")
           .defaultValue(true)
           .build();
+
+  public static final int METADATA_CACHE_MAX_BYTES_NO_CACHING = 0;
+  public static final int METADATA_CACHE_MAX_BYTES_INFINITE_CACHING = -1;
+  public static final PolarisConfiguration<Integer> METADATA_CACHE_MAX_BYTES =
+      PolarisConfiguration.<Integer>builder()
+          .key("METADATA_CACHE_MAX_BYTES")
+          .catalogConfig("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(METADATA_CACHE_MAX_BYTES_NO_CACHING)
+          .validation(value -> value >= -1)
+          .build();
 }

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

@@ -29,6 +29,15 @@ public class TableLikeEntity extends PolarisEntity {
   // 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`
+  private 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
+  private 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";
 
@@ -67,6 +76,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(
@@ -121,6 +140,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/test/java/org/apache/polaris/core/persistence/EntityCacheTest.java

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

polaris-service/src/main/java/org/apache/polaris/service/catalog/BasePolarisCatalog.java

@@ -23,6 +23,8 @@
 import com.google.common.base.Objects;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Sets;
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Arrays;
@@ -35,6 +37,7 @@
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Predicate;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.commons.lang3.exception.ExceptionUtils;
@@ -46,6 +49,7 @@
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.TableMetadataParser;
 import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.aws.s3.S3FileIOProperties;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
@@ -64,7 +68,9 @@
 import org.apache.iceberg.io.CloseableGroup;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.SupportsBulkOperations;
 import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.Tasks;
 import org.apache.iceberg.view.BaseMetastoreViewCatalog;
 import org.apache.iceberg.view.BaseViewOperations;
 import org.apache.iceberg.view.ViewBuilder;
@@ -103,6 +109,7 @@
 import org.apache.polaris.core.storage.aws.PolarisS3FileIOClientFactory;
 import org.apache.polaris.service.catalog.io.FileIOFactory;
 import org.apache.polaris.service.exception.IcebergExceptionMapper;
+import org.apache.polaris.service.persistence.MetadataCacheManager;
 import org.apache.polaris.service.task.TaskExecutor;
 import org.apache.polaris.service.types.NotificationRequest;
 import org.apache.polaris.service.types.NotificationType;
@@ -822,6 +829,34 @@ public Map<String, String> getCredentialConfig(
         storageInfo.get());
   }
 
+  public TableMetadata loadTableMetadata(TableIdentifier identifier) {
+    int maxMetadataCacheBytes =
+        callContext
+            .getPolarisCallContext()
+            .getConfigurationStore()
+            .getConfiguration(
+                callContext.getPolarisCallContext(), PolarisConfiguration.METADATA_CACHE_MAX_BYTES);
+    if (maxMetadataCacheBytes == PolarisConfiguration.METADATA_CACHE_MAX_BYTES_NO_CACHING) {
+      return loadTableMetadata(loadTable(identifier));
+    } else {
+      Supplier<TableMetadata> fallback = () -> loadTableMetadata(loadTable(identifier));
+      return MetadataCacheManager.loadTableMetadata(
+          identifier,
+          maxMetadataCacheBytes,
+          callContext.getPolarisCallContext(),
+          metaStoreManager,
+          resolvedEntityView,
+          fallback);
+    }
+  }
+
+  private static TableMetadata loadTableMetadata(Table table) {
+    if (table instanceof BaseTable baseTable) {
+      return baseTable.operations().current();
+    }
+    throw new IllegalArgumentException("Cannot load metadata for " + table.name());
+  }
+
   /**
    * Based on configuration settings, for callsites that need to handle potentially setting a new
    * base location for a TableLike entity, produces the transformed location if applicable, or else
@@ -1344,24 +1379,51 @@ public void doCommit(TableMetadata base, TableMetadata metadata) {
       TableLikeEntity entity =
           TableLikeEntity.of(resolvedEntities == null ? null : resolvedEntities.getRawLeafEntity());
       String existingLocation;
+      int maxMetadataCacheBytes =
+          callContext
+              .getPolarisCallContext()
+              .getConfigurationStore()
+              .getConfiguration(
+                  callContext.getPolarisCallContext(),
+                  catalogEntity,
+                  PolarisConfiguration.METADATA_CACHE_MAX_BYTES);
+      Optional<String> metadataJsonOpt = Optional.empty();
+      boolean shouldPersistMetadata =
+          switch (maxMetadataCacheBytes) {
+            case PolarisConfiguration.METADATA_CACHE_MAX_BYTES_NO_CACHING -> false;
+            case PolarisConfiguration.METADATA_CACHE_MAX_BYTES_INFINITE_CACHING -> {
+              metadataJsonOpt = MetadataCacheManager.toBoundedJson(metadata, maxMetadataCacheBytes);
+              yield true;
+            }
+            default -> {
+              metadataJsonOpt = MetadataCacheManager.toBoundedJson(metadata, maxMetadataCacheBytes);
+              yield metadataJsonOpt
+                  .map(String::length)
+                  .map(l -> l <= maxMetadataCacheBytes)
+                  .orElse(false);
+            }
+          };
+      final TableLikeEntity.Builder builder;
       if (null == entity) {
         existingLocation = null;
-        entity =
+        builder =
             new TableLikeEntity.Builder(tableIdentifier, newLocation)
                 .setCatalogId(getCatalogId())
                 .setSubType(PolarisEntitySubType.TABLE)
                 .setBaseLocation(metadata.location())
                 .setId(
-                    getMetaStoreManager().generateNewEntityId(getCurrentPolarisContext()).getId())
-                .build();
+                    getMetaStoreManager().generateNewEntityId(getCurrentPolarisContext()).getId());
       } else {
         existingLocation = entity.getMetadataLocation();
-        entity =
+        builder =
             new TableLikeEntity.Builder(entity)
                 .setBaseLocation(metadata.location())
-                .setMetadataLocation(newLocation)
-                .build();
+                .setMetadataLocation(newLocation);
+      }
+      if (shouldPersistMetadata && metadataJsonOpt.isPresent()) {
+        builder.setMetadataContent(newLocation, metadataJsonOpt.get());
       }
+      entity = builder.build();
       if (!Objects.equal(existingLocation, oldLocation)) {
         if (null == base) {
           throw new AlreadyExistsException("Table already exists: %s", tableName());
@@ -1383,6 +1445,98 @@ public void doCommit(TableMetadata base, TableMetadata metadata) {
       }
     }
 
+    /**
+     * Copied from {@link BaseMetastoreTableOperations} but without the requirement that base ==
+     * current()
+     *
+     * @param base table metadata on which changes were based
+     * @param metadata new table metadata with updates
+     */
+    @Override
+    public void commit(TableMetadata base, TableMetadata metadata) {
+      TableMetadata currentMetadata = current();
+
+      // if the metadata is already out of date, reject it
+      if (base == null) {
+        if (currentMetadata != null) {
+          // when current is non-null, the table exists. but when base is null, the commit is trying
+          // to create the table
+          throw new AlreadyExistsException("Table already exists: %s", tableName());
+        }
+      } else if (base.metadataFileLocation() != null
+          && !base.metadataFileLocation().equals(currentMetadata.metadataFileLocation())) {
+        throw new CommitFailedException("Cannot commit: stale table metadata");
+      } else if (base != currentMetadata) {
+        // This branch is different from BaseMetastoreTableOperations
+        LOGGER.debug(
+            "Base object differs from current metadata; proceeding because locations match");
+      } else if (base.metadataFileLocation().equals(metadata.metadataFileLocation())) {
+        // if the metadata is not changed, return early
+        LOGGER.info("Nothing to commit.");
+        return;
+      }
+
+      long start = System.currentTimeMillis();
+      doCommit(base, metadata);
+      deleteRemovedMetadataFiles(base, metadata);
+      requestRefresh();
+
+      LOGGER.info(
+          "Successfully committed to table {} in {} ms",
+          tableName(),
+          System.currentTimeMillis() - start);
+    }
+
+    /**
+     * Copied from {@link BaseMetastoreTableOperations} as the method is private there This is moved
+     * to `CatalogUtils` in Iceberg 1.7.0 and can be called from there once we depend on Iceberg
+     * 1.7.0
+     *
+     * <p>Deletes the oldest metadata files if {@link
+     * TableProperties#METADATA_DELETE_AFTER_COMMIT_ENABLED} is true.
+     *
+     * @param base table metadata on which previous versions were based
+     * @param metadata new table metadata with updated previous versions
+     */
+    protected void deleteRemovedMetadataFiles(TableMetadata base, TableMetadata metadata) {
+      if (base == null) {
+        return;
+      }
+
+      boolean deleteAfterCommit =
+          metadata.propertyAsBoolean(
+              TableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED,
+              TableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED_DEFAULT);
+
+      if (deleteAfterCommit) {
+        Set<TableMetadata.MetadataLogEntry> removedPreviousMetadataFiles =
+            Sets.newHashSet(base.previousFiles());
+        // TableMetadata#addPreviousFile builds up the metadata log and uses
+        // TableProperties.METADATA_PREVIOUS_VERSIONS_MAX to determine how many files should stay in
+        // the log, thus we don't include metadata.previousFiles() for deletion - everything else
+        // can
+        // be removed
+        removedPreviousMetadataFiles.removeAll(metadata.previousFiles());
+        if (io() instanceof SupportsBulkOperations) {
+          ((SupportsBulkOperations) io())
+              .deleteFiles(
+                  Iterables.transform(
+                      removedPreviousMetadataFiles, TableMetadata.MetadataLogEntry::file));
+        } else {
+          Tasks.foreach(removedPreviousMetadataFiles)
+              .noRetry()
+              .suppressFailureWhenFinished()
+              .onFailure(
+                  (previousMetadataFile, exc) ->
+                      LOGGER.warn(
+                          "Delete failed for previous metadata file: {}",
+                          previousMetadataFile,
+                          exc))
+              .run(previousMetadataFile -> io().deleteFile(previousMetadataFile.file()));
+        }
+      }
+    }
+
     @Override
     public FileIO io() {
       return tableFileIO;
@@ -1867,6 +2021,7 @@ private void updateTableLike(TableIdentifier identifier, PolarisEntity entity) {
       }
     }
 
+    // Drop the table:
     return getMetaStoreManager()
         .dropEntityIfExists(
             getCurrentPolarisContext(),