Core: Freshness-aware table loading in REST catalog

This is the client-side improvement for the freshness-aware table
loading in REST catalog. The main design is the following:
 - REST server can send an ETag with the LoadTableResponse
 - The client can use this ETag to populate the IF_NONE_MATCH header
   with the next loadTable request
 - The server can send a 304-NOT_MODIFIED response without a body if
   the table has not been changed based on the ETag
 - The client when receives a 304, then returns the latest table
   object associated with the ETag from cache

Author: gaborkaszab

core/src/main/java/org/apache/iceberg/rest/HTTPClient.java

@@ -326,6 +326,16 @@ protected <T extends RESTResponse> T execute(
 
       // Skip parsing the response stream for any successful request not expecting a response body
       if (emptyBody(response, responseType)) {
+        if (response.getCode() == HttpStatus.SC_NOT_MODIFIED
+            && !req.headers().contains(HttpHeaders.IF_NONE_MATCH)) {
+          // 304-NOT_MODIFIED is used for freshness-aware loading and requires an ETag sent to the
+          // server via IF_NONE_MATCH header in the request. If no ETag was sent, we shouldn't
+          // receive a 304.
+          throw new RESTException(
+              "Invalid (NOT_MODIFIED) response for request: method=%s, path=%s",
+              req.method(), req.path());
+        }
+
         return null;
       }
 

core/src/main/java/org/apache/iceberg/rest/RESTCatalog.java

@@ -39,6 +39,7 @@
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.hadoop.Configurable;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.view.View;
@@ -96,6 +97,11 @@ public void initialize(String name, Map<String, String> props) {
     sessionCatalog.initialize(name, props);
   }
 
+  @VisibleForTesting
+  RESTSessionCatalog sessionCatalog() {
+    return sessionCatalog;
+  }
+
   @Override
   public String name() {
     return sessionCatalog.name();

core/src/main/java/org/apache/iceberg/rest/RESTCatalogProperties.java

@@ -18,6 +18,8 @@
  */
 package org.apache.iceberg.rest;
 
+import java.util.concurrent.TimeUnit;
+
 public final class RESTCatalogProperties {
 
   private RESTCatalogProperties() {}
@@ -41,6 +43,14 @@ private RESTCatalogProperties() {}
   public static final String REST_SCAN_PLANNING_ENABLED = "rest-scan-planning-enabled";
   public static final boolean REST_SCAN_PLANNING_ENABLED_DEFAULT = false;
 
+  // Properties that control the behaviour of the table cache used for freshness-aware table
+  // loading.
+  public static final String TABLE_CACHE_EXPIRE_AFTER_WRITE_MS = "rest-table-cache-expire-after-ms";
+  public static final long TABLE_CACHE_EXPIRE_AFTER_WRITE_MS_DEFAULT = TimeUnit.MINUTES.toMillis(5);
+
+  public static final String TABLE_CACHE_MAX_ENTRIES = "rest-table-cache-max-entries";
+  public static final int TABLE_CACHE_MAX_ENTRIES_DEFAULT = 100;
+
   public enum SnapshotMode {
     ALL,
     REFS