Skip to content

Commit

Permalink
Core: Use v2 format in new tables by default (apache#8381)
Browse files Browse the repository at this point in the history
  • Loading branch information
@aokolnychyi
aokolnychyi authored Aug 30, 2023
1 parent 4e4f5a8 commit 2f03e7a
Show file tree
Hide file tree
Showing 60 changed files with 1,092 additions and 780 deletions.
16 changes: 14 additions & 2 deletions core/src/main/java/org/apache/iceberg/TableMetadata.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@
public class TableMetadata implements Serializable {
static final long INITIAL_SEQUENCE_NUMBER = 0;
static final long INVALID_SEQUENCE_NUMBER = -1;
static final int DEFAULT_TABLE_FORMAT_VERSION = 1;
static final int DEFAULT_TABLE_FORMAT_VERSION = 2;
static final int SUPPORTED_TABLE_FORMAT_VERSION = 2;
static final int INITIAL_SPEC_ID = 0;
static final int INITIAL_SORT_ORDER_ID = 1;
Expand Down Expand Up @@ -124,7 +124,7 @@ static TableMetadata newTableMetadata(
MetricsConfig.fromProperties(properties).validateReferencedColumns(schema);

return new Builder()
.upgradeFormatVersion(formatVersion)
.setInitialFormatVersion(formatVersion)
.setCurrentSchema(freshSchema, lastColumnId.get())
.setDefaultPartitionSpec(freshSpec)
.setDefaultSortOrder(freshSortOrder)
Expand Down Expand Up @@ -965,6 +965,18 @@ public Builder assignUUID(String newUuid) {
return this;
}

// it is only safe to set the format version directly while creating tables
// in all other cases, use upgradeFormatVersion
private Builder setInitialFormatVersion(int newFormatVersion) {
Preconditions.checkArgument(
newFormatVersion <= SUPPORTED_TABLE_FORMAT_VERSION,
"Unsupported format version: v%s (supported: v%s)",
newFormatVersion,
SUPPORTED_TABLE_FORMAT_VERSION);
this.formatVersion = newFormatVersion;
return this;
}

public Builder upgradeFormatVersion(int newFormatVersion) {
Preconditions.checkArgument(
newFormatVersion <= SUPPORTED_TABLE_FORMAT_VERSION,
Expand Down
7 changes: 6 additions & 1 deletion core/src/test/java/org/apache/iceberg/TestTableMetadata.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1086,7 +1086,12 @@ public void testInvalidUpdatePartitionSpecForV1Table() throws Exception {
String location = "file://tmp/db/table";
TableMetadata metadata =
TableMetadata.newTableMetadata(
schema, PartitionSpec.unpartitioned(), location, ImmutableMap.of());
schema,
PartitionSpec.unpartitioned(),
SortOrder.unsorted(),
location,
ImmutableMap.of(),
1);

Assertions.assertThatThrownBy(() -> metadata.updatePartitionSpec(spec))
.isInstanceOf(ValidationException.class)
Expand Down
22 changes: 22 additions & 0 deletions core/src/test/java/org/apache/iceberg/catalog/CatalogTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.relocated.com.google.common.collect.Streams;
import org.apache.iceberg.rest.RESTCatalog;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.CharSequenceSet;
import org.assertj.core.api.Assertions;
Expand Down Expand Up @@ -2182,6 +2183,9 @@ public void testReplaceTransactionRequiresTableExists() {
public void testConcurrentReplaceTransactions() {
C catalog = catalog();

// TODO: temporarily ignore this test for REST catalogs (issue #8390)
Assumptions.assumeFalse(catalog instanceof RESTCatalog);

if (requiresNamespaceCreate()) {
catalog.createNamespace(NS);
}
Expand Down Expand Up @@ -2233,6 +2237,9 @@ public void testConcurrentReplaceTransactions() {
public void testConcurrentReplaceTransactionSchema() {
C catalog = catalog();

// TODO: temporarily ignore this test for REST catalogs (issue #8390)
Assumptions.assumeFalse(catalog instanceof RESTCatalog);

if (requiresNamespaceCreate()) {
catalog.createNamespace(NS);
}
Expand Down Expand Up @@ -2272,6 +2279,9 @@ public void testConcurrentReplaceTransactionSchema() {
public void testConcurrentReplaceTransactionSchema2() {
C catalog = catalog();

// TODO: temporarily ignore this test for REST catalogs (issue #8390)
Assumptions.assumeFalse(catalog instanceof RESTCatalog);

if (requiresNamespaceCreate()) {
catalog.createNamespace(NS);
}
Expand Down Expand Up @@ -2351,6 +2361,9 @@ public void testConcurrentReplaceTransactionSchemaConflict() {
public void testConcurrentReplaceTransactionPartitionSpec() {
C catalog = catalog();

// TODO: temporarily ignore this test for REST catalogs (issue #8390)
Assumptions.assumeFalse(catalog instanceof RESTCatalog);

if (requiresNamespaceCreate()) {
catalog.createNamespace(NS);
}
Expand Down Expand Up @@ -2391,6 +2404,9 @@ public void testConcurrentReplaceTransactionPartitionSpec() {
public void testConcurrentReplaceTransactionPartitionSpec2() {
C catalog = catalog();

// TODO: temporarily ignore this test for REST catalogs (issue #8390)
Assumptions.assumeFalse(catalog instanceof RESTCatalog);

if (requiresNamespaceCreate()) {
catalog.createNamespace(NS);
}
Expand Down Expand Up @@ -2471,6 +2487,9 @@ public void testConcurrentReplaceTransactionPartitionSpecConflict() {
public void testConcurrentReplaceTransactionSortOrder() {
C catalog = catalog();

// TODO: temporarily ignore this test for REST catalogs (issue #8390)
Assumptions.assumeFalse(catalog instanceof RESTCatalog);

if (requiresNamespaceCreate()) {
catalog.createNamespace(NS);
}
Expand Down Expand Up @@ -2511,6 +2530,9 @@ public void testConcurrentReplaceTransactionSortOrder() {
public void testConcurrentReplaceTransactionSortOrderConflict() {
C catalog = catalog();

// TODO: temporarily ignore this test for REST catalogs (issue #8390)
Assumptions.assumeFalse(catalog instanceof RESTCatalog);

if (requiresNamespaceCreate()) {
catalog.createNamespace(NS);
}
Expand Down
49 changes: 34 additions & 15 deletions core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableOperations;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.Transaction;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
Expand All @@ -52,19 +53,23 @@
import org.apache.iceberg.transforms.Transforms;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;

public class TestHadoopCatalog extends HadoopTableTestBase {
private static ImmutableMap<String, String> meta = ImmutableMap.of();

@Test
public void testCreateTableBuilder() throws Exception {
@ParameterizedTest
@ValueSource(ints = {1, 2})
public void testCreateTableBuilder(int formatVersion) throws Exception {
TableIdentifier tableIdent = TableIdentifier.of("db", "ns1", "ns2", "tbl");
Table table =
hadoopCatalog()
.buildTable(tableIdent, SCHEMA)
.withPartitionSpec(SPEC)
.withProperties(null)
.withProperty("key1", "value1")
.withProperty(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion))
.withProperties(ImmutableMap.of("key2", "value2"))
.create();

Expand All @@ -75,21 +80,27 @@ public void testCreateTableBuilder() throws Exception {
.containsEntry("key2", "value2");
}

@Test
public void testCreateTableTxnBuilder() throws Exception {
@ParameterizedTest
@ValueSource(ints = {1, 2})
public void testCreateTableTxnBuilder(int formatVersion) throws Exception {
HadoopCatalog catalog = hadoopCatalog();
TableIdentifier tableIdent = TableIdentifier.of("db", "ns1", "ns2", "tbl");
Transaction txn =
catalog.buildTable(tableIdent, SCHEMA).withPartitionSpec(null).createTransaction();
catalog
.buildTable(tableIdent, SCHEMA)
.withPartitionSpec(null)
.withProperty(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion))
.createTransaction();
txn.commitTransaction();
Table table = catalog.loadTable(tableIdent);

Assertions.assertThat(table.schema().toString()).isEqualTo(TABLE_SCHEMA.toString());
Assertions.assertThat(table.spec().isUnpartitioned()).isTrue();
}

@Test
public void testReplaceTxnBuilder() throws Exception {
@ParameterizedTest
@ValueSource(ints = {1, 2})
public void testReplaceTxnBuilder(int formatVersion) throws Exception {
HadoopCatalog catalog = hadoopCatalog();
TableIdentifier tableIdent = TableIdentifier.of("db", "ns1", "ns2", "tbl");

Expand All @@ -98,6 +109,7 @@ public void testReplaceTxnBuilder() throws Exception {
.buildTable(tableIdent, SCHEMA)
.withPartitionSpec(SPEC)
.withProperty("key1", "value1")
.withProperty(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion))
.createOrReplaceTransaction();

createTxn.newAppend().appendFile(FILE_A).commit();
Expand All @@ -113,14 +125,21 @@ public void testReplaceTxnBuilder() throws Exception {

table = catalog.loadTable(tableIdent);
Assertions.assertThat(table.currentSnapshot()).isNull();
PartitionSpec v1Expected =
PartitionSpec.builderFor(table.schema())
.alwaysNull("data", "data_bucket")
.withSpecId(1)
.build();
Assertions.assertThat(table.spec())
.as("Table should have a spec with one void field")
.isEqualTo(v1Expected);

if (formatVersion == 1) {
PartitionSpec v1Expected =
PartitionSpec.builderFor(table.schema())
.alwaysNull("data", "data_bucket")
.withSpecId(1)
.build();
Assertions.assertThat(table.spec())
.as("Table should have a spec with one void field")
.isEqualTo(v1Expected);
} else {
Assertions.assertThat(table.spec().isUnpartitioned())
.as("Table spec should be unpartitioned")
.isTrue();
}

Assertions.assertThat(table.properties())
.containsEntry("key1", "value1")
Expand Down
29 changes: 19 additions & 10 deletions core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableOperations;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.Transaction;
import org.apache.iceberg.catalog.CatalogTests;
import org.apache.iceberg.catalog.Namespace;
Expand All @@ -73,6 +74,8 @@
import org.junit.jupiter.api.BeforeEach;
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.ValueSource;

public class TestJdbcCatalog extends CatalogTests<JdbcCatalog> {

Expand Down Expand Up @@ -190,8 +193,9 @@ public void testCreateTableTxnBuilder() {
assertThat(table.properties()).containsEntry("key1", "testval1");
}

@Test
public void testReplaceTxnBuilder() {
@ParameterizedTest
@ValueSource(ints = {1, 2})
public void testReplaceTxnBuilder(int formatVersion) {
TableIdentifier tableIdent = TableIdentifier.of("db", "ns1", "ns2", "tbl");

final DataFile fileA =
Expand All @@ -207,6 +211,7 @@ public void testReplaceTxnBuilder() {
.buildTable(tableIdent, SCHEMA)
.withPartitionSpec(PARTITION_SPEC)
.withProperty("key1", "value1")
.withProperty(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion))
.createOrReplaceTransaction();

createTxn.newAppend().appendFile(fileA).commit();
Expand All @@ -222,14 +227,18 @@ public void testReplaceTxnBuilder() {

table = catalog.loadTable(tableIdent);
assertThat(table.currentSnapshot()).isNull();
PartitionSpec v1Expected =
PartitionSpec.builderFor(table.schema())
.alwaysNull("data", "data_bucket")
.withSpecId(1)
.build();
assertThat(table.spec())
.as("Table should have a spec with one void field")
.isEqualTo(v1Expected);
if (formatVersion == 1) {
PartitionSpec v1Expected =
PartitionSpec.builderFor(table.schema())
.alwaysNull("data", "data_bucket")
.withSpecId(1)
.build();
assertThat(table.spec())
.as("Table should have a spec with one void field")
.isEqualTo(v1Expected);
} else {
assertThat(table.spec().isUnpartitioned()).as("Table spec must be unpartitioned").isTrue();
}

assertThat(table.properties()).containsEntry("key1", "value1").containsEntry("key2", "value2");
}
Expand Down
56 changes: 40 additions & 16 deletions hive-metastore/src/test/java/org/apache/iceberg/hive/HiveCreateReplaceTableTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,16 +32,20 @@
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.Transaction;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
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.ValueSource;

public class HiveCreateReplaceTableTest extends HiveMetastoreTest {

Expand Down Expand Up @@ -143,20 +147,30 @@ public void testCreateTableTxnTableAlreadyExists() {
.hasMessage("Table already exists: hivedb.tbl");
}

@Test
public void testReplaceTableTxn() {
catalog.createTable(TABLE_IDENTIFIER, SCHEMA, SPEC, tableLocation, Maps.newHashMap());
@ParameterizedTest
@ValueSource(ints = {1, 2})
public void testReplaceTableTxn(int formatVersion) {
catalog.createTable(
TABLE_IDENTIFIER,
SCHEMA,
SPEC,
tableLocation,
ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion)));
assertThat(catalog.tableExists(TABLE_IDENTIFIER)).as("Table should exist").isTrue();

Transaction txn = catalog.newReplaceTableTransaction(TABLE_IDENTIFIER, SCHEMA, false);
txn.commitTransaction();

Table table = catalog.loadTable(TABLE_IDENTIFIER);
PartitionSpec v1Expected =
PartitionSpec.builderFor(table.schema()).alwaysNull("id", "id").withSpecId(1).build();
assertThat(table.spec())
.as("Table should have a spec with one void field")
.isEqualTo(v1Expected);
if (formatVersion == 1) {
PartitionSpec v1Expected =
PartitionSpec.builderFor(table.schema()).alwaysNull("id", "id").withSpecId(1).build();
assertThat(table.spec())
.as("Table should have a spec with one void field")
.isEqualTo(v1Expected);
} else {
assertThat(table.spec().isUnpartitioned()).as("Table spec must be unpartitioned").isTrue();
}
}

@Test
Expand Down Expand Up @@ -217,20 +231,30 @@ public void testCreateOrReplaceTableTxnTableNotExists() {
assertThat(table.properties()).as("Table props should match").containsEntry("prop", "value");
}

@Test
public void testCreateOrReplaceTableTxnTableExists() {
catalog.createTable(TABLE_IDENTIFIER, SCHEMA, SPEC, tableLocation, Maps.newHashMap());
@ParameterizedTest
@ValueSource(ints = {1, 2})
public void testCreateOrReplaceTableTxnTableExists(int formatVersion) {
catalog.createTable(
TABLE_IDENTIFIER,
SCHEMA,
SPEC,
tableLocation,
ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion)));
assertThat(catalog.tableExists(TABLE_IDENTIFIER)).as("Table should exist").isTrue();

Transaction txn = catalog.newReplaceTableTransaction(TABLE_IDENTIFIER, SCHEMA, true);
txn.commitTransaction();

Table table = catalog.loadTable(TABLE_IDENTIFIER);
PartitionSpec v1Expected =
PartitionSpec.builderFor(table.schema()).alwaysNull("id", "id").withSpecId(1).build();
assertThat(table.spec())
.as("Table should have a spec with one void field")
.isEqualTo(v1Expected);
if (formatVersion == 1) {
PartitionSpec v1Expected =
PartitionSpec.builderFor(table.schema()).alwaysNull("id", "id").withSpecId(1).build();
assertThat(table.spec())
.as("Table should have a spec with one void field")
.isEqualTo(v1Expected);
} else {
assertThat(table.spec().isUnpartitioned()).as("Table spec must be unpartitioned").isTrue();
}
}

@Test
Expand Down
Loading

0 comments on commit 2f03e7a

Please sign in to comment.