Core: Add metadataFileLocation in TableUtil

core/src/main/java/org/apache/iceberg/SerializableTable.java

@@ -104,6 +104,14 @@ public static Table copyOf(Table table) {
     }
   }
 
+  public String metadataFileLocation() {
+    if (metadataFileLocation == null) {
+      throw new UnsupportedOperationException(
+          this.getClass().getName() + " does not have a metadata file location");
+    }
+    return metadataFileLocation;
+  }
+
   private String metadataFileLocation(Table table) {
     if (table instanceof HasTableOperations) {
       TableOperations ops = ((HasTableOperations) table).operations();

core/src/main/java/org/apache/iceberg/TableUtil.java

@@ -38,4 +38,23 @@ public static int formatVersion(Table table) {
           String.format("%s does not have a format version", table.getClass().getSimpleName()));
     }
   }
+
+  /** Returns the metadata file location of the given table */
+  public static String metadataFileLocation(Table table) {
+    Preconditions.checkArgument(null != table, "Invalid table: null");
+
+    if (table instanceof SerializableTable) {
+      SerializableTable serializableTable = (SerializableTable) table;
+      return serializableTable.metadataFileLocation();
+    } else if (table instanceof HasTableOperations) {
+      HasTableOperations ops = (HasTableOperations) table;
+      return ops.operations().current().metadataFileLocation();
+    } else if (table instanceof BaseMetadataTable) {
+      return ((BaseMetadataTable) table).table().operations().current().metadataFileLocation();
+    } else {
+      throw new IllegalArgumentException(
+          String.format(
+              "%s does not have a metadata file location", table.getClass().getSimpleName()));
+    }
+  }
 }

core/src/test/java/org/apache/iceberg/TestTableUtil.java

@@ -31,11 +31,14 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
 import org.junit.jupiter.params.provider.ValueSource;
 
 public class TestTableUtil {
   private static final Namespace NS = Namespace.of("ns");
   private static final TableIdentifier IDENTIFIER = TableIdentifier.of(NS, "test");
+  private static final Schema SCHEMA =
+      new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get()));
 
   @TempDir private File tmp;
 
@@ -53,6 +56,10 @@ public void nullTable() {
     assertThatThrownBy(() -> TableUtil.formatVersion(null))
         .isInstanceOf(IllegalArgumentException.class)
         .hasMessage("Invalid table: null");
+
+    assertThatThrownBy(() -> TableUtil.metadataFileLocation(null))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Invalid table: null");
   }
 
   @ParameterizedTest
@@ -61,32 +68,51 @@ public void formatVersionForBaseTable(int formatVersion) {
     Table table =
         catalog.createTable(
             IDENTIFIER,
-            new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())),
+            SCHEMA,
             PartitionSpec.unpartitioned(),
             ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion)));
 
     assertThat(TableUtil.formatVersion(table)).isEqualTo(formatVersion);
     assertThat(TableUtil.formatVersion(SerializableTable.copyOf(table))).isEqualTo(formatVersion);
   }
 
+  @ParameterizedTest
+  @EnumSource(MetadataTableType.class)
+  public void formatVersionForMetadataTables(MetadataTableType type) {
+    Table table = catalog.createTable(IDENTIFIER, SCHEMA);
+
+    Table metadataTable = MetadataTableUtils.createMetadataTableInstance(table, type);
+    assertThatThrownBy(() -> TableUtil.formatVersion(metadataTable))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("%s does not have a format version", metadataTable.getClass().getSimpleName());
+
+    assertThatThrownBy(() -> TableUtil.formatVersion(SerializableTable.copyOf(metadataTable)))
+        .isInstanceOf(UnsupportedOperationException.class)
+        .hasMessage(
+            "%s does not have a format version",
+            SerializableTable.SerializableMetadataTable.class.getName());
+  }
+
   @Test
