This is an automated email from the ASF dual-hosted git repository.
etudenhoefner pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 24d26b6a35 Core: Fix create v1 table on REST Catalog (#10369)
24d26b6a35 is described below
commit 24d26b6a35a1287531e72357691d9dbd3d7f79bd
Author: dongwang <[email protected]>
AuthorDate: Fri Jul 5 15:08:44 2024 +0800
Core: Fix create v1 table on REST Catalog (#10369)
Co-authored-by: Amogh Jahagirdar <[email protected]>
---
.../java/org/apache/iceberg/TableMetadata.java | 12 +++++--
.../org/apache/iceberg/rest/CatalogHandlers.java | 13 +++++--
.../org/apache/iceberg/catalog/CatalogTests.java | 41 ++++++++++++++++++++++
3 files changed, 61 insertions(+), 5 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/TableMetadata.java
b/core/src/main/java/org/apache/iceberg/TableMetadata.java
index 6b23c337ae..e8dcfc85fb 100644
--- 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 class TableMetadata implements Serializable {
}
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 class TableMetadata implements Serializable {
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
index 746da5ffca..f1b7aa32d6 100644
--- 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.OffsetDateTime;
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 @@ public class CatalogHandlers {
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
index 8c95f2785b..5402a13d7d 100644
--- 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.types.Types;
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 abstract class CatalogTests<C extends Catalog &
SupportsNamespaces> {
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();
