Added freshness aware table loading using metadata file location for ETag

service/common/src/main/java/org/apache/polaris/service/catalog/IcebergCatalogAdapter.java

@@ -26,6 +26,8 @@
 import com.google.common.collect.ImmutableSet;
 import jakarta.enterprise.context.RequestScoped;
 import jakarta.inject.Inject;
+import jakarta.ws.rs.WebApplicationException;
+import jakarta.ws.rs.core.HttpHeaders;
 import jakarta.ws.rs.core.Response;
 import jakarta.ws.rs.core.Response.Status;
 import jakarta.ws.rs.core.SecurityContext;
@@ -54,6 +56,7 @@
 import org.apache.iceberg.rest.requests.ReportMetricsRequest;
 import org.apache.iceberg.rest.requests.UpdateNamespacePropertiesRequest;
 import org.apache.iceberg.rest.responses.ConfigResponse;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
 import org.apache.polaris.core.PolarisConfigurationStore;
 import org.apache.polaris.core.PolarisDiagnostics;
 import org.apache.polaris.core.auth.AuthenticatedPolarisPrincipal;
@@ -308,9 +311,11 @@ public Response createTable(
                   .build();
             }
           } else if (delegationModes.isEmpty()) {
-            return Response.ok(catalog.createTableDirect(ns, createTableRequest)).build();
+            LoadTableResponse loadTableResponse = catalog.createTableDirect(ns, createTableRequest);
+            return Response.ok(loadTableResponse).header(HttpHeaders.ETAG, loadTableResponse.metadataLocation()).build();
           } else {
-            return Response.ok(catalog.createTableDirectWithWriteDelegation(ns, createTableRequest))
+            LoadTableResponse loadTableResponse = catalog.createTableDirectWithWriteDelegation(ns, createTableRequest);
+            return Response.ok(loadTableResponse).header(HttpHeaders.ETAG, loadTableResponse.metadataLocation())
                 .build();
           }
         });
@@ -335,6 +340,7 @@ public Response loadTable(
       String namespace,
       String table,
       String accessDelegationMode,
+      String etag,
       String snapshots,
       RealmContext realmContext,
       SecurityContext securityContext) {
@@ -347,9 +353,12 @@ public Response loadTable(
         prefix,
         catalog -> {
           if (delegationModes.isEmpty()) {
-            return Response.ok(catalog.loadTable(tableIdentifier, snapshots)).build();
+            Optional<LoadTableResponse> optionalLoadTableResponse = catalog.loadTableFreshnessAware(tableIdentifier, etag, snapshots);
+            LoadTableResponse loadTableResponse = optionalLoadTableResponse.orElseThrow(() -> new WebApplicationException(Status.NOT_MODIFIED));
+            return Response.ok(loadTableResponse).header(HttpHeaders.ETAG, loadTableResponse.metadataLocation()).build();
           } else {
-            return Response.ok(catalog.loadTableWithAccessDelegation(tableIdentifier, snapshots))
+            LoadTableResponse loadTableResponse = catalog.loadTableWithAccessDelegation(tableIdentifier, snapshots);
+            return Response.ok(loadTableResponse).header(HttpHeaders.ETAG, loadTableResponse.metadataLocation())
                 .build();
           }
         });
@@ -407,7 +416,10 @@ public Response registerTable(
     return withCatalog(
         securityContext,
         prefix,
-        catalog -> Response.ok(catalog.registerTable(ns, registerTableRequest)).build());
+        catalog -> {
+          LoadTableResponse loadTableResponse = catalog.registerTable(ns, registerTableRequest);
+          return Response.ok(loadTableResponse).header(HttpHeaders.ETAG, loadTableResponse.metadataLocation()).build();
+          });
   }
 
   @Override

service/common/src/main/java/org/apache/polaris/service/catalog/PolarisCatalogHandlerWrapper.java

@@ -81,6 +81,7 @@
 import org.apache.polaris.core.entity.CatalogEntity;
 import org.apache.polaris.core.entity.PolarisEntitySubType;
 import org.apache.polaris.core.entity.PolarisEntityType;
+import org.apache.polaris.core.entity.TableLikeEntity;
 import org.apache.polaris.core.persistence.PolarisEntityManager;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.apache.polaris.core.persistence.PolarisMetaStoreSession;
@@ -786,15 +787,44 @@ public boolean sendNotification(TableIdentifier identifier, NotificationRequest
         && notificationCatalog.sendNotification(identifier, request);
   }
 