-  public void formatVersionForMetadataTables() {
-    Table table =
-        catalog.createTable(
-            IDENTIFIER, new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())));
-
-    for (MetadataTableType type : MetadataTableType.values()) {
-      Table metadataTable = MetadataTableUtils.createMetadataTableInstance(table, type);
-      assertThatThrownBy(() -> TableUtil.formatVersion(metadataTable))
-          .isInstanceOf(IllegalArgumentException.class)
-          .hasMessage(
-              "%s does not have a format version", metadataTable.getClass().getSimpleName());
-
-      assertThatThrownBy(() -> TableUtil.formatVersion(SerializableTable.copyOf(metadataTable)))
-          .isInstanceOf(UnsupportedOperationException.class)
-          .hasMessage(
-              "%s does not have a format version",
-              SerializableTable.SerializableMetadataTable.class.getName());
-    }
+  public void metadataFileLocationForBaseTable() {
+    Table table = catalog.createTable(IDENTIFIER, SCHEMA);
+
+    TableMetadata metadata = ((HasTableOperations) table).operations().current();
+    assertThat(TableUtil.metadataFileLocation(table)).isEqualTo(metadata.metadataFileLocation());
+    assertThat(TableUtil.metadataFileLocation(SerializableTable.copyOf(table)))
+        .isEqualTo(metadata.metadataFileLocation());
+  }
+
+  @ParameterizedTest
+  @EnumSource(MetadataTableType.class)
+  public void metadataFileLocationForMetadataTables(MetadataTableType type) {
+    Table table = catalog.createTable(IDENTIFIER, SCHEMA);
+
+    Table metadataTable = MetadataTableUtils.createMetadataTableInstance(table, type);
+    TableMetadata metadata = ((HasTableOperations) table).operations().current();
+    assertThat(TableUtil.metadataFileLocation(metadataTable))
+        .isEqualTo(metadata.metadataFileLocation());
+    assertThat(TableUtil.metadataFileLocation(SerializableTable.copyOf(metadataTable)))
+        .isEqualTo(metadata.metadataFileLocation());
   }
 }

core/src/test/java/org/apache/iceberg/hadoop/TestTableSerialization.java

@@ -49,6 +49,7 @@
 import org.apache.iceberg.types.Types;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
 import org.junit.jupiter.params.provider.ValueSource;
 
 public class TestTableSerialization extends HadoopTableTestBase {
@@ -69,6 +70,8 @@ public void testSerializableTable() throws IOException, ClassNotFoundException {
     assertThat(((HasTableOperations) serializableTable).operations())
         .isInstanceOf(StaticTableOperations.class);
     assertThat(TableUtil.formatVersion(serializableTable)).isEqualTo(2);
+    assertThat(TableUtil.metadataFileLocation(serializableTable))
+        .isEqualTo(TableUtil.metadataFileLocation(table));
   }
 
   @Test
@@ -96,22 +99,24 @@ public void testSerializableTxnTable() throws IOException, ClassNotFoundExceptio
     TestHelpers.assertSerializedMetadata(txn.table(), TestHelpers.roundTripSerialize(txn.table()));
   }
 
-  @Test
-  public void testSerializableMetadataTable() throws IOException, ClassNotFoundException {
-    for (MetadataTableType type : MetadataTableType.values()) {
-      Table metadataTable = getMetaDataTable(table, type);
-      TestHelpers.assertSerializedAndLoadedMetadata(
-          metadataTable, TestHelpers.roundTripSerialize(metadataTable));
-      Table serializableTable = SerializableTable.copyOf(metadataTable);
-      TestHelpers.assertSerializedAndLoadedMetadata(
-          serializableTable, TestHelpers.KryoHelpers.roundTripSerialize(serializableTable));
-      assertThatThrownBy(() -> ((HasTableOperations) serializableTable).operations())
-          .isInstanceOf(UnsupportedOperationException.class)
-          .hasMessageEndingWith("does not support operations()");
-      assertThatThrownBy(() -> TableUtil.formatVersion(serializableTable))
-          .isInstanceOf(UnsupportedOperationException.class)
-          .hasMessageEndingWith("does not have a format version");
-    }
+  @ParameterizedTest
+  @EnumSource(MetadataTableType.class)
+  public void testSerializableMetadataTable(MetadataTableType type)
+      throws IOException, ClassNotFoundException {
+    Table metadataTable = getMetaDataTable(table, type);
+    TestHelpers.assertSerializedAndLoadedMetadata(
+        metadataTable, TestHelpers.roundTripSerialize(metadataTable));
+    Table serializableTable = SerializableTable.copyOf(metadataTable);
+    TestHelpers.assertSerializedAndLoadedMetadata(
+        serializableTable, TestHelpers.KryoHelpers.roundTripSerialize(serializableTable));
+    assertThatThrownBy(() -> ((HasTableOperations) serializableTable).operations())
+        .isInstanceOf(UnsupportedOperationException.class)
+        .hasMessageEndingWith("does not support operations()");
+    assertThatThrownBy(() -> TableUtil.formatVersion(serializableTable))
+        .isInstanceOf(UnsupportedOperationException.class)
+        .hasMessageEndingWith("does not have a format version");
+    assertThat(TableUtil.metadataFileLocation(serializableTable))
+        .isEqualTo(TableUtil.metadataFileLocation(table));
   }
 
   @Test

spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRegisterTableProcedure.java

@@ -20,8 +20,8 @@
 
 import java.util.List;
 import java.util.Map;
-import org.apache.iceberg.HasTableOperations;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.TableUtil;
 import org.apache.iceberg.spark.Spark3Util;
 import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
 import org.apache.spark.sql.catalyst.parser.ParseException;
@@ -65,8 +65,7 @@ public void testRegisterTable() throws NoSuchTableException, ParseException {
     Table table = Spark3Util.loadIcebergTable(spark, tableName);
     long originalFileCount = (long) scalarSql("SELECT COUNT(*) from %s.files", tableName);
     long currentSnapshotId = table.currentSnapshot().snapshotId();
-    String metadataJson =
-        (((HasTableOperations) table).operations()).current().metadataFileLocation();
+    String metadataJson = TableUtil.metadataFileLocation(table);
 
     List<Object[]> result =
         sql("CALL %s.system.register_table('%s', '%s')", catalogName, targetName, metadataJson);

spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRegisterTableProcedure.java

@@ -20,8 +20,8 @@
 
 import java.util.List;
 import java.util.Map;
-import org.apache.iceberg.HasTableOperations;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.TableUtil;
 import org.apache.iceberg.spark.Spark3Util;
 import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
 import org.apache.spark.sql.catalyst.parser.ParseException;
@@ -65,8 +65,7 @@ public void testRegisterTable() throws NoSuchTableException, ParseException {
     Table table = Spark3Util.loadIcebergTable(spark, tableName);
     long originalFileCount = (long) scalarSql("SELECT COUNT(*) from %s.files", tableName);
     long currentSnapshotId = table.currentSnapshot().snapshotId();
-    String metadataJson =
-        (((HasTableOperations) table).operations()).current().metadataFileLocation();
+    String metadataJson = TableUtil.metadataFileLocation(table);
 
     List<Object[]> result =
         sql("CALL %s.system.register_table('%s', '%s')", catalogName, targetName, metadataJson);

spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRegisterTableProcedure.java

@@ -21,9 +21,9 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.util.List;
-import org.apache.iceberg.HasTableOperations;
 import org.apache.iceberg.ParameterizedTestExtension;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.TableUtil;
 import org.apache.iceberg.spark.Spark3Util;
 import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
 import org.apache.spark.sql.catalyst.parser.ParseException;
@@ -64,8 +64,7 @@ public void testRegisterTable() throws NoSuchTableException, ParseException {
     Table table = Spark3Util.loadIcebergTable(spark, tableName);
     long originalFileCount = (long) scalarSql("SELECT COUNT(*) from %s.files", tableName);
     long currentSnapshotId = table.currentSnapshot().snapshotId();
-    String metadataJson =
-        (((HasTableOperations) table).operations()).current().metadataFileLocation();
+    String metadataJson = TableUtil.metadataFileLocation(table);
 
     List<Object[]> result =
         sql("CALL %s.system.register_table('%s', '%s')", catalogName, targetName, metadataJson);

spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteTablePathProcedure.java

@@ -26,6 +26,7 @@
 import org.apache.iceberg.HasTableOperations;
 import org.apache.iceberg.RewriteTablePathUtil;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.TableUtil;
 import org.apache.spark.sql.AnalysisException;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -50,8 +51,7 @@ public void removeTables() {
   public void testRewriteTablePathWithPositionalArgument() {
     String location = targetTableDir.toFile().toURI().toString();
     Table table = validationCatalog.loadTable(tableIdent);
-    String metadataJson =
-        (((HasTableOperations) table).operations()).current().metadataFileLocation();
+    String metadataJson = TableUtil.metadataFileLocation(table);
 
     List<Object[]> result =
         sql(
@@ -72,9 +72,7 @@ public void testRewriteTablePathWithPositionalArgument() {
   @TestTemplate
   public void testRewriteTablePathWithNamedArgument() {
     Table table = validationCatalog.loadTable(tableIdent);
-    String v0Metadata =
-        RewriteTablePathUtil.fileName(
-            (((HasTableOperations) table).operations()).current().metadataFileLocation());
+    String v0Metadata = RewriteTablePathUtil.fileName(TableUtil.metadataFileLocation(table));
     sql("INSERT INTO TABLE %s VALUES (1, 'a')", tableName);
     String v1Metadata =
         RewriteTablePathUtil.fileName(
@@ -132,9 +130,7 @@ public void testProcedureWithInvalidInput() {
         .hasMessageContaining("Couldn't load table");
 
     Table table = validationCatalog.loadTable(tableIdent);
-    String v0Metadata =
-        RewriteTablePathUtil.fileName(
-            (((HasTableOperations) table).operations()).current().metadataFileLocation());
+    String v0Metadata = RewriteTablePathUtil.fileName(TableUtil.metadataFileLocation(table));
     assertThatThrownBy(
             () ->
                 sql(