[#9736][followup] feat(iceberg-rest): Skip full table load on ETag match via SupportsMetadataLocation

iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java

@@ -204,6 +204,23 @@ public LoadTableResponse loadTable(TableIdentifier tableIdentifier) {
     return loadTableResponse;
   }
 
+  /**
+   * Retrieves the metadata file location for the specified table without loading full table
+   * metadata. This is an optional fast path for catalogs that implement {@link
+   * SupportsMetadataLocation}.
+   *
+   * @param tableIdentifier the table identifier
+   * @return an Optional containing the metadata file location, or empty if the catalog doesn't
+   *     support this operation
+   */
+  public Optional<String> getTableMetadataLocation(TableIdentifier tableIdentifier) {
+    if (catalog instanceof SupportsMetadataLocation) {
+      return Optional.ofNullable(
+          ((SupportsMetadataLocation) catalog).metadataLocation(tableIdentifier));
+    }
+    return Optional.empty();
+  }
+
   public boolean tableExists(TableIdentifier tableIdentifier) {
     return catalog.tableExists(tableIdentifier);
   }

iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableEventDispatcher.java

@@ -300,4 +300,10 @@ public PlanTableScanResponse planTableScan(
     eventBus.dispatchEvent(new IcebergPlanTableScanEvent(context, gravitinoNameIdentifier));
     return planTableScanResponse;
   }
+
+  @Override
+  public Optional<String> getTableMetadataLocation(
+      IcebergRequestContext context, TableIdentifier tableIdentifier) {
+    return icebergTableOperationDispatcher.getTableMetadataLocation(context, tableIdentifier);
+  }
 }

iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableHookDispatcher.java

@@ -20,6 +20,7 @@
 
 import java.io.IOException;
 import java.time.Instant;
+import java.util.Optional;
 import org.apache.gravitino.Entity;
 import org.apache.gravitino.EntityStore;
 import org.apache.gravitino.GravitinoEnv;
@@ -179,6 +180,12 @@ public PlanTableScanResponse planTableScan(
     return dispatcher.planTableScan(context, tableIdentifier, scanRequest);
   }
 
+  @Override
+  public Optional<String> getTableMetadataLocation(
+      IcebergRequestContext context, TableIdentifier tableIdentifier) {
+    return dispatcher.getTableMetadataLocation(context, tableIdentifier);
+  }
+
   private void importTable(String catalogName, Namespace namespace, String tableName) {
     TableDispatcher tableDispatcher = GravitinoEnv.getInstance().tableDispatcher();
     if (tableDispatcher != null) {

iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationDispatcher.java

@@ -19,6 +19,7 @@
 
 package org.apache.gravitino.iceberg.service.dispatcher;
 
+import java.util.Optional;
 import org.apache.gravitino.listener.api.event.IcebergRequestContext;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
@@ -133,4 +134,16 @@ PlanTableScanResponse planTableScan(
       IcebergRequestContext context,
       TableIdentifier tableIdentifier,
       PlanTableScanRequest scanRequest);
+
+  /**
+   * Retrieves the metadata file location for a table without loading full table metadata. This is
+   * an optional fast path for catalogs that support cheap metadata location retrieval.
+   *
+   * @param context Iceberg REST request context information.
+   * @param tableIdentifier The Iceberg table identifier.
+   * @return an Optional containing the metadata file location, or empty if the catalog doesn't
+   *     support this operation
+   */
+  Optional<String> getTableMetadataLocation(
+      IcebergRequestContext context, TableIdentifier tableIdentifier);
 }

iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java

@@ -21,6 +21,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 import org.apache.gravitino.Entity;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.auth.AuthConstants;
@@ -186,4 +187,12 @@ public PlanTableScanResponse planTableScan(
         .getCatalogWrapper(context.catalogName())
         .planTableScan(tableIdentifier, scanRequest);
   }
+
+  @Override
+  public Optional<String> getTableMetadataLocation(
+      IcebergRequestContext context, TableIdentifier tableIdentifier) {
+    return icebergCatalogWrapperManager
+        .getCatalogWrapper(context.catalogName())
+        .getTableMetadataLocation(tableIdentifier);
+  }
 }

iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java

@@ -28,6 +28,7 @@
 import java.security.NoSuchAlgorithmException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 import javax.inject.Inject;
 import javax.servlet.http.HttpServletRequest;
 import javax.ws.rs.Consumes;
