Core: Fix create v1 table on REST Catalog (apache#10369)

Co-authored-by: Amogh Jahagirdar <[email protected]>
kevinjqliu · Jul 5, 2024 · 24d26b6 · 24d26b6
1 parent 83dd59a
commit 24d26b6
Show file tree

Hide file tree

Showing 3 changed files with 61 additions and 5 deletions.
diff --git a/core/src/main/java/org/apache/iceberg/TableMetadata.java b/core/src/main/java/org/apache/iceberg/TableMetadata.java
@@ -853,7 +853,11 @@ public static Builder buildFrom(TableMetadata base) {
   }
 
   public static Builder buildFromEmpty() {
-    return new Builder();
+    return new Builder(DEFAULT_TABLE_FORMAT_VERSION);
+  }
+
+  public static Builder buildFromEmpty(int formatVersion) {
+    return new Builder(formatVersion);
   }
 
   public static class Builder {
@@ -903,8 +907,12 @@ public static class Builder {
     private final Map<Integer, SortOrder> sortOrdersById;
 
     private Builder() {
+      this(DEFAULT_TABLE_FORMAT_VERSION);
+    }
+
+    public Builder(int formatVersion) {
       this.base = null;
-      this.formatVersion = DEFAULT_TABLE_FORMAT_VERSION;
+      this.formatVersion = formatVersion;
       this.lastSequenceNumber = INITIAL_SEQUENCE_NUMBER;
       this.uuid = UUID.randomUUID().toString();
       this.schemas = Lists.newArrayList();

diff --git a/core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java b/core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java
@@ -27,12 +27,14 @@
 import java.time.ZoneOffset;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 import org.apache.iceberg.BaseMetadataTable;
 import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.BaseTransaction;
+import org.apache.iceberg.MetadataUpdate.UpgradeFormatVersion;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.SortOrder;
@@ -373,10 +375,15 @@ private static boolean isCreate(UpdateTableRequest request) {
   private static TableMetadata create(TableOperations ops, UpdateTableRequest request) {
     // the only valid requirement is that the table will be created
     request.requirements().forEach(requirement -> requirement.validate(ops.current()));
-
-    TableMetadata.Builder builder = TableMetadata.buildFromEmpty();
+    Optional<Integer> formatVersion =
+        request.updates().stream()
+            .filter(update -> update instanceof UpgradeFormatVersion)
+            .map(update -> ((UpgradeFormatVersion) update).formatVersion())
+            .findFirst();
+
+    TableMetadata.Builder builder =
+        formatVersion.map(TableMetadata::buildFromEmpty).orElseGet(TableMetadata::buildFromEmpty);
     request.updates().forEach(update -> update.applyTo(builder));
-
     // create transactions do not retry. if the table exists, retrying is not a solution
     ops.commit(null, builder.build());
 

diff --git a/core/src/test/java/org/apache/iceberg/catalog/CatalogTests.java b/core/src/test/java/org/apache/iceberg/catalog/CatalogTests.java
@@ -68,6 +68,8 @@
 import org.apache.iceberg.util.CharSequenceSet;
 import org.junit.jupiter.api.Assumptions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 public abstract class CatalogTests<C extends Catalog & SupportsNamespaces> {
   protected static final Namespace NS = Namespace.of("newdb");
@@ -2507,6 +2509,45 @@ public void testConcurrentReplaceTransactionSortOrderConflict() {
     assertFiles(afterSecondReplace, FILE_C);
   }
 
+  @ParameterizedTest
+  @ValueSource(ints = {1, 2})
+  public void createTableTransaction(int formatVersion) {
+    if (requiresNamespaceCreate()) {
+      catalog().createNamespace(NS);
+    }
+
+    catalog()
+        .newCreateTableTransaction(
+            TABLE,
+            SCHEMA,
+            PartitionSpec.unpartitioned(),
+            ImmutableMap.of("format-version", String.valueOf(formatVersion)))
+        .commitTransaction();
+
+    BaseTable table = (BaseTable) catalog().loadTable(TABLE);
+    assertThat(table.operations().current().formatVersion()).isEqualTo(formatVersion);
+  }
+
+  @ParameterizedTest
+  @ValueSource(ints = {1, 2})
+  public void replaceTableTransaction(int formatVersion) {
+    if (requiresNamespaceCreate()) {
+      catalog().createNamespace(NS);
+    }
+
+    catalog()
+        .newReplaceTableTransaction(
+            TABLE,
+            SCHEMA,
+            PartitionSpec.unpartitioned(),
+            ImmutableMap.of("format-version", String.valueOf(formatVersion)),
+            true)
+        .commitTransaction();
+
+    BaseTable table = (BaseTable) catalog().loadTable(TABLE);
+    assertThat(table.operations().current().formatVersion()).isEqualTo(formatVersion);
+  }
+
   @Test
   public void testMetadataFileLocationsRemovalAfterCommit() {
     C catalog = catalog();