+  private boolean etagMatchesCurrentMetadataLocation(String etag, TableIdentifier tableIdentifier) {
+    if (etag != null) {
+      PolarisResolvedPathWrapper target = resolutionManifest
+              .getResolvedPath(tableIdentifier, PolarisEntitySubType.TABLE, true);
+
+      String currentEntityMetadataVersion = target
+              .getRawLeafEntity()
+              .getInternalPropertiesAsMap()
+              .get(TableLikeEntity.METADATA_LOCATION_KEY);
+
+      if (etag.equals(currentEntityMetadataVersion)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
   public LoadTableResponse loadTable(TableIdentifier tableIdentifier, String snapshots) {
-    PolarisAuthorizableOperation op = PolarisAuthorizableOperation.LOAD_TABLE;
-    authorizeBasicTableLikeOperationOrThrow(op, PolarisEntitySubType.TABLE, tableIdentifier);
+    return loadTableFreshnessAware(tableIdentifier, null, snapshots).get();
+  }
 
-    return CatalogHandlers.loadTable(baseCatalog, tableIdentifier);
+  public Optional<LoadTableResponse> loadTableFreshnessAware(TableIdentifier tableIdentifier, String etag, String snapshots) {
+      PolarisAuthorizableOperation op = PolarisAuthorizableOperation.LOAD_TABLE;
+      authorizeBasicTableLikeOperationOrThrow(op, PolarisEntitySubType.TABLE, tableIdentifier);
+
+      if (etagMatchesCurrentMetadataLocation(etag, tableIdentifier)) {
+        return Optional.empty();
+      }
+
+      return Optional.of(CatalogHandlers.loadTable(baseCatalog, tableIdentifier));
+  }
+
+  public LoadTableResponse loadTableWithAccessDelegation(TableIdentifier tableIdentifier, String snapshots) {
+    return loadTableWithAccessDelegationFreshnessAware(tableIdentifier, null, snapshots).get();
   }
 
-  public LoadTableResponse loadTableWithAccessDelegation(
-      TableIdentifier tableIdentifier, String snapshots) {
+  public Optional<LoadTableResponse> loadTableWithAccessDelegationFreshnessAware(
+      TableIdentifier tableIdentifier, String etag, String snapshots) {
     // Here we have a single method that falls through multiple candidate
     // PolarisAuthorizableOperations because instead of identifying the desired operation up-front
     // and
@@ -832,6 +862,10 @@ public LoadTableResponse loadTableWithAccessDelegation(
           PolarisConfiguration.ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING.catalogConfig());
     }
 
+    if (etagMatchesCurrentMetadataLocation(etag, tableIdentifier)) {
+      return Optional.empty();
+    }
+
     // TODO: Find a way for the configuration or caller to better express whether to fail or omit
     // when data-access is specified but access delegation grants are not found.
     Table table = baseCatalog.loadTable(tableIdentifier);
@@ -850,7 +884,7 @@ public LoadTableResponse loadTableWithAccessDelegation(
             credentialDelegation.getCredentialConfig(
                 tableIdentifier, tableMetadata, actionsRequested));
       }
-      return responseBuilder.build();
+      return Optional.of(responseBuilder.build());
     } else if (table instanceof BaseMetadataTable) {
       // metadata tables are loaded on the client side, return NoSuchTableException for now
       throw new NoSuchTableException("Table does not exist: %s", tableIdentifier.toString());

spec/rest-catalog-open-api.yaml

@@ -683,6 +683,15 @@ paths:
         key. For example, "urn:ietf:params:oauth:token-type:jwt=<JWT-token>".
       parameters:
         - $ref: '#/components/parameters/data-access'
+        - name: If-None-Match
+          in: header
+          description:
+            An optional header that allows the server to return 304 (Not Modified) if the metadata
+            is current. The content is the value of the ETag received in a CreateTableResponse or
+            LoadTableResponse.
+          required: false
+          schema:
+            type: string
         - in: query
           name: snapshots
           description:
@@ -698,6 +707,10 @@ paths:
       responses:
         200:
           $ref: '#/components/responses/LoadTableResponse'
+        304:
+          description:
+            Not Modified - Based on the content of the 'If-None-Match' header the table metadata has
+            not changed since.
         400:
           $ref: '#/components/responses/BadRequestErrorResponse'
         401:
@@ -1688,6 +1701,14 @@ components:
         type: integer
         minimum: 1
 
+    etag:
+      name: ETag
+      in: header
+      description: Identifies a unique version of the table metadata.
+      required: false
+      schema:
+        type: string
+
   ##############################
   # Application Schema Objects #
   ##############################
@@ -4193,13 +4214,19 @@ components:
         application/json:
           schema:
             $ref: '#/components/schemas/LoadTableResult'
+      headers:
+        etag:
+          $ref: '#/components/parameters/etag'
 
     LoadTableResponse:
       description: Table metadata result when loading a table
       content:
         application/json:
           schema:
             $ref: '#/components/schemas/LoadTableResult'
+      headers:
+        etag:
+          $ref: '#/components/parameters/etag'
 
     LoadViewResponse:
       description: View metadata result when loading a view