@@ -92,6 +93,8 @@ public class IcebergTableOperations {
 
   @VisibleForTesting public static final String IF_NONE_MATCH = "If-None-Match";
 
+  @VisibleForTesting static final String DEFAULT_SNAPSHOTS = "all";
+
   private IcebergMetricsManager icebergMetricsManager;
 
   private ObjectMapper icebergObjectMapper;
@@ -288,7 +291,7 @@ public Response loadTable(
           String namespace,
       @IcebergAuthorizationMetadata(type = RequestType.LOAD_TABLE) @Encoded() @PathParam("table")
           String table,
-      @DefaultValue("all") @QueryParam("snapshots") String snapshots,
+      @DefaultValue(DEFAULT_SNAPSHOTS) @QueryParam("snapshots") String snapshots,
       @HeaderParam(X_ICEBERG_ACCESS_DELEGATION) String accessDelegation,
       @HeaderParam(IF_NONE_MATCH) String ifNoneMatch) {
     String catalogName = IcebergRESTUtils.getCatalogName(prefix);
@@ -311,6 +314,19 @@ public Response loadTable(
             TableIdentifier tableIdentifier = TableIdentifier.of(icebergNS, tableName);
             IcebergRequestContext context =
                 new IcebergRequestContext(httpServletRequest(), catalogName, isCredentialVending);
+
+            // Fast path: if client sent If-None-Match, try to resolve ETag without full table load
+            if (StringUtils.isNotBlank(ifNoneMatch)) {
+              Optional<String> metadataLocation =
+                  tableOperationDispatcher.getTableMetadataLocation(context, tableIdentifier);
+              if (metadataLocation.isPresent()) {
+                EntityTag etag = generateETag(metadataLocation.get(), snapshots);
+                if (etag != null && etagMatches(ifNoneMatch, etag)) {
+                  return Response.notModified(etag).build();
+                }
+              }
+            }
+
             LoadTableResponse loadTableResponse =
                 tableOperationDispatcher.loadTable(context, tableIdentifier);
             EntityTag etag =
@@ -521,13 +537,16 @@ public Response planTableScan(
   }
 
   /**
-   * Builds an OK response with the ETag header derived from the table metadata location.
+   * Builds an OK response with the ETag header derived from the table metadata location. Uses the
+   * default snapshots value to ensure ETags from create/update are consistent with the default
+   * loadTable endpoint.
    *
    * @param loadTableResponse the table response to include in the body
    * @return a Response with ETag header set
    */
   private static Response buildResponseWithETag(LoadTableResponse loadTableResponse) {
-    EntityTag etag = generateETag(loadTableResponse.tableMetadata().metadataFileLocation());
+    EntityTag etag =
+        generateETag(loadTableResponse.tableMetadata().metadataFileLocation(), DEFAULT_SNAPSHOTS);
     return buildResponseWithETag(loadTableResponse, etag);
   }
 
@@ -605,7 +624,7 @@ static EntityTag generateETag(String metadataLocation, String snapshots) {
    * @return true if the ETag matches (table unchanged), false otherwise
    */
   private static boolean etagMatches(String ifNoneMatch, EntityTag etag) {
-    if (ifNoneMatch == null || ifNoneMatch.isEmpty()) {
+    if (StringUtils.isBlank(ifNoneMatch)) {
       return false;
     }
     // Strip quotes if present to compare the raw value

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergTableOperations.java

@@ -771,6 +771,65 @@ void testCreateTableReturnsETag(Namespace namespace) {
     Assertions.assertFalse(etag.isEmpty(), "ETag header should not be empty");
   }
 
+  @ParameterizedTest
+  @MethodSource("org.apache.gravitino.iceberg.service.rest.IcebergRestTestUtil#testNamespaces")
+  void testCreateTableETagMatchesLoadTableETag(Namespace namespace) {
+    verifyCreateNamespaceSucc(namespace);
+    Response createResponse = doCreateTable(namespace, "create_load_etag_foo1");
+    Assertions.assertEquals(Status.OK.getStatusCode(), createResponse.getStatus());
+    String createEtag = createResponse.getHeaderString("ETag");
+    Assertions.assertNotNull(createEtag, "ETag should be present in create response");
+
+    // Load the same table with default snapshots — ETag should match
+    Response loadResponse = doLoadTable(namespace, "create_load_etag_foo1");
+    Assertions.assertEquals(Status.OK.getStatusCode(), loadResponse.getStatus());
+    String loadEtag = loadResponse.getHeaderString("ETag");
+    Assertions.assertNotNull(loadEtag, "ETag should be present in load response");
+
+    Assertions.assertEquals(
+        createEtag, loadEtag, "ETag from createTable should match ETag from default loadTable");
+
+    // The create ETag should be reusable for If-None-Match on a subsequent loadTable
+    Response conditionalResponse =
+        getTableClientBuilder(namespace, Optional.of("create_load_etag_foo1"))
+            .header(IcebergTableOperations.IF_NONE_MATCH, createEtag)
+            .get();
+    Assertions.assertEquals(
+        Status.NOT_MODIFIED.getStatusCode(),
+        conditionalResponse.getStatus(),
+        "Create ETag should produce 304 on subsequent unchanged loadTable");
+  }
+
+  @ParameterizedTest
+  @MethodSource("org.apache.gravitino.iceberg.service.rest.IcebergRestTestUtil#testNamespaces")
+  void testUpdateTableETagMatchesLoadTableETag(Namespace namespace) {
+    verifyCreateNamespaceSucc(namespace);
+    verifyCreateTableSucc(namespace, "update_load_etag_foo1");
+    TableMetadata metadata = getTableMeta(namespace, "update_load_etag_foo1");
+    Response updateResponse = doUpdateTable(namespace, "update_load_etag_foo1", metadata);
+    Assertions.assertEquals(Status.OK.getStatusCode(), updateResponse.getStatus());
+    String updateEtag = updateResponse.getHeaderString("ETag");
+    Assertions.assertNotNull(updateEtag, "ETag should be present in update response");
+
+    // Load the same table — ETag should match
+    Response loadResponse = doLoadTable(namespace, "update_load_etag_foo1");
+    Assertions.assertEquals(Status.OK.getStatusCode(), loadResponse.getStatus());
+    String loadEtag = loadResponse.getHeaderString("ETag");
+
+    Assertions.assertEquals(
+        updateEtag, loadEtag, "ETag from updateTable should match ETag from default loadTable");
+
+    // The update ETag should be reusable for If-None-Match
+    Response conditionalResponse =
+        getTableClientBuilder(namespace, Optional.of("update_load_etag_foo1"))
+            .header(IcebergTableOperations.IF_NONE_MATCH, updateEtag)
+            .get();
+    Assertions.assertEquals(
+        Status.NOT_MODIFIED.getStatusCode(),
+        conditionalResponse.getStatus(),
+        "Update ETag should produce 304 on subsequent unchanged loadTable");
+  }
+
   @ParameterizedTest
   @MethodSource("org.apache.gravitino.iceberg.service.rest.IcebergRestTestUtil#testNamespaces")
   void testUpdateTableReturnsETag(Namespace namespace) {