Core: Add SerializableTable

api/src/test/java/org/apache/iceberg/TestHelpers.java

@@ -102,6 +102,24 @@ public static void assertSameSchemaList(List<Schema> list1, List<Schema> list2)
     );
   }
 
+  public static void assertSerializedMetadata(Table expected, Table actual) {
+    Assert.assertEquals("Name must match", expected.name(), actual.name());
+    Assert.assertEquals("Location must match", expected.location(), actual.location());
+    Assert.assertEquals("Props must match", expected.properties(), actual.properties());
+    Assert.assertEquals("Schema must match", expected.schema().asStruct(), actual.schema().asStruct());
+    Assert.assertEquals("Spec must match", expected.spec(), actual.spec());
+    Assert.assertEquals("Sort order must match", expected.sortOrder(), actual.sortOrder());
+  }
+
+  public static void assertSerializedAndLoadedMetadata(Table expected, Table actual) {
+    assertSerializedMetadata(expected, actual);
+    Assert.assertEquals("Specs must match", expected.specs(), actual.specs());
+    Assert.assertEquals("Sort orders must match", expected.sortOrders(), actual.sortOrders());
+    Assert.assertEquals("Current snapshot must match", expected.currentSnapshot(), actual.currentSnapshot());
+    Assert.assertEquals("Snapshots must match", expected.snapshots(), actual.snapshots());
+    Assert.assertEquals("History must match", expected.history(), actual.history());
+  }
+
   private static class CheckReferencesBound extends ExpressionVisitors.ExpressionVisitor<Void> {
     private final String message;
 

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

@@ -219,37 +219,6 @@ public String toString() {
   }
 
   final Object writeReplace() {
-    String metadataLocation = ops.current().metadataFileLocation();
-    return new TableProxy(io(), table().name(), name(), metadataLocation, metadataTableType(), locationProvider());
-  }
-
-  static class TableProxy implements Serializable {
-    private FileIO io;
-    private String baseTableName;
-    private String metadataTableName;
-    private String metadataLocation;
-    private MetadataTableType type;
-    private LocationProvider locationProvider;
-
-    TableProxy(FileIO io, String baseTableName, String metadataTableName, String metadataLocation,
-               MetadataTableType type, LocationProvider locationProvider) {
-      this.io = io;
-      this.baseTableName = baseTableName;
-      this.metadataTableName = metadataTableName;
-      this.metadataLocation = metadataLocation;
-      this.type = type;
-      this.locationProvider = locationProvider;
-    }
-
-    /**
-     * Returns a table with {@link StaticTableOperations} so after deserialization no Catalog related calls are
-     * needed for accessing the table snapshot data.
-     * @return The metadata Table object for reading the table data at the time of the serialization of the original
-     *         object
-     */
-    private Object readResolve()  {
-      TableOperations ops = new StaticTableOperations(metadataLocation, io, locationProvider);
-      return MetadataTableUtils.createMetadataTableInstance(ops, baseTableName, metadataTableName, type);
-    }
+    return SerializableTableFactory.copyOf(this);
   }
 }

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

@@ -225,30 +225,6 @@ public String toString() {
   }
 
   Object writeReplace() {
-    return new TableProxy(this);
-  }
-
-  private static class TableProxy implements Serializable {
-    private FileIO io;
-    private String name;
-    private String metadataLocation;
-    private LocationProvider locationProvider;
-
-    private TableProxy(BaseTable table) {
-      io = table.io();
-      name = table.name();
-      metadataLocation = table.operations().current().metadataFileLocation();
-      locationProvider = table.locationProvider();
-    }
-
-    /**
-     * Returns a BaseTable with {@link StaticTableOperations} so after deserialization no Catalog related calls are
-     * needed for accessing the table snapshot data.
-     * @return The BaseTable object for reading the table data at the time of the serialization of the original
-     *         BaseTable object
-     */
-    private Object readResolve()  {
-      return new BaseTable(new StaticTableOperations(metadataLocation, io, locationProvider), name);
-    }
+    return SerializableTableFactory.copyOf(this);
   }
 }

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

@@ -19,6 +19,7 @@
 
 package org.apache.iceberg;
 
+import java.io.Serializable;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -499,7 +500,7 @@ public long newSnapshotId() {
     }
   }
 
-  public class TransactionTable implements Table, HasTableOperations {
+  public class TransactionTable implements Table, HasTableOperations, Serializable {
 
     @Override
     public TableOperations operations() {
@@ -679,6 +680,10 @@ public LocationProvider locationProvider() {
     public String toString() {
       return name();
     }
+
+    Object writeReplace() {
+      return SerializableTableFactory.copyOf(this);
+    }
   }
 
   @VisibleForTesting