You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ao...@apache.org on 2023/08/30 20:59:01 UTC
[iceberg] branch master updated: Core: Use v2 format in new tables by default (#8381)
This is an automated email from the ASF dual-hosted git repository.
aokolnychyi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new 2f03e7a4c0 Core: Use v2 format in new tables by default (#8381)
2f03e7a4c0 is described below
commit 2f03e7a4c05b63b11ccd27af7b28e620f19d6eb8
Author: Anton Okolnychyi <ao...@apple.com>
AuthorDate: Wed Aug 30 13:58:54 2023 -0700
Core: Use v2 format in new tables by default (#8381)
---
.../java/org/apache/iceberg/TableMetadata.java | 16 +-
.../java/org/apache/iceberg/TestTableMetadata.java | 7 +-
.../org/apache/iceberg/catalog/CatalogTests.java | 22 ++
.../apache/iceberg/hadoop/TestHadoopCatalog.java | 49 ++--
.../org/apache/iceberg/jdbc/TestJdbcCatalog.java | 29 ++-
.../iceberg/hive/HiveCreateReplaceTableTest.java | 56 +++--
.../org/apache/iceberg/hive/TestHiveCatalog.java | 28 ++-
.../extensions/TestAlterTablePartitionFields.java | 268 +++++++++++++--------
.../java/org/apache/iceberg/spark/SmokeTest.java | 6 +-
.../iceberg/spark/actions/TestCreateActions.java | 5 +-
.../spark/actions/TestRewriteDataFilesAction.java | 4 +-
.../source/TestIcebergSourceHadoopTables.java | 9 +-
.../spark/source/TestIcebergSourceHiveTables.java | 6 +-
.../spark/source/TestIcebergSourceTablesBase.java | 27 ++-
.../TestMetadataTablesWithPartitionEvolution.java | 22 +-
.../spark/source/TestSparkMetadataColumns.java | 19 +-
.../apache/iceberg/spark/source/TestTables.java | 8 +-
.../iceberg/spark/sql/TestCreateTableAsSelect.java | 1 -
.../extensions/TestAlterTablePartitionFields.java | 268 +++++++++++++--------
.../TestCreateChangelogViewProcedure.java | 3 -
.../iceberg/spark/extensions/TestDelete.java | 5 +-
.../java/org/apache/iceberg/spark/SmokeTest.java | 6 +-
.../iceberg/spark/actions/TestCreateActions.java | 5 +-
.../actions/TestDeleteReachableFilesAction.java | 8 +-
.../spark/actions/TestRewriteDataFilesAction.java | 4 +-
.../source/TestIcebergSourceHadoopTables.java | 9 +-
.../spark/source/TestIcebergSourceHiveTables.java | 6 +-
.../spark/source/TestIcebergSourceTablesBase.java | 27 ++-
.../TestMetadataTablesWithPartitionEvolution.java | 61 ++---
.../spark/source/TestSparkMetadataColumns.java | 19 +-
.../apache/iceberg/spark/source/TestTables.java | 8 +-
.../iceberg/spark/sql/TestCreateTableAsSelect.java | 1 -
.../extensions/TestAlterTablePartitionFields.java | 268 +++++++++++++--------
.../TestCreateChangelogViewProcedure.java | 3 -
.../iceberg/spark/extensions/TestDelete.java | 5 +-
.../java/org/apache/iceberg/spark/SmokeTest.java | 6 +-
.../iceberg/spark/actions/TestCreateActions.java | 5 +-
.../actions/TestDeleteReachableFilesAction.java | 8 +-
.../spark/actions/TestRewriteDataFilesAction.java | 4 +-
.../source/TestIcebergSourceHadoopTables.java | 9 +-
.../spark/source/TestIcebergSourceHiveTables.java | 6 +-
.../spark/source/TestIcebergSourceTablesBase.java | 27 ++-
.../TestMetadataTablesWithPartitionEvolution.java | 61 ++---
.../spark/source/TestSparkMetadataColumns.java | 19 +-
.../apache/iceberg/spark/source/TestTables.java | 8 +-
.../iceberg/spark/sql/TestCreateTableAsSelect.java | 1 -
.../extensions/TestAlterTablePartitionFields.java | 268 +++++++++++++--------
.../TestCreateChangelogViewProcedure.java | 3 -
.../iceberg/spark/extensions/TestDelete.java | 5 +-
.../java/org/apache/iceberg/spark/SmokeTest.java | 6 +-
.../iceberg/spark/actions/TestCreateActions.java | 5 +-
.../actions/TestDeleteReachableFilesAction.java | 8 +-
.../spark/actions/TestRewriteDataFilesAction.java | 4 +-
.../source/TestIcebergSourceHadoopTables.java | 9 +-
.../spark/source/TestIcebergSourceHiveTables.java | 6 +-
.../spark/source/TestIcebergSourceTablesBase.java | 27 ++-
.../TestMetadataTablesWithPartitionEvolution.java | 61 ++---
.../spark/source/TestSparkMetadataColumns.java | 19 +-
.../apache/iceberg/spark/source/TestTables.java | 8 +-
.../iceberg/spark/sql/TestCreateTableAsSelect.java | 1 -
60 files changed, 1092 insertions(+), 780 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/TableMetadata.java b/core/src/main/java/org/apache/iceberg/TableMetadata.java
index 25af350d5e..be510dea20 100644
--- a/core/src/main/java/org/apache/iceberg/TableMetadata.java
+++ b/core/src/main/java/org/apache/iceberg/TableMetadata.java
@@ -50,7 +50,7 @@ import org.apache.iceberg.util.SerializableSupplier;
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;
@@ -124,7 +124,7 @@ public class TableMetadata implements Serializable {
MetricsConfig.fromProperties(properties).validateReferencedColumns(schema);
return new Builder()
- .upgradeFormatVersion(formatVersion)
+ .setInitialFormatVersion(formatVersion)
.setCurrentSchema(freshSchema, lastColumnId.get())
.setDefaultPartitionSpec(freshSpec)
.setDefaultSortOrder(freshSortOrder)
@@ -965,6 +965,18 @@ public class TableMetadata implements Serializable {
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,
diff --git a/core/src/test/java/org/apache/iceberg/TestTableMetadata.java b/core/src/test/java/org/apache/iceberg/TestTableMetadata.java
index 3e4ff15042..9a75beb59d 100644
--- a/core/src/test/java/org/apache/iceberg/TestTableMetadata.java
+++ b/core/src/test/java/org/apache/iceberg/TestTableMetadata.java
@@ -1086,7 +1086,12 @@ public class TestTableMetadata {
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)
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 203b5fc381..7e9294fb9e 100644
--- a/core/src/test/java/org/apache/iceberg/catalog/CatalogTests.java
+++ b/core/src/test/java/org/apache/iceberg/catalog/CatalogTests.java
@@ -60,6 +60,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.Lists;
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;
@@ -2182,6 +2183,9 @@ public abstract class CatalogTests<C extends Catalog & SupportsNamespaces> {
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);
}
@@ -2233,6 +2237,9 @@ public abstract class CatalogTests<C extends Catalog & SupportsNamespaces> {
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);
}
@@ -2272,6 +2279,9 @@ public abstract class CatalogTests<C extends Catalog & SupportsNamespaces> {
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);
}
@@ -2351,6 +2361,9 @@ public abstract class CatalogTests<C extends Catalog & SupportsNamespaces> {
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);
}
@@ -2391,6 +2404,9 @@ public abstract class CatalogTests<C extends Catalog & SupportsNamespaces> {
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);
}
@@ -2471,6 +2487,9 @@ public abstract class CatalogTests<C extends Catalog & SupportsNamespaces> {
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);
}
@@ -2511,6 +2530,9 @@ public abstract class CatalogTests<C extends Catalog & SupportsNamespaces> {
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);
}
diff --git a/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java b/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java
index 4f889b24ca..4b01eead5d 100644
--- a/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java
@@ -36,6 +36,7 @@ import org.apache.iceberg.PartitionSpec;
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;
@@ -52,12 +53,15 @@ import org.apache.iceberg.transforms.Transform;
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()
@@ -65,6 +69,7 @@ public class TestHadoopCatalog extends HadoopTableTestBase {
.withPartitionSpec(SPEC)
.withProperties(null)
.withProperty("key1", "value1")
+ .withProperty(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion))
.withProperties(ImmutableMap.of("key2", "value2"))
.create();
@@ -75,12 +80,17 @@ public class TestHadoopCatalog extends HadoopTableTestBase {
.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);
@@ -88,8 +98,9 @@ public class TestHadoopCatalog extends HadoopTableTestBase {
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");
@@ -98,6 +109,7 @@ public class TestHadoopCatalog extends HadoopTableTestBase {
.buildTable(tableIdent, SCHEMA)
.withPartitionSpec(SPEC)
.withProperty("key1", "value1")
+ .withProperty(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion))
.createOrReplaceTransaction();
createTxn.newAppend().appendFile(FILE_A).commit();
@@ -113,14 +125,21 @@ public class TestHadoopCatalog extends HadoopTableTestBase {
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")
diff --git a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
index 4634de5707..59a701cb54 100644
--- a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
@@ -49,6 +49,7 @@ import org.apache.iceberg.Schema;
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;
@@ -73,6 +74,8 @@ import org.assertj.core.api.Assertions;
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> {
@@ -190,8 +193,9 @@ public class TestJdbcCatalog extends CatalogTests<JdbcCatalog> {
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 =
@@ -207,6 +211,7 @@ public class TestJdbcCatalog extends CatalogTests<JdbcCatalog> {
.buildTable(tableIdent, SCHEMA)
.withPartitionSpec(PARTITION_SPEC)
.withProperty("key1", "value1")
+ .withProperty(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion))
.createOrReplaceTransaction();
createTxn.newAppend().appendFile(fileA).commit();
@@ -222,14 +227,18 @@ public class TestJdbcCatalog extends CatalogTests<JdbcCatalog> {
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");
}
diff --git a/hive-metastore/src/test/java/org/apache/iceberg/hive/HiveCreateReplaceTableTest.java b/hive-metastore/src/test/java/org/apache/iceberg/hive/HiveCreateReplaceTableTest.java
index 61d3659740..e463801917 100644
--- a/hive-metastore/src/test/java/org/apache/iceberg/hive/HiveCreateReplaceTableTest.java
+++ b/hive-metastore/src/test/java/org/apache/iceberg/hive/HiveCreateReplaceTableTest.java
@@ -32,16 +32,20 @@ import org.apache.iceberg.PartitionSpec;
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 {
@@ -143,20 +147,30 @@ public class HiveCreateReplaceTableTest extends HiveMetastoreTest {
.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
@@ -217,20 +231,30 @@ public class HiveCreateReplaceTableTest extends HiveMetastoreTest {
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
diff --git a/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCatalog.java b/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCatalog.java
index d4ac498684..6b06c33a73 100644
--- a/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCatalog.java
+++ b/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCatalog.java
@@ -84,6 +84,8 @@ import org.apache.iceberg.util.JsonUtil;
import org.apache.thrift.TException;
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 TestHiveCatalog extends HiveMetastoreTest {
private static ImmutableMap meta =
@@ -202,8 +204,9 @@ public class TestHiveCatalog extends HiveMetastoreTest {
}
}
- @Test
- public void testReplaceTxnBuilder() throws Exception {
+ @ParameterizedTest
+ @ValueSource(ints = {1, 2})
+ public void testReplaceTxnBuilder(int formatVersion) {
Schema schema = getTestSchema();
PartitionSpec spec = PartitionSpec.builderFor(schema).bucket("data", 16).build();
TableIdentifier tableIdent = TableIdentifier.of(DB_NAME, "tbl");
@@ -216,6 +219,7 @@ public class TestHiveCatalog extends HiveMetastoreTest {
.withPartitionSpec(spec)
.withLocation(location)
.withProperty("key1", "value1")
+ .withProperty(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion))
.createOrReplaceTransaction();
createTxn.commitTransaction();
@@ -235,14 +239,18 @@ public class TestHiveCatalog extends HiveMetastoreTest {
table = catalog.loadTable(tableIdent);
assertThat(table.location()).isEqualTo(newLocation);
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");
assertThat(table.properties()).containsEntry("key2", "value2");
diff --git a/spark/v3.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java b/spark/v3.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java
index 8aee7c9775..5102da8442 100644
--- a/spark/v3.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java
+++ b/spark/v3.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java
@@ -18,9 +18,11 @@
*/
package org.apache.iceberg.spark.extensions;
-import java.util.Map;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.PartitionSpecParser;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.spark.SparkCatalogConfig;
import org.apache.iceberg.spark.source.SparkTable;
import org.apache.spark.sql.connector.catalog.CatalogManager;
import org.apache.spark.sql.connector.catalog.Identifier;
@@ -28,11 +30,23 @@ import org.apache.spark.sql.connector.catalog.TableCatalog;
import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
+import org.junit.runners.Parameterized;
public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
- public TestAlterTablePartitionFields(
- String catalogName, String implementation, Map<String, String> config) {
- super(catalogName, implementation, config);
+
+ @Parameterized.Parameters(name = "catalogConfig = {0}, formatVersion = {1}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {SparkCatalogConfig.HIVE, 1},
+ {SparkCatalogConfig.SPARK, 2}
+ };
+ }
+
+ private final int formatVersion;
+
+ public TestAlterTablePartitionFields(SparkCatalogConfig catalogConfig, int formatVersion) {
+ super(catalogConfig.catalogName(), catalogConfig.implementation(), catalogConfig.properties());
+ this.formatVersion = formatVersion;
}
@After
@@ -42,9 +56,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddIdentityPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -61,9 +73,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddBucketPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -83,9 +93,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddTruncatePartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -105,9 +113,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddYearsPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -124,9 +130,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddMonthsPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -143,9 +147,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddDaysPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -162,9 +164,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddHoursPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -181,9 +181,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddNamedPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -200,9 +198,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testDropIdentityPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg PARTITIONED BY (category)",
- tableName);
+ createTable("id bigint NOT NULL, category string, data string", "category");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertEquals(
@@ -212,20 +208,21 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
table.refresh();
- PartitionSpec expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(1)
- .alwaysNull("category", "category")
- .build();
-
- Assert.assertEquals("Should have new spec field", expected, table.spec());
+ if (formatVersion == 1) {
+ PartitionSpec expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(1)
+ .alwaysNull("category", "category")
+ .build();
+ Assert.assertEquals("Should have new spec field", expected, table.spec());
+ } else {
+ Assert.assertTrue("New spec must be unpartitioned", table.spec().isUnpartitioned());
+ }
}
@Test
public void testDropDaysPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, ts timestamp, data string) USING iceberg PARTITIONED BY (days(ts))",
- tableName);
+ createTable("id bigint NOT NULL, ts timestamp, data string", "days(ts)");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertEquals(
@@ -235,17 +232,18 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
table.refresh();
- PartitionSpec expected =
- PartitionSpec.builderFor(table.schema()).withSpecId(1).alwaysNull("ts", "ts_day").build();
-
- Assert.assertEquals("Should have new spec field", expected, table.spec());
+ if (formatVersion == 1) {
+ PartitionSpec expected =
+ PartitionSpec.builderFor(table.schema()).withSpecId(1).alwaysNull("ts", "ts_day").build();
+ Assert.assertEquals("Should have new spec field", expected, table.spec());
+ } else {
+ Assert.assertTrue("New spec must be unpartitioned", table.spec().isUnpartitioned());
+ }
}
@Test
public void testDropBucketPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg PARTITIONED BY (bucket(16, id))",
- tableName);
+ createTable("id bigint NOT NULL, data string", "bucket(16, id)");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertEquals(
@@ -255,20 +253,21 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
table.refresh();
- PartitionSpec expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(1)
- .alwaysNull("id", "id_bucket")
- .build();
-
- Assert.assertEquals("Should have new spec field", expected, table.spec());
+ if (formatVersion == 1) {
+ PartitionSpec expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(1)
+ .alwaysNull("id", "id_bucket")
+ .build();
+ Assert.assertEquals("Should have new spec field", expected, table.spec());
+ } else {
+ Assert.assertTrue("New spec must be unpartitioned", table.spec().isUnpartitioned());
+ }
}
@Test
public void testDropPartitionByName() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -284,17 +283,18 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
table.refresh();
- PartitionSpec expected =
- PartitionSpec.builderFor(table.schema()).withSpecId(2).alwaysNull("id", "shard").build();
-
- Assert.assertEquals("Should have new spec field", expected, table.spec());
+ if (formatVersion == 1) {
+ PartitionSpec expected =
+ PartitionSpec.builderFor(table.schema()).withSpecId(2).alwaysNull("id", "shard").build();
+ Assert.assertEquals("Should have new spec field", expected, table.spec());
+ } else {
+ Assert.assertTrue("New spec must be unpartitioned", table.spec().isUnpartitioned());
+ }
}
@Test
public void testReplacePartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -306,21 +306,34 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
sql("ALTER TABLE %s REPLACE PARTITION FIELD days(ts) WITH hours(ts)", tableName);
table.refresh();
- expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(2)
- .alwaysNull("ts", "ts_day")
- .hour("ts")
- .build();
+ if (formatVersion == 1) {
+ expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(2)
+ .alwaysNull("ts", "ts_day")
+ .hour("ts")
+ .build();
+ } else {
+ expected =
+ PartitionSpecParser.fromJson(
+ table.schema(),
+ "{\n"
+ + " \"spec-id\" : 2,\n"
+ + " \"fields\" : [ {\n"
+ + " \"name\" : \"ts_hour\",\n"
+ + " \"transform\" : \"hour\",\n"
+ + " \"source-id\" : 3,\n"
+ + " \"field-id\" : 1001\n"
+ + " } ]\n"
+ + "}");
+ }
Assert.assertEquals(
"Should changed from daily to hourly partitioned field", expected, table.spec());
}
@Test
public void testReplacePartitionAndRename() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -332,21 +345,34 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
sql("ALTER TABLE %s REPLACE PARTITION FIELD days(ts) WITH hours(ts) AS hour_col", tableName);
table.refresh();
- expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(2)
- .alwaysNull("ts", "ts_day")
- .hour("ts", "hour_col")
- .build();
+ if (formatVersion == 1) {
+ expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(2)
+ .alwaysNull("ts", "ts_day")
+ .hour("ts", "hour_col")
+ .build();
+ } else {
+ expected =
+ PartitionSpecParser.fromJson(
+ table.schema(),
+ "{\n"
+ + " \"spec-id\" : 2,\n"
+ + " \"fields\" : [ {\n"
+ + " \"name\" : \"hour_col\",\n"
+ + " \"transform\" : \"hour\",\n"
+ + " \"source-id\" : 3,\n"
+ + " \"field-id\" : 1001\n"
+ + " } ]\n"
+ + "}");
+ }
Assert.assertEquals(
"Should changed from daily to hourly partitioned field", expected, table.spec());
}
@Test
public void testReplaceNamedPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -358,21 +384,34 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
sql("ALTER TABLE %s REPLACE PARTITION FIELD day_col WITH hours(ts)", tableName);
table.refresh();
- expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(2)
- .alwaysNull("ts", "day_col")
- .hour("ts")
- .build();
+ if (formatVersion == 1) {
+ expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(2)
+ .alwaysNull("ts", "day_col")
+ .hour("ts")
+ .build();
+ } else {
+ expected =
+ PartitionSpecParser.fromJson(
+ table.schema(),
+ "{\n"
+ + " \"spec-id\" : 2,\n"
+ + " \"fields\" : [ {\n"
+ + " \"name\" : \"ts_hour\",\n"
+ + " \"transform\" : \"hour\",\n"
+ + " \"source-id\" : 3,\n"
+ + " \"field-id\" : 1001\n"
+ + " } ]\n"
+ + "}");
+ }
Assert.assertEquals(
"Should changed from daily to hourly partitioned field", expected, table.spec());
}
@Test
public void testReplaceNamedPartitionAndRenameDifferently() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -384,19 +423,34 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
sql("ALTER TABLE %s REPLACE PARTITION FIELD day_col WITH hours(ts) AS hour_col", tableName);
table.refresh();
- expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(2)
- .alwaysNull("ts", "day_col")
- .hour("ts", "hour_col")
- .build();
+ if (formatVersion == 1) {
+ expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(2)
+ .alwaysNull("ts", "day_col")
+ .hour("ts", "hour_col")
+ .build();
+ } else {
+ expected =
+ PartitionSpecParser.fromJson(
+ table.schema(),
+ "{\n"
+ + " \"spec-id\" : 2,\n"
+ + " \"fields\" : [ {\n"
+ + " \"name\" : \"hour_col\",\n"
+ + " \"transform\" : \"hour\",\n"
+ + " \"source-id\" : 3,\n"
+ + " \"field-id\" : 1001\n"
+ + " } ]\n"
+ + "}");
+ }
Assert.assertEquals(
"Should changed from daily to hourly partitioned field", expected, table.spec());
}
@Test
public void testSparkTableAddDropPartitions() throws Exception {
- sql("CREATE TABLE %s (id bigint NOT NULL, ts timestamp, data string) USING iceberg", tableName);
+ createTable("id bigint NOT NULL, ts timestamp, data string");
Assert.assertEquals(
"spark table partition should be empty", 0, sparkTable().partitioning().length);
@@ -436,4 +490,20 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
Identifier identifier = Identifier.of(tableIdent.namespace().levels(), tableIdent.name());
return (SparkTable) catalog.loadTable(identifier);
}
+
+ private void createTable(String schema) {
+ createTable(schema, null);
+ }
+
+ private void createTable(String schema, String spec) {
+ if (spec == null) {
+ sql(
+ "CREATE TABLE %s (%s) USING iceberg TBLPROPERTIES ('%s' '%d')",
+ tableName, schema, TableProperties.FORMAT_VERSION, formatVersion);
+ } else {
+ sql(
+ "CREATE TABLE %s (%s) USING iceberg PARTITIONED BY (%s) TBLPROPERTIES ('%s' '%d')",
+ tableName, schema, spec, TableProperties.FORMAT_VERSION, formatVersion);
+ }
+ }
}
diff --git a/spark/v3.1/spark-runtime/src/integration/java/org/apache/iceberg/spark/SmokeTest.java b/spark/v3.1/spark-runtime/src/integration/java/org/apache/iceberg/spark/SmokeTest.java
index 510108af8c..d6fab897d6 100644
--- a/spark/v3.1/spark-runtime/src/integration/java/org/apache/iceberg/spark/SmokeTest.java
+++ b/spark/v3.1/spark-runtime/src/integration/java/org/apache/iceberg/spark/SmokeTest.java
@@ -117,11 +117,7 @@ public class SmokeTest extends SparkExtensionsTestBase {
sql("ALTER TABLE %s DROP PARTITION FIELD shard", tableName);
table = getTable();
- Assert.assertEquals("Table should have 4 partition fields", 4, table.spec().fields().size());
- // VoidTransform is package private so we can't reach it here, just checking name
- Assert.assertTrue(
- "All transforms should be void",
- table.spec().fields().stream().allMatch(pf -> pf.transform().toString().equals("void")));
+ Assert.assertTrue("Table should be unpartitioned", table.spec().isUnpartitioned());
// Sort order examples
sql("ALTER TABLE %s WRITE ORDERED BY category, id", tableName);
diff --git a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestCreateActions.java b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestCreateActions.java
index 544c730bfe..b3ad9c0034 100644
--- a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestCreateActions.java
+++ b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestCreateActions.java
@@ -30,6 +30,7 @@ import org.apache.commons.io.FileUtils;
import org.apache.commons.io.filefilter.TrueFileFilter;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
import org.apache.iceberg.actions.MigrateTable;
import org.apache.iceberg.actions.SnapshotTable;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
@@ -517,7 +518,9 @@ public class TestCreateActions extends SparkCatalogTestBase {
String source = sourceName("test_reserved_properties_table");
String dest = destName(destTableName);
createSourceTable(CREATE_PARQUET, source);
- assertSnapshotFileCount(SparkActions.get().snapshotTable(source).as(dest), source, dest);
+ SnapshotTable action = SparkActions.get().snapshotTable(source).as(dest);
+ action.tableProperty(TableProperties.FORMAT_VERSION, "1");
+ assertSnapshotFileCount(action, source, dest);
SparkTable table = loadTable(dest);
// set sort orders
table.table().replaceSortOrder().asc("id").desc("data").commit();
diff --git a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
index 07157a1e0b..13f7f1604b 100644
--- a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
+++ b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
@@ -80,6 +80,7 @@ import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -334,7 +335,8 @@ public class TestRewriteDataFilesAction extends SparkTestBase {
@Test
public void testBinPackWithStartingSequenceNumberV1Compatibility() {
- Table table = createTablePartitioned(4, 2);
+ Map<String, String> properties = ImmutableMap.of(TableProperties.FORMAT_VERSION, "1");
+ Table table = createTablePartitioned(4, 2, SCALE, properties);
shouldHaveFiles(table, 8);
List<Object[]> expectedRecords = currentData();
table.refresh();
diff --git a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java
index b55ba0e219..c233403e63 100644
--- a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java
+++ b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java
@@ -19,6 +19,7 @@
package org.apache.iceberg.spark.source;
import java.io.File;
+import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
@@ -43,11 +44,9 @@ public class TestIcebergSourceHadoopTables extends TestIcebergSourceTablesBase {
}
@Override
- public Table createTable(TableIdentifier ident, Schema schema, PartitionSpec spec) {
- if (spec.equals(PartitionSpec.unpartitioned())) {
- return TABLES.create(schema, tableLocation);
- }
- return TABLES.create(schema, spec, tableLocation);
+ public Table createTable(
+ TableIdentifier ident, Schema schema, PartitionSpec spec, Map<String, String> properties) {
+ return TABLES.create(schema, spec, properties, tableLocation);
}
@Override
diff --git a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHiveTables.java b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHiveTables.java
index f6df8d495b..68588ca234 100644
--- a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHiveTables.java
+++ b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHiveTables.java
@@ -19,6 +19,7 @@
package org.apache.iceberg.spark.source;
import java.io.IOException;
+import java.util.Map;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.iceberg.PartitionSpec;
@@ -54,9 +55,10 @@ public class TestIcebergSourceHiveTables extends TestIcebergSourceTablesBase {
}
@Override
- public Table createTable(TableIdentifier ident, Schema schema, PartitionSpec spec) {
+ public Table createTable(
+ TableIdentifier ident, Schema schema, PartitionSpec spec, Map<String, String> properties) {
TestIcebergSourceHiveTables.currentIdentifier = ident;
- return TestIcebergSourceHiveTables.catalog.createTable(ident, schema, spec);
+ return TestIcebergSourceHiveTables.catalog.createTable(ident, schema, spec, properties);
}
@Override
diff --git a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java
index f84a39b735..8e4336cb98 100644
--- a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java
+++ b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java
@@ -28,6 +28,7 @@ import java.io.UncheckedIOException;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
+import java.util.Map;
import java.util.StringJoiner;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -108,7 +109,8 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
@Rule public TemporaryFolder temp = new TemporaryFolder();
- public abstract Table createTable(TableIdentifier ident, Schema schema, PartitionSpec spec);
+ public abstract Table createTable(
+ TableIdentifier ident, Schema schema, PartitionSpec spec, Map<String, String> properties);
public abstract Table loadTable(TableIdentifier ident, String entriesSuffix);
@@ -116,6 +118,10 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
public abstract String loadLocation(TableIdentifier ident);
+ private Table createTable(TableIdentifier ident, Schema schema, PartitionSpec spec) {
+ return createTable(ident, schema, spec, ImmutableMap.of());
+ }
+
@Test
public synchronized void testTablesSupport() {
TableIdentifier tableIdentifier = TableIdentifier.of("db", "table");
@@ -174,8 +180,8 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
// each row must inherit snapshot_id and sequence_number
rows.forEach(
row -> {
- row.put(2, 0L); // data sequence number
- row.put(3, 0L); // file sequence number
+ row.put(2, 1L); // data sequence number
+ row.put(3, 1L); // file sequence number
GenericData.Record file = (GenericData.Record) row.get("data_file");
asMetadataRecord(file);
expected.add(row);
@@ -369,8 +375,13 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
// each row must inherit snapshot_id and sequence_number
rows.forEach(
row -> {
- row.put(2, 0L); // data sequence number
- row.put(3, 0L); // file sequence number
+ if (row.get("snapshot_id").equals(table.currentSnapshot().snapshotId())) {
+ row.put(2, 3L); // data sequence number
+ row.put(3, 3L); // file sequence number
+ } else {
+ row.put(2, 1L); // data sequence number
+ row.put(3, 1L); // file sequence number
+ }
GenericData.Record file = (GenericData.Record) row.get("data_file");
asMetadataRecord(file);
expected.add(row);
@@ -531,12 +542,12 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
}
@Test
- public void testEntriesTableWithSnapshotIdInheritance() throws Exception {
+ public void testV1EntriesTableWithSnapshotIdInheritance() throws Exception {
spark.sql("DROP TABLE IF EXISTS parquet_table");
TableIdentifier tableIdentifier = TableIdentifier.of("db", "entries_inheritance_test");
- PartitionSpec spec = SPEC;
- Table table = createTable(tableIdentifier, SCHEMA, spec);
+ Map<String, String> properties = ImmutableMap.of(TableProperties.FORMAT_VERSION, "1");
+ Table table = createTable(tableIdentifier, SCHEMA, SPEC, properties);
table.updateProperties().set(TableProperties.SNAPSHOT_ID_INHERITANCE_ENABLED, "true").commit();
diff --git a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java
index 950c1c1b40..1245e1fc1f 100644
--- a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java
+++ b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java
@@ -127,10 +127,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testFilesMetadataTable() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'a1', 'b1')", tableName);
@@ -191,10 +188,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testEntriesMetadataTable() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'a1', 'b1')", tableName);
@@ -255,10 +249,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testMetadataTablesWithUnknownTransforms() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'a1', 'b1')", tableName);
@@ -335,10 +326,9 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
return spark.read().format("iceberg").load(tableName + "." + tableType.name());
}
- private void initTable() {
+ private void createTable(String schema) {
sql(
- "ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
- tableName, DEFAULT_FILE_FORMAT, fileFormat.name());
- sql("ALTER TABLE %s SET TBLPROPERTIES('%s' '%d')", tableName, FORMAT_VERSION, formatVersion);
+ "CREATE TABLE %s (%s) USING iceberg TBLPROPERTIES ('%s' '%s', '%s' '%d')",
+ tableName, schema, DEFAULT_FILE_FORMAT, fileFormat.name(), FORMAT_VERSION, formatVersion);
}
}
diff --git a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java
index 0df86a1ecb..1b7098827f 100644
--- a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java
+++ b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java
@@ -25,6 +25,7 @@ import static org.apache.iceberg.TableProperties.PARQUET_VECTORIZATION_ENABLED;
import java.io.IOException;
import java.util.List;
+import java.util.Map;
import org.apache.iceberg.AssertHelpers;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.HasTableOperations;
@@ -35,12 +36,12 @@ import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.TableOperations;
-import org.apache.iceberg.UpdateProperties;
import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.spark.SparkTestBase;
import org.apache.iceberg.types.Types;
import org.junit.After;
@@ -62,6 +63,7 @@ public class TestSparkMetadataColumns extends SparkTestBase {
Types.NestedField.required(1, "id", Types.LongType.get()),
Types.NestedField.optional(2, "category", Types.StringType.get()),
Types.NestedField.optional(3, "data", Types.StringType.get()));
+ private static final PartitionSpec SPEC = PartitionSpec.unpartitioned();
private static final PartitionSpec UNKNOWN_SPEC =
PartitionSpecParser.fromJson(
SCHEMA,
@@ -206,25 +208,22 @@ public class TestSparkMetadataColumns extends SparkTestBase {
}
private void createAndInitTable() throws IOException {
- this.table =
- TestTables.create(temp.newFolder(), TABLE_NAME, SCHEMA, PartitionSpec.unpartitioned());
-
- UpdateProperties updateProperties = table.updateProperties();
- updateProperties.set(FORMAT_VERSION, String.valueOf(formatVersion));
- updateProperties.set(DEFAULT_FILE_FORMAT, fileFormat.name());
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(FORMAT_VERSION, String.valueOf(formatVersion));
+ properties.put(DEFAULT_FILE_FORMAT, fileFormat.name());
switch (fileFormat) {
case PARQUET:
- updateProperties.set(PARQUET_VECTORIZATION_ENABLED, String.valueOf(vectorized));
+ properties.put(PARQUET_VECTORIZATION_ENABLED, String.valueOf(vectorized));
break;
case ORC:
- updateProperties.set(ORC_VECTORIZATION_ENABLED, String.valueOf(vectorized));
+ properties.put(ORC_VECTORIZATION_ENABLED, String.valueOf(vectorized));
break;
default:
Preconditions.checkState(
!vectorized, "File format %s does not support vectorized reads", fileFormat);
}
- updateProperties.commit();
+ this.table = TestTables.create(temp.newFolder(), TABLE_NAME, SCHEMA, SPEC, properties);
}
}
diff --git a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestTables.java b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestTables.java
index ef2f73c380..439a9eaf33 100644
--- a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestTables.java
+++ b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestTables.java
@@ -44,12 +44,16 @@ class TestTables {
private TestTables() {}
static TestTable create(File temp, String name, Schema schema, PartitionSpec spec) {
+ return create(temp, name, schema, spec, ImmutableMap.of());
+ }
+
+ static TestTable create(
+ File temp, String name, Schema schema, PartitionSpec spec, Map<String, String> properties) {
TestTableOperations ops = new TestTableOperations(name);
if (ops.current() != null) {
throw new AlreadyExistsException("Table %s already exists at location: %s", name, temp);
}
- ops.commit(
- null, TableMetadata.newTableMetadata(schema, spec, temp.toString(), ImmutableMap.of()));
+ ops.commit(null, TableMetadata.newTableMetadata(schema, spec, temp.toString(), properties));
return new TestTable(ops, name);
}
diff --git a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTableAsSelect.java b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTableAsSelect.java
index 2581c0fd3c..1c08b1fd5a 100644
--- a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTableAsSelect.java
+++ b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTableAsSelect.java
@@ -388,7 +388,6 @@ public class TestCreateTableAsSelect extends SparkCatalogTestBase {
PartitionSpec expectedSpec =
PartitionSpec.builderFor(expectedSchema)
- .alwaysNull("part", "part_1000")
.identity("part")
.identity("id")
.withSpecId(2) // The Spec is new
diff --git a/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java b/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java
index 8528ab296b..9b4bd12ec1 100644
--- a/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java
+++ b/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java
@@ -18,9 +18,11 @@
*/
package org.apache.iceberg.spark.extensions;
-import java.util.Map;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.PartitionSpecParser;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.spark.SparkCatalogConfig;
import org.apache.iceberg.spark.source.SparkTable;
import org.apache.spark.sql.connector.catalog.CatalogManager;
import org.apache.spark.sql.connector.catalog.Identifier;
@@ -28,11 +30,23 @@ import org.apache.spark.sql.connector.catalog.TableCatalog;
import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
+import org.junit.runners.Parameterized;
public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
- public TestAlterTablePartitionFields(
- String catalogName, String implementation, Map<String, String> config) {
- super(catalogName, implementation, config);
+
+ @Parameterized.Parameters(name = "catalogConfig = {0}, formatVersion = {1}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {SparkCatalogConfig.HIVE, 1},
+ {SparkCatalogConfig.SPARK, 2}
+ };
+ }
+
+ private final int formatVersion;
+
+ public TestAlterTablePartitionFields(SparkCatalogConfig catalogConfig, int formatVersion) {
+ super(catalogConfig.catalogName(), catalogConfig.implementation(), catalogConfig.properties());
+ this.formatVersion = formatVersion;
}
@After
@@ -42,9 +56,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddIdentityPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -61,9 +73,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddBucketPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -83,9 +93,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddTruncatePartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -105,9 +113,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddYearsPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -124,9 +130,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddMonthsPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -143,9 +147,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddDaysPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -162,9 +164,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddHoursPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -181,9 +181,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddNamedPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -200,9 +198,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testDropIdentityPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg PARTITIONED BY (category)",
- tableName);
+ createTable("id bigint NOT NULL, category string, data string", "category");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertEquals(
@@ -212,20 +208,21 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
table.refresh();
- PartitionSpec expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(1)
- .alwaysNull("category", "category")
- .build();
-
- Assert.assertEquals("Should have new spec field", expected, table.spec());
+ if (formatVersion == 1) {
+ PartitionSpec expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(1)
+ .alwaysNull("category", "category")
+ .build();
+ Assert.assertEquals("Should have new spec field", expected, table.spec());
+ } else {
+ Assert.assertTrue("New spec must be unpartitioned", table.spec().isUnpartitioned());
+ }
}
@Test
public void testDropDaysPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, ts timestamp, data string) USING iceberg PARTITIONED BY (days(ts))",
- tableName);
+ createTable("id bigint NOT NULL, ts timestamp, data string", "days(ts)");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertEquals(
@@ -235,17 +232,18 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
table.refresh();
- PartitionSpec expected =
- PartitionSpec.builderFor(table.schema()).withSpecId(1).alwaysNull("ts", "ts_day").build();
-
- Assert.assertEquals("Should have new spec field", expected, table.spec());
+ if (formatVersion == 1) {
+ PartitionSpec expected =
+ PartitionSpec.builderFor(table.schema()).withSpecId(1).alwaysNull("ts", "ts_day").build();
+ Assert.assertEquals("Should have new spec field", expected, table.spec());
+ } else {
+ Assert.assertTrue("New spec must be unpartitioned", table.spec().isUnpartitioned());
+ }
}
@Test
public void testDropBucketPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg PARTITIONED BY (bucket(16, id))",
- tableName);
+ createTable("id bigint NOT NULL, data string", "bucket(16, id)");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertEquals(
@@ -255,20 +253,21 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
table.refresh();
- PartitionSpec expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(1)
- .alwaysNull("id", "id_bucket")
- .build();
-
- Assert.assertEquals("Should have new spec field", expected, table.spec());
+ if (formatVersion == 1) {
+ PartitionSpec expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(1)
+ .alwaysNull("id", "id_bucket")
+ .build();
+ Assert.assertEquals("Should have new spec field", expected, table.spec());
+ } else {
+ Assert.assertTrue("New spec must be unpartitioned", table.spec().isUnpartitioned());
+ }
}
@Test
public void testDropPartitionByName() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -284,17 +283,18 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
table.refresh();
- PartitionSpec expected =
- PartitionSpec.builderFor(table.schema()).withSpecId(2).alwaysNull("id", "shard").build();
-
- Assert.assertEquals("Should have new spec field", expected, table.spec());
+ if (formatVersion == 1) {
+ PartitionSpec expected =
+ PartitionSpec.builderFor(table.schema()).withSpecId(2).alwaysNull("id", "shard").build();
+ Assert.assertEquals("Should have new spec field", expected, table.spec());
+ } else {
+ Assert.assertTrue("New spec must be unpartitioned", table.spec().isUnpartitioned());
+ }
}
@Test
public void testReplacePartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -306,21 +306,34 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
sql("ALTER TABLE %s REPLACE PARTITION FIELD days(ts) WITH hours(ts)", tableName);
table.refresh();
- expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(2)
- .alwaysNull("ts", "ts_day")
- .hour("ts")
- .build();
+ if (formatVersion == 1) {
+ expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(2)
+ .alwaysNull("ts", "ts_day")
+ .hour("ts")
+ .build();
+ } else {
+ expected =
+ PartitionSpecParser.fromJson(
+ table.schema(),
+ "{\n"
+ + " \"spec-id\" : 2,\n"
+ + " \"fields\" : [ {\n"
+ + " \"name\" : \"ts_hour\",\n"
+ + " \"transform\" : \"hour\",\n"
+ + " \"source-id\" : 3,\n"
+ + " \"field-id\" : 1001\n"
+ + " } ]\n"
+ + "}");
+ }
Assert.assertEquals(
"Should changed from daily to hourly partitioned field", expected, table.spec());
}
@Test
public void testReplacePartitionAndRename() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -332,21 +345,34 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
sql("ALTER TABLE %s REPLACE PARTITION FIELD days(ts) WITH hours(ts) AS hour_col", tableName);
table.refresh();
- expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(2)
- .alwaysNull("ts", "ts_day")
- .hour("ts", "hour_col")
- .build();
+ if (formatVersion == 1) {
+ expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(2)
+ .alwaysNull("ts", "ts_day")
+ .hour("ts", "hour_col")
+ .build();
+ } else {
+ expected =
+ PartitionSpecParser.fromJson(
+ table.schema(),
+ "{\n"
+ + " \"spec-id\" : 2,\n"
+ + " \"fields\" : [ {\n"
+ + " \"name\" : \"hour_col\",\n"
+ + " \"transform\" : \"hour\",\n"
+ + " \"source-id\" : 3,\n"
+ + " \"field-id\" : 1001\n"
+ + " } ]\n"
+ + "}");
+ }
Assert.assertEquals(
"Should changed from daily to hourly partitioned field", expected, table.spec());
}
@Test
public void testReplaceNamedPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -358,21 +384,34 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
sql("ALTER TABLE %s REPLACE PARTITION FIELD day_col WITH hours(ts)", tableName);
table.refresh();
- expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(2)
- .alwaysNull("ts", "day_col")
- .hour("ts")
- .build();
+ if (formatVersion == 1) {
+ expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(2)
+ .alwaysNull("ts", "day_col")
+ .hour("ts")
+ .build();
+ } else {
+ expected =
+ PartitionSpecParser.fromJson(
+ table.schema(),
+ "{\n"
+ + " \"spec-id\" : 2,\n"
+ + " \"fields\" : [ {\n"
+ + " \"name\" : \"ts_hour\",\n"
+ + " \"transform\" : \"hour\",\n"
+ + " \"source-id\" : 3,\n"
+ + " \"field-id\" : 1001\n"
+ + " } ]\n"
+ + "}");
+ }
Assert.assertEquals(
"Should changed from daily to hourly partitioned field", expected, table.spec());
}
@Test
public void testReplaceNamedPartitionAndRenameDifferently() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -384,19 +423,34 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
sql("ALTER TABLE %s REPLACE PARTITION FIELD day_col WITH hours(ts) AS hour_col", tableName);
table.refresh();
- expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(2)
- .alwaysNull("ts", "day_col")
- .hour("ts", "hour_col")
- .build();
+ if (formatVersion == 1) {
+ expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(2)
+ .alwaysNull("ts", "day_col")
+ .hour("ts", "hour_col")
+ .build();
+ } else {
+ expected =
+ PartitionSpecParser.fromJson(
+ table.schema(),
+ "{\n"
+ + " \"spec-id\" : 2,\n"
+ + " \"fields\" : [ {\n"
+ + " \"name\" : \"hour_col\",\n"
+ + " \"transform\" : \"hour\",\n"
+ + " \"source-id\" : 3,\n"
+ + " \"field-id\" : 1001\n"
+ + " } ]\n"
+ + "}");
+ }
Assert.assertEquals(
"Should changed from daily to hourly partitioned field", expected, table.spec());
}
@Test
public void testSparkTableAddDropPartitions() throws Exception {
- sql("CREATE TABLE %s (id bigint NOT NULL, ts timestamp, data string) USING iceberg", tableName);
+ createTable("id bigint NOT NULL, ts timestamp, data string");
Assert.assertEquals(
"spark table partition should be empty", 0, sparkTable().partitioning().length);
@@ -459,4 +513,20 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
Identifier identifier = Identifier.of(tableIdent.namespace().levels(), tableIdent.name());
return (SparkTable) catalog.loadTable(identifier);
}
+
+ private void createTable(String schema) {
+ createTable(schema, null);
+ }
+
+ private void createTable(String schema, String spec) {
+ if (spec == null) {
+ sql(
+ "CREATE TABLE %s (%s) USING iceberg TBLPROPERTIES ('%s' '%d')",
+ tableName, schema, TableProperties.FORMAT_VERSION, formatVersion);
+ } else {
+ sql(
+ "CREATE TABLE %s (%s) USING iceberg PARTITIONED BY (%s) TBLPROPERTIES ('%s' '%d')",
+ tableName, schema, spec, TableProperties.FORMAT_VERSION, formatVersion);
+ }
+ }
}
diff --git a/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java b/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java
index dc12b0145d..1e00153e35 100644
--- a/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java
+++ b/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java
@@ -47,19 +47,16 @@ public class TestCreateChangelogViewProcedure extends SparkExtensionsTestBase {
public void createTableWith2Columns() {
sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName);
- sql("ALTER TABLE %s SET TBLPROPERTIES ('format-version'='%d')", tableName, 1);
sql("ALTER TABLE %s ADD PARTITION FIELD data", tableName);
}
private void createTableWith3Columns() {
sql("CREATE TABLE %s (id INT, data STRING, age INT) USING iceberg", tableName);
- sql("ALTER TABLE %s SET TBLPROPERTIES ('format-version'='%d')", tableName, 1);
sql("ALTER TABLE %s ADD PARTITION FIELD id", tableName);
}
private void createTableWithIdentifierField() {
sql("CREATE TABLE %s (id INT NOT NULL, data STRING) USING iceberg", tableName);
- sql("ALTER TABLE %s SET TBLPROPERTIES ('format-version'='%d')", tableName, 1);
sql("ALTER TABLE %s SET IDENTIFIER FIELDS id", tableName);
}
diff --git a/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java b/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
index 7352ab0809..59112912a3 100644
--- a/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
+++ b/spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
@@ -991,10 +991,7 @@ public abstract class TestDelete extends SparkRowLevelOperationsTestBase {
Snapshot currentSnapshot = table.currentSnapshot();
if (mode(table) == COPY_ON_WRITE) {
- // copy-on-write is tested against v1 and such tables have different partition evolution
- // behavior
- // that's why the number of changed partitions is 4 for copy-on-write
- validateCopyOnWrite(currentSnapshot, "4", "4", "1");
+ validateCopyOnWrite(currentSnapshot, "3", "4", "1");
} else {
validateMergeOnRead(currentSnapshot, "3", "3", null);
}
diff --git a/spark/v3.2/spark-runtime/src/integration/java/org/apache/iceberg/spark/SmokeTest.java b/spark/v3.2/spark-runtime/src/integration/java/org/apache/iceberg/spark/SmokeTest.java
index 510108af8c..d6fab897d6 100644
--- a/spark/v3.2/spark-runtime/src/integration/java/org/apache/iceberg/spark/SmokeTest.java
+++ b/spark/v3.2/spark-runtime/src/integration/java/org/apache/iceberg/spark/SmokeTest.java
@@ -117,11 +117,7 @@ public class SmokeTest extends SparkExtensionsTestBase {
sql("ALTER TABLE %s DROP PARTITION FIELD shard", tableName);
table = getTable();
- Assert.assertEquals("Table should have 4 partition fields", 4, table.spec().fields().size());
- // VoidTransform is package private so we can't reach it here, just checking name
- Assert.assertTrue(
- "All transforms should be void",
- table.spec().fields().stream().allMatch(pf -> pf.transform().toString().equals("void")));
+ Assert.assertTrue("Table should be unpartitioned", table.spec().isUnpartitioned());
// Sort order examples
sql("ALTER TABLE %s WRITE ORDERED BY category, id", tableName);
diff --git a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestCreateActions.java b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestCreateActions.java
index 96950e8227..3dccd79527 100644
--- a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestCreateActions.java
+++ b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestCreateActions.java
@@ -33,6 +33,7 @@ import org.apache.commons.io.filefilter.TrueFileFilter;
import org.apache.hadoop.fs.Path;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
import org.apache.iceberg.actions.MigrateTable;
import org.apache.iceberg.actions.SnapshotTable;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -531,7 +532,9 @@ public class TestCreateActions extends SparkCatalogTestBase {
String source = sourceName("test_reserved_properties_table");
String dest = destName(destTableName);
createSourceTable(CREATE_PARQUET, source);
- assertSnapshotFileCount(SparkActions.get().snapshotTable(source).as(dest), source, dest);
+ SnapshotTableSparkAction action = SparkActions.get().snapshotTable(source).as(dest);
+ action.tableProperty(TableProperties.FORMAT_VERSION, "1");
+ assertSnapshotFileCount(action, source, dest);
SparkTable table = loadTable(dest);
// set sort orders
table.table().replaceSortOrder().asc("id").desc("data").commit();
diff --git a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestDeleteReachableFilesAction.java b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestDeleteReachableFilesAction.java
index 154e940519..0e333f216c 100644
--- a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestDeleteReachableFilesAction.java
+++ b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestDeleteReachableFilesAction.java
@@ -267,8 +267,6 @@ public class TestDeleteReachableFilesAction extends SparkTestBase {
@Test
public void testPositionDeleteFiles() {
- table.updateProperties().set(TableProperties.FORMAT_VERSION, "2").commit();
-
table.newAppend().appendFile(FILE_A).commit();
table.newAppend().appendFile(FILE_B).commit();
@@ -277,13 +275,11 @@ public class TestDeleteReachableFilesAction extends SparkTestBase {
DeleteReachableFiles baseRemoveFilesSparkAction =
sparkActions().deleteReachableFiles(metadataLocation(table)).io(table.io());
- checkRemoveFilesResults(2, 1, 0, 3, 3, 6, baseRemoveFilesSparkAction.execute());
+ checkRemoveFilesResults(2, 1, 0, 3, 3, 5, baseRemoveFilesSparkAction.execute());
}
@Test
public void testEqualityDeleteFiles() {
- table.updateProperties().set(TableProperties.FORMAT_VERSION, "2").commit();
-
table.newAppend().appendFile(FILE_A).commit();
table.newAppend().appendFile(FILE_B).commit();
@@ -292,7 +288,7 @@ public class TestDeleteReachableFilesAction extends SparkTestBase {
DeleteReachableFiles baseRemoveFilesSparkAction =
sparkActions().deleteReachableFiles(metadataLocation(table)).io(table.io());
- checkRemoveFilesResults(2, 0, 1, 3, 3, 6, baseRemoveFilesSparkAction.execute());
+ checkRemoveFilesResults(2, 0, 1, 3, 3, 5, baseRemoveFilesSparkAction.execute());
}
@Test
diff --git a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
index 6fbc3a31aa..27fb4e1796 100644
--- a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
+++ b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
@@ -84,6 +84,7 @@ import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -370,7 +371,8 @@ public class TestRewriteDataFilesAction extends SparkTestBase {
@Test
public void testBinPackWithStartingSequenceNumberV1Compatibility() {
- Table table = createTablePartitioned(4, 2);
+ Map<String, String> properties = ImmutableMap.of(TableProperties.FORMAT_VERSION, "1");
+ Table table = createTablePartitioned(4, 2, SCALE, properties);
shouldHaveFiles(table, 8);
List<Object[]> expectedRecords = currentData();
table.refresh();
diff --git a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java
index b55ba0e219..c233403e63 100644
--- a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java
+++ b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java
@@ -19,6 +19,7 @@
package org.apache.iceberg.spark.source;
import java.io.File;
+import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
@@ -43,11 +44,9 @@ public class TestIcebergSourceHadoopTables extends TestIcebergSourceTablesBase {
}
@Override
- public Table createTable(TableIdentifier ident, Schema schema, PartitionSpec spec) {
- if (spec.equals(PartitionSpec.unpartitioned())) {
- return TABLES.create(schema, tableLocation);
- }
- return TABLES.create(schema, spec, tableLocation);
+ public Table createTable(
+ TableIdentifier ident, Schema schema, PartitionSpec spec, Map<String, String> properties) {
+ return TABLES.create(schema, spec, properties, tableLocation);
}
@Override
diff --git a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHiveTables.java b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHiveTables.java
index de26f5f82c..a159c57f0f 100644
--- a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHiveTables.java
+++ b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHiveTables.java
@@ -19,6 +19,7 @@
package org.apache.iceberg.spark.source;
import java.io.IOException;
+import java.util.Map;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.iceberg.PartitionSpec;
@@ -51,9 +52,10 @@ public class TestIcebergSourceHiveTables extends TestIcebergSourceTablesBase {
}
@Override
- public Table createTable(TableIdentifier ident, Schema schema, PartitionSpec spec) {
+ public Table createTable(
+ TableIdentifier ident, Schema schema, PartitionSpec spec, Map<String, String> properties) {
TestIcebergSourceHiveTables.currentIdentifier = ident;
- return TestIcebergSourceHiveTables.catalog.createTable(ident, schema, spec);
+ return TestIcebergSourceHiveTables.catalog.createTable(ident, schema, spec, properties);
}
@Override
diff --git a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java
index 5250b217d4..68ec1bfe1d 100644
--- a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java
+++ b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java
@@ -28,6 +28,7 @@ import java.io.UncheckedIOException;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
+import java.util.Map;
import java.util.StringJoiner;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -110,7 +111,8 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
@Rule public TemporaryFolder temp = new TemporaryFolder();
- public abstract Table createTable(TableIdentifier ident, Schema schema, PartitionSpec spec);
+ public abstract Table createTable(
+ TableIdentifier ident, Schema schema, PartitionSpec spec, Map<String, String> properties);
public abstract Table loadTable(TableIdentifier ident, String entriesSuffix);
@@ -118,6 +120,10 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
public abstract String loadLocation(TableIdentifier ident);
+ private Table createTable(TableIdentifier ident, Schema schema, PartitionSpec spec) {
+ return createTable(ident, schema, spec, ImmutableMap.of());
+ }
+
@Test
public synchronized void testTablesSupport() {
TableIdentifier tableIdentifier = TableIdentifier.of("db", "table");
@@ -176,8 +182,8 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
// each row must inherit snapshot_id and sequence_number
rows.forEach(
row -> {
- row.put(2, 0L); // data sequence number
- row.put(3, 0L); // file sequence number
+ row.put(2, 1L); // data sequence number
+ row.put(3, 1L); // file sequence number
GenericData.Record file = (GenericData.Record) row.get("data_file");
TestHelpers.asMetadataRecord(file);
expected.add(row);
@@ -371,8 +377,13 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
// each row must inherit snapshot_id and sequence_number
rows.forEach(
row -> {
- row.put(2, 0L); // data sequence number
- row.put(3, 0L); // file sequence number
+ if (row.get("snapshot_id").equals(table.currentSnapshot().snapshotId())) {
+ row.put(2, 3L); // data sequence number
+ row.put(3, 3L); // file sequence number
+ } else {
+ row.put(2, 1L); // data sequence number
+ row.put(3, 1L); // file sequence number
+ }
GenericData.Record file = (GenericData.Record) row.get("data_file");
TestHelpers.asMetadataRecord(file);
expected.add(row);
@@ -534,12 +545,12 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
}
@Test
- public void testEntriesTableWithSnapshotIdInheritance() throws Exception {
+ public void testV1EntriesTableWithSnapshotIdInheritance() throws Exception {
spark.sql("DROP TABLE IF EXISTS parquet_table");
TableIdentifier tableIdentifier = TableIdentifier.of("db", "entries_inheritance_test");
- PartitionSpec spec = SPEC;
- Table table = createTable(tableIdentifier, SCHEMA, spec);
+ Map<String, String> properties = ImmutableMap.of(TableProperties.FORMAT_VERSION, "1");
+ Table table = createTable(tableIdentifier, SCHEMA, SPEC, properties);
table.updateProperties().set(TableProperties.SNAPSHOT_ID_INHERITANCE_ENABLED, "true").commit();
diff --git a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java
index 82c9a58e33..a62199181a 100644
--- a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java
+++ b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java
@@ -28,6 +28,7 @@ import static org.apache.iceberg.MetadataTableType.FILES;
import static org.apache.iceberg.MetadataTableType.PARTITIONS;
import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
import static org.apache.iceberg.TableProperties.FORMAT_VERSION;
+import static org.apache.iceberg.TableProperties.MANIFEST_MERGE_ENABLED;
import java.util.Arrays;
import java.util.List;
@@ -139,10 +140,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testFilesMetadataTable() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'a1', 'b1')", tableName);
@@ -203,11 +201,8 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testFilesMetadataTableFilter() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg "
- + "TBLPROPERTIES ('commit.manifest-merge.enabled' 'false')",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
+ sql("ALTER TABLE %s SET TBLPROPERTIES ('%s' 'false')", tableName, MANIFEST_MERGE_ENABLED);
sql("INSERT INTO TABLE %s VALUES (1, 'c1', 'd1')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'c2', 'd2')", tableName);
@@ -297,10 +292,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testEntriesMetadataTable() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'a1', 'b1')", tableName);
@@ -361,10 +353,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testPartitionsTableAddRemoveFields() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg ",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'c1', 'd1')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'c2', 'd2')", tableName);
@@ -419,10 +408,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testPartitionsTableRenameFields() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
Table table = validationCatalog.loadTable(tableIdent);
@@ -449,10 +435,8 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testPartitionsTableSwitchFields() throws Exception {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
+
Table table = validationCatalog.loadTable(tableIdent);
// verify the metadata tables after re-adding the first dropped column in the second location
@@ -514,10 +498,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testPartitionTableFilterAddRemoveFields() throws ParseException {
// Create un-partitioned table
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'c1', 'd1')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'c2', 'd2')", tableName);
@@ -576,10 +557,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
// In V2, re-added field currently conflicts with its deleted form
Assume.assumeTrue(formatVersion == 1);
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'c1', 'd1')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'c2', 'd2')", tableName);
Table table = validationCatalog.loadTable(tableIdent);
@@ -619,10 +597,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testPartitionsTableFilterRenameFields() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
Table table = validationCatalog.loadTable(tableIdent);
@@ -645,10 +620,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testMetadataTablesWithUnknownTransforms() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'a1', 'b1')", tableName);
@@ -750,10 +722,9 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
return spark.read().format("iceberg").load(tableName + "." + tableType.name());
}
- private void initTable() {
+ private void createTable(String schema) {
sql(
- "ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
- tableName, DEFAULT_FILE_FORMAT, fileFormat.name());
- sql("ALTER TABLE %s SET TBLPROPERTIES('%s' '%d')", tableName, FORMAT_VERSION, formatVersion);
+ "CREATE TABLE %s (%s) USING iceberg TBLPROPERTIES ('%s' '%s', '%s' '%d')",
+ tableName, schema, DEFAULT_FILE_FORMAT, fileFormat.name(), FORMAT_VERSION, formatVersion);
}
}
diff --git a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java
index e399852285..79b755872d 100644
--- a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java
+++ b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java
@@ -28,6 +28,7 @@ import static org.apache.spark.sql.functions.lit;
import java.io.IOException;
import java.util.List;
+import java.util.Map;
import java.util.stream.Collectors;
import org.apache.iceberg.AssertHelpers;
import org.apache.iceberg.FileFormat;
@@ -39,13 +40,13 @@ import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.TableOperations;
-import org.apache.iceberg.UpdateProperties;
import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.spark.SparkTestBase;
import org.apache.iceberg.types.Types;
import org.apache.spark.sql.Dataset;
@@ -72,6 +73,7 @@ public class TestSparkMetadataColumns extends SparkTestBase {
Types.NestedField.optional(1, "id", Types.LongType.get()),
Types.NestedField.optional(2, "category", Types.StringType.get()),
Types.NestedField.optional(3, "data", Types.StringType.get()));
+ private static final PartitionSpec SPEC = PartitionSpec.unpartitioned();
private static final PartitionSpec UNKNOWN_SPEC =
PartitionSpecParser.fromJson(
SCHEMA,
@@ -264,25 +266,22 @@ public class TestSparkMetadataColumns extends SparkTestBase {
}
private void createAndInitTable() throws IOException {
- this.table =
- TestTables.create(temp.newFolder(), TABLE_NAME, SCHEMA, PartitionSpec.unpartitioned());
-
- UpdateProperties updateProperties = table.updateProperties();
- updateProperties.set(FORMAT_VERSION, String.valueOf(formatVersion));
- updateProperties.set(DEFAULT_FILE_FORMAT, fileFormat.name());
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(FORMAT_VERSION, String.valueOf(formatVersion));
+ properties.put(DEFAULT_FILE_FORMAT, fileFormat.name());
switch (fileFormat) {
case PARQUET:
- updateProperties.set(PARQUET_VECTORIZATION_ENABLED, String.valueOf(vectorized));
+ properties.put(PARQUET_VECTORIZATION_ENABLED, String.valueOf(vectorized));
break;
case ORC:
- updateProperties.set(ORC_VECTORIZATION_ENABLED, String.valueOf(vectorized));
+ properties.put(ORC_VECTORIZATION_ENABLED, String.valueOf(vectorized));
break;
default:
Preconditions.checkState(
!vectorized, "File format %s does not support vectorized reads", fileFormat);
}
- updateProperties.commit();
+ this.table = TestTables.create(temp.newFolder(), TABLE_NAME, SCHEMA, SPEC, properties);
}
}
diff --git a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestTables.java b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestTables.java
index 0650cb9738..b54bb315c5 100644
--- a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestTables.java
+++ b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestTables.java
@@ -44,12 +44,16 @@ class TestTables {
private TestTables() {}
static TestTable create(File temp, String name, Schema schema, PartitionSpec spec) {
+ return create(temp, name, schema, spec, ImmutableMap.of());
+ }
+
+ static TestTable create(
+ File temp, String name, Schema schema, PartitionSpec spec, Map<String, String> properties) {
TestTableOperations ops = new TestTableOperations(name);
if (ops.current() != null) {
throw new AlreadyExistsException("Table %s already exists at location: %s", name, temp);
}
- ops.commit(
- null, TableMetadata.newTableMetadata(schema, spec, temp.toString(), ImmutableMap.of()));
+ ops.commit(null, TableMetadata.newTableMetadata(schema, spec, temp.toString(), properties));
return new TestTable(ops, name);
}
diff --git a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTableAsSelect.java b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTableAsSelect.java
index 2581c0fd3c..1c08b1fd5a 100644
--- a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTableAsSelect.java
+++ b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTableAsSelect.java
@@ -388,7 +388,6 @@ public class TestCreateTableAsSelect extends SparkCatalogTestBase {
PartitionSpec expectedSpec =
PartitionSpec.builderFor(expectedSchema)
- .alwaysNull("part", "part_1000")
.identity("part")
.identity("id")
.withSpecId(2) // The Spec is new
diff --git a/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java b/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java
index a4928e6552..a43f2a041b 100644
--- a/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java
+++ b/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java
@@ -18,9 +18,11 @@
*/
package org.apache.iceberg.spark.extensions;
-import java.util.Map;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.PartitionSpecParser;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.spark.SparkCatalogConfig;
import org.apache.iceberg.spark.source.SparkTable;
import org.apache.spark.sql.connector.catalog.CatalogManager;
import org.apache.spark.sql.connector.catalog.Identifier;
@@ -28,11 +30,23 @@ import org.apache.spark.sql.connector.catalog.TableCatalog;
import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
+import org.junit.runners.Parameterized;
public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
- public TestAlterTablePartitionFields(
- String catalogName, String implementation, Map<String, String> config) {
- super(catalogName, implementation, config);
+
+ @Parameterized.Parameters(name = "catalogConfig = {0}, formatVersion = {1}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {SparkCatalogConfig.HIVE, 1},
+ {SparkCatalogConfig.SPARK, 2}
+ };
+ }
+
+ private final int formatVersion;
+
+ public TestAlterTablePartitionFields(SparkCatalogConfig catalogConfig, int formatVersion) {
+ super(catalogConfig.catalogName(), catalogConfig.implementation(), catalogConfig.properties());
+ this.formatVersion = formatVersion;
}
@After
@@ -42,9 +56,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddIdentityPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -61,9 +73,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddBucketPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -83,9 +93,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddTruncatePartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -105,9 +113,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddYearsPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -124,9 +130,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddMonthsPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -143,9 +147,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddDaysPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -162,9 +164,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddHoursPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -181,9 +181,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddNamedPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -200,9 +198,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testDropIdentityPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg PARTITIONED BY (category)",
- tableName);
+ createTable("id bigint NOT NULL, category string, data string", "category");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertEquals(
@@ -212,20 +208,21 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
table.refresh();
- PartitionSpec expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(1)
- .alwaysNull("category", "category")
- .build();
-
- Assert.assertEquals("Should have new spec field", expected, table.spec());
+ if (formatVersion == 1) {
+ PartitionSpec expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(1)
+ .alwaysNull("category", "category")
+ .build();
+ Assert.assertEquals("Should have new spec field", expected, table.spec());
+ } else {
+ Assert.assertTrue("New spec must be unpartitioned", table.spec().isUnpartitioned());
+ }
}
@Test
public void testDropDaysPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, ts timestamp, data string) USING iceberg PARTITIONED BY (days(ts))",
- tableName);
+ createTable("id bigint NOT NULL, ts timestamp, data string", "days(ts)");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertEquals(
@@ -235,17 +232,18 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
table.refresh();
- PartitionSpec expected =
- PartitionSpec.builderFor(table.schema()).withSpecId(1).alwaysNull("ts", "ts_day").build();
-
- Assert.assertEquals("Should have new spec field", expected, table.spec());
+ if (formatVersion == 1) {
+ PartitionSpec expected =
+ PartitionSpec.builderFor(table.schema()).withSpecId(1).alwaysNull("ts", "ts_day").build();
+ Assert.assertEquals("Should have new spec field", expected, table.spec());
+ } else {
+ Assert.assertTrue("New spec must be unpartitioned", table.spec().isUnpartitioned());
+ }
}
@Test
public void testDropBucketPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg PARTITIONED BY (bucket(16, id))",
- tableName);
+ createTable("id bigint NOT NULL, data string", "bucket(16, id)");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertEquals(
@@ -255,20 +253,21 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
table.refresh();
- PartitionSpec expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(1)
- .alwaysNull("id", "id_bucket")
- .build();
-
- Assert.assertEquals("Should have new spec field", expected, table.spec());
+ if (formatVersion == 1) {
+ PartitionSpec expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(1)
+ .alwaysNull("id", "id_bucket")
+ .build();
+ Assert.assertEquals("Should have new spec field", expected, table.spec());
+ } else {
+ Assert.assertTrue("New spec must be unpartitioned", table.spec().isUnpartitioned());
+ }
}
@Test
public void testDropPartitionByName() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -284,17 +283,18 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
table.refresh();
- PartitionSpec expected =
- PartitionSpec.builderFor(table.schema()).withSpecId(2).alwaysNull("id", "shard").build();
-
- Assert.assertEquals("Should have new spec field", expected, table.spec());
+ if (formatVersion == 1) {
+ PartitionSpec expected =
+ PartitionSpec.builderFor(table.schema()).withSpecId(2).alwaysNull("id", "shard").build();
+ Assert.assertEquals("Should have new spec field", expected, table.spec());
+ } else {
+ Assert.assertTrue("New spec must be unpartitioned", table.spec().isUnpartitioned());
+ }
}
@Test
public void testReplacePartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -306,21 +306,34 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
sql("ALTER TABLE %s REPLACE PARTITION FIELD days(ts) WITH hours(ts)", tableName);
table.refresh();
- expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(2)
- .alwaysNull("ts", "ts_day")
- .hour("ts")
- .build();
+ if (formatVersion == 1) {
+ expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(2)
+ .alwaysNull("ts", "ts_day")
+ .hour("ts")
+ .build();
+ } else {
+ expected =
+ PartitionSpecParser.fromJson(
+ table.schema(),
+ "{\n"
+ + " \"spec-id\" : 2,\n"
+ + " \"fields\" : [ {\n"
+ + " \"name\" : \"ts_hour\",\n"
+ + " \"transform\" : \"hour\",\n"
+ + " \"source-id\" : 3,\n"
+ + " \"field-id\" : 1001\n"
+ + " } ]\n"
+ + "}");
+ }
Assert.assertEquals(
"Should changed from daily to hourly partitioned field", expected, table.spec());
}
@Test
public void testReplacePartitionAndRename() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -332,21 +345,34 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
sql("ALTER TABLE %s REPLACE PARTITION FIELD days(ts) WITH hours(ts) AS hour_col", tableName);
table.refresh();
- expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(2)
- .alwaysNull("ts", "ts_day")
- .hour("ts", "hour_col")
- .build();
+ if (formatVersion == 1) {
+ expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(2)
+ .alwaysNull("ts", "ts_day")
+ .hour("ts", "hour_col")
+ .build();
+ } else {
+ expected =
+ PartitionSpecParser.fromJson(
+ table.schema(),
+ "{\n"
+ + " \"spec-id\" : 2,\n"
+ + " \"fields\" : [ {\n"
+ + " \"name\" : \"hour_col\",\n"
+ + " \"transform\" : \"hour\",\n"
+ + " \"source-id\" : 3,\n"
+ + " \"field-id\" : 1001\n"
+ + " } ]\n"
+ + "}");
+ }
Assert.assertEquals(
"Should changed from daily to hourly partitioned field", expected, table.spec());
}
@Test
public void testReplaceNamedPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -358,21 +384,34 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
sql("ALTER TABLE %s REPLACE PARTITION FIELD day_col WITH hours(ts)", tableName);
table.refresh();
- expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(2)
- .alwaysNull("ts", "day_col")
- .hour("ts")
- .build();
+ if (formatVersion == 1) {
+ expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(2)
+ .alwaysNull("ts", "day_col")
+ .hour("ts")
+ .build();
+ } else {
+ expected =
+ PartitionSpecParser.fromJson(
+ table.schema(),
+ "{\n"
+ + " \"spec-id\" : 2,\n"
+ + " \"fields\" : [ {\n"
+ + " \"name\" : \"ts_hour\",\n"
+ + " \"transform\" : \"hour\",\n"
+ + " \"source-id\" : 3,\n"
+ + " \"field-id\" : 1001\n"
+ + " } ]\n"
+ + "}");
+ }
Assert.assertEquals(
"Should changed from daily to hourly partitioned field", expected, table.spec());
}
@Test
public void testReplaceNamedPartitionAndRenameDifferently() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -384,19 +423,34 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
sql("ALTER TABLE %s REPLACE PARTITION FIELD day_col WITH hours(ts) AS hour_col", tableName);
table.refresh();
- expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(2)
- .alwaysNull("ts", "day_col")
- .hour("ts", "hour_col")
- .build();
+ if (formatVersion == 1) {
+ expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(2)
+ .alwaysNull("ts", "day_col")
+ .hour("ts", "hour_col")
+ .build();
+ } else {
+ expected =
+ PartitionSpecParser.fromJson(
+ table.schema(),
+ "{\n"
+ + " \"spec-id\" : 2,\n"
+ + " \"fields\" : [ {\n"
+ + " \"name\" : \"hour_col\",\n"
+ + " \"transform\" : \"hour\",\n"
+ + " \"source-id\" : 3,\n"
+ + " \"field-id\" : 1001\n"
+ + " } ]\n"
+ + "}");
+ }
Assert.assertEquals(
"Should changed from daily to hourly partitioned field", expected, table.spec());
}
@Test
public void testSparkTableAddDropPartitions() throws Exception {
- sql("CREATE TABLE %s (id bigint NOT NULL, ts timestamp, data string) USING iceberg", tableName);
+ createTable("id bigint NOT NULL, ts timestamp, data string");
Assert.assertEquals(
"spark table partition should be empty", 0, sparkTable().partitioning().length);
@@ -459,4 +513,20 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
Identifier identifier = Identifier.of(tableIdent.namespace().levels(), tableIdent.name());
return (SparkTable) catalog.loadTable(identifier);
}
+
+ private void createTable(String schema) {
+ createTable(schema, null);
+ }
+
+ private void createTable(String schema, String spec) {
+ if (spec == null) {
+ sql(
+ "CREATE TABLE %s (%s) USING iceberg TBLPROPERTIES ('%s' '%d')",
+ tableName, schema, TableProperties.FORMAT_VERSION, formatVersion);
+ } else {
+ sql(
+ "CREATE TABLE %s (%s) USING iceberg PARTITIONED BY (%s) TBLPROPERTIES ('%s' '%d')",
+ tableName, schema, spec, TableProperties.FORMAT_VERSION, formatVersion);
+ }
+ }
}
diff --git a/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java b/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java
index 0a4a1073c3..015f2ad5fd 100644
--- a/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java
+++ b/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java
@@ -49,19 +49,16 @@ public class TestCreateChangelogViewProcedure extends SparkExtensionsTestBase {
public void createTableWithTwoColumns() {
sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName);
- sql("ALTER TABLE %s SET TBLPROPERTIES ('format-version'='%d')", tableName, 1);
sql("ALTER TABLE %s ADD PARTITION FIELD data", tableName);
}
private void createTableWithThreeColumns() {
sql("CREATE TABLE %s (id INT, data STRING, age INT) USING iceberg", tableName);
- sql("ALTER TABLE %s SET TBLPROPERTIES ('format-version'='%d')", tableName, 1);
sql("ALTER TABLE %s ADD PARTITION FIELD id", tableName);
}
private void createTableWithIdentifierField() {
sql("CREATE TABLE %s (id INT NOT NULL, data STRING) USING iceberg", tableName);
- sql("ALTER TABLE %s SET TBLPROPERTIES ('format-version'='%d')", tableName, 1);
sql("ALTER TABLE %s SET IDENTIFIER FIELDS id", tableName);
}
diff --git a/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java b/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
index 934f1b91f7..4b60d580ae 100644
--- a/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
+++ b/spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
@@ -1055,10 +1055,7 @@ public abstract class TestDelete extends SparkRowLevelOperationsTestBase {
Snapshot currentSnapshot = SnapshotUtil.latestSnapshot(table, branch);
if (mode(table) == COPY_ON_WRITE) {
- // copy-on-write is tested against v1 and such tables have different partition evolution
- // behavior
- // that's why the number of changed partitions is 4 for copy-on-write
- validateCopyOnWrite(currentSnapshot, "4", "4", "1");
+ validateCopyOnWrite(currentSnapshot, "3", "4", "1");
} else {
validateMergeOnRead(currentSnapshot, "3", "3", null);
}
diff --git a/spark/v3.3/spark-runtime/src/integration/java/org/apache/iceberg/spark/SmokeTest.java b/spark/v3.3/spark-runtime/src/integration/java/org/apache/iceberg/spark/SmokeTest.java
index 510108af8c..d6fab897d6 100644
--- a/spark/v3.3/spark-runtime/src/integration/java/org/apache/iceberg/spark/SmokeTest.java
+++ b/spark/v3.3/spark-runtime/src/integration/java/org/apache/iceberg/spark/SmokeTest.java
@@ -117,11 +117,7 @@ public class SmokeTest extends SparkExtensionsTestBase {
sql("ALTER TABLE %s DROP PARTITION FIELD shard", tableName);
table = getTable();
- Assert.assertEquals("Table should have 4 partition fields", 4, table.spec().fields().size());
- // VoidTransform is package private so we can't reach it here, just checking name
- Assert.assertTrue(
- "All transforms should be void",
- table.spec().fields().stream().allMatch(pf -> pf.transform().toString().equals("void")));
+ Assert.assertTrue("Table should be unpartitioned", table.spec().isUnpartitioned());
// Sort order examples
sql("ALTER TABLE %s WRITE ORDERED BY category, id", tableName);
diff --git a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestCreateActions.java b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestCreateActions.java
index 96950e8227..3dccd79527 100644
--- a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestCreateActions.java
+++ b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestCreateActions.java
@@ -33,6 +33,7 @@ import org.apache.commons.io.filefilter.TrueFileFilter;
import org.apache.hadoop.fs.Path;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
import org.apache.iceberg.actions.MigrateTable;
import org.apache.iceberg.actions.SnapshotTable;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -531,7 +532,9 @@ public class TestCreateActions extends SparkCatalogTestBase {
String source = sourceName("test_reserved_properties_table");
String dest = destName(destTableName);
createSourceTable(CREATE_PARQUET, source);
- assertSnapshotFileCount(SparkActions.get().snapshotTable(source).as(dest), source, dest);
+ SnapshotTableSparkAction action = SparkActions.get().snapshotTable(source).as(dest);
+ action.tableProperty(TableProperties.FORMAT_VERSION, "1");
+ assertSnapshotFileCount(action, source, dest);
SparkTable table = loadTable(dest);
// set sort orders
table.table().replaceSortOrder().asc("id").desc("data").commit();
diff --git a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestDeleteReachableFilesAction.java b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestDeleteReachableFilesAction.java
index 154e940519..0e333f216c 100644
--- a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestDeleteReachableFilesAction.java
+++ b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestDeleteReachableFilesAction.java
@@ -267,8 +267,6 @@ public class TestDeleteReachableFilesAction extends SparkTestBase {
@Test
public void testPositionDeleteFiles() {
- table.updateProperties().set(TableProperties.FORMAT_VERSION, "2").commit();
-
table.newAppend().appendFile(FILE_A).commit();
table.newAppend().appendFile(FILE_B).commit();
@@ -277,13 +275,11 @@ public class TestDeleteReachableFilesAction extends SparkTestBase {
DeleteReachableFiles baseRemoveFilesSparkAction =
sparkActions().deleteReachableFiles(metadataLocation(table)).io(table.io());
- checkRemoveFilesResults(2, 1, 0, 3, 3, 6, baseRemoveFilesSparkAction.execute());
+ checkRemoveFilesResults(2, 1, 0, 3, 3, 5, baseRemoveFilesSparkAction.execute());
}
@Test
public void testEqualityDeleteFiles() {
- table.updateProperties().set(TableProperties.FORMAT_VERSION, "2").commit();
-
table.newAppend().appendFile(FILE_A).commit();
table.newAppend().appendFile(FILE_B).commit();
@@ -292,7 +288,7 @@ public class TestDeleteReachableFilesAction extends SparkTestBase {
DeleteReachableFiles baseRemoveFilesSparkAction =
sparkActions().deleteReachableFiles(metadataLocation(table)).io(table.io());
- checkRemoveFilesResults(2, 0, 1, 3, 3, 6, baseRemoveFilesSparkAction.execute());
+ checkRemoveFilesResults(2, 0, 1, 3, 3, 5, baseRemoveFilesSparkAction.execute());
}
@Test
diff --git a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
index 7a63da6068..c5bf7548c6 100644
--- a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
+++ b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
@@ -84,6 +84,7 @@ import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -371,7 +372,8 @@ public class TestRewriteDataFilesAction extends SparkTestBase {
@Test
public void testBinPackWithStartingSequenceNumberV1Compatibility() {
- Table table = createTablePartitioned(4, 2);
+ Map<String, String> properties = ImmutableMap.of(TableProperties.FORMAT_VERSION, "1");
+ Table table = createTablePartitioned(4, 2, SCALE, properties);
shouldHaveFiles(table, 8);
List<Object[]> expectedRecords = currentData();
table.refresh();
diff --git a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java
index 9bd7220b90..746415818c 100644
--- a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java
+++ b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java
@@ -19,6 +19,7 @@
package org.apache.iceberg.spark.source;
import java.io.File;
+import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
@@ -43,11 +44,9 @@ public class TestIcebergSourceHadoopTables extends TestIcebergSourceTablesBase {
}
@Override
- public Table createTable(TableIdentifier ident, Schema schema, PartitionSpec spec) {
- if (spec.equals(PartitionSpec.unpartitioned())) {
- return TABLES.create(schema, tableLocation);
- }
- return TABLES.create(schema, spec, tableLocation);
+ public Table createTable(
+ TableIdentifier ident, Schema schema, PartitionSpec spec, Map<String, String> properties) {
+ return TABLES.create(schema, spec, properties, tableLocation);
}
@Override
diff --git a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHiveTables.java b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHiveTables.java
index 6292a2c1a8..2a264b74b0 100644
--- a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHiveTables.java
+++ b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHiveTables.java
@@ -19,6 +19,7 @@
package org.apache.iceberg.spark.source;
import java.io.IOException;
+import java.util.Map;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.iceberg.PartitionSpec;
@@ -51,9 +52,10 @@ public class TestIcebergSourceHiveTables extends TestIcebergSourceTablesBase {
}
@Override
- public Table createTable(TableIdentifier ident, Schema schema, PartitionSpec spec) {
+ public Table createTable(
+ TableIdentifier ident, Schema schema, PartitionSpec spec, Map<String, String> properties) {
TestIcebergSourceHiveTables.currentIdentifier = ident;
- return TestIcebergSourceHiveTables.catalog.createTable(ident, schema, spec);
+ return TestIcebergSourceHiveTables.catalog.createTable(ident, schema, spec, properties);
}
@Override
diff --git a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java
index 94b3836884..143488176f 100644
--- a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java
+++ b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java
@@ -31,6 +31,7 @@ import java.time.ZoneOffset;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
+import java.util.Map;
import java.util.StringJoiner;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -119,7 +120,8 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
@Rule public TemporaryFolder temp = new TemporaryFolder();
- public abstract Table createTable(TableIdentifier ident, Schema schema, PartitionSpec spec);
+ public abstract Table createTable(
+ TableIdentifier ident, Schema schema, PartitionSpec spec, Map<String, String> properties);
public abstract Table loadTable(TableIdentifier ident, String entriesSuffix);
@@ -134,6 +136,10 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
spark.sql("DROP TABLE IF EXISTS parquet_table");
}
+ private Table createTable(TableIdentifier ident, Schema schema, PartitionSpec spec) {
+ return createTable(ident, schema, spec, ImmutableMap.of());
+ }
+
@Test
public synchronized void testTablesSupport() {
TableIdentifier tableIdentifier = TableIdentifier.of("db", "table");
@@ -192,8 +198,8 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
// each row must inherit snapshot_id and sequence_number
rows.forEach(
row -> {
- row.put(2, 0L); // data sequence number
- row.put(3, 0L); // file sequence number
+ row.put(2, 1L); // data sequence number
+ row.put(3, 1L); // file sequence number
GenericData.Record file = (GenericData.Record) row.get("data_file");
TestHelpers.asMetadataRecord(file);
expected.add(row);
@@ -387,8 +393,13 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
// each row must inherit snapshot_id and sequence_number
rows.forEach(
row -> {
- row.put(2, 0L); // data sequence number
- row.put(3, 0L); // file sequence number
+ if (row.get("snapshot_id").equals(table.currentSnapshot().snapshotId())) {
+ row.put(2, 3L); // data sequence number
+ row.put(3, 3L); // file sequence number
+ } else {
+ row.put(2, 1L); // data sequence number
+ row.put(3, 1L); // file sequence number
+ }
GenericData.Record file = (GenericData.Record) row.get("data_file");
TestHelpers.asMetadataRecord(file);
expected.add(row);
@@ -545,10 +556,10 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
}
@Test
- public void testEntriesTableWithSnapshotIdInheritance() throws Exception {
+ public void testV1EntriesTableWithSnapshotIdInheritance() throws Exception {
TableIdentifier tableIdentifier = TableIdentifier.of("db", "entries_inheritance_test");
- PartitionSpec spec = SPEC;
- Table table = createTable(tableIdentifier, SCHEMA, spec);
+ Map<String, String> properties = ImmutableMap.of(TableProperties.FORMAT_VERSION, "1");
+ Table table = createTable(tableIdentifier, SCHEMA, SPEC, properties);
table.updateProperties().set(TableProperties.SNAPSHOT_ID_INHERITANCE_ENABLED, "true").commit();
diff --git a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java
index 82c9a58e33..a62199181a 100644
--- a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java
+++ b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java
@@ -28,6 +28,7 @@ import static org.apache.iceberg.MetadataTableType.FILES;
import static org.apache.iceberg.MetadataTableType.PARTITIONS;
import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
import static org.apache.iceberg.TableProperties.FORMAT_VERSION;
+import static org.apache.iceberg.TableProperties.MANIFEST_MERGE_ENABLED;
import java.util.Arrays;
import java.util.List;
@@ -139,10 +140,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testFilesMetadataTable() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'a1', 'b1')", tableName);
@@ -203,11 +201,8 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testFilesMetadataTableFilter() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg "
- + "TBLPROPERTIES ('commit.manifest-merge.enabled' 'false')",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
+ sql("ALTER TABLE %s SET TBLPROPERTIES ('%s' 'false')", tableName, MANIFEST_MERGE_ENABLED);
sql("INSERT INTO TABLE %s VALUES (1, 'c1', 'd1')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'c2', 'd2')", tableName);
@@ -297,10 +292,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testEntriesMetadataTable() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'a1', 'b1')", tableName);
@@ -361,10 +353,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testPartitionsTableAddRemoveFields() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg ",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'c1', 'd1')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'c2', 'd2')", tableName);
@@ -419,10 +408,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testPartitionsTableRenameFields() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
Table table = validationCatalog.loadTable(tableIdent);
@@ -449,10 +435,8 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testPartitionsTableSwitchFields() throws Exception {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
+
Table table = validationCatalog.loadTable(tableIdent);
// verify the metadata tables after re-adding the first dropped column in the second location
@@ -514,10 +498,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testPartitionTableFilterAddRemoveFields() throws ParseException {
// Create un-partitioned table
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'c1', 'd1')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'c2', 'd2')", tableName);
@@ -576,10 +557,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
// In V2, re-added field currently conflicts with its deleted form
Assume.assumeTrue(formatVersion == 1);
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'c1', 'd1')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'c2', 'd2')", tableName);
Table table = validationCatalog.loadTable(tableIdent);
@@ -619,10 +597,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testPartitionsTableFilterRenameFields() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
Table table = validationCatalog.loadTable(tableIdent);
@@ -645,10 +620,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testMetadataTablesWithUnknownTransforms() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'a1', 'b1')", tableName);
@@ -750,10 +722,9 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
return spark.read().format("iceberg").load(tableName + "." + tableType.name());
}
- private void initTable() {
+ private void createTable(String schema) {
sql(
- "ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
- tableName, DEFAULT_FILE_FORMAT, fileFormat.name());
- sql("ALTER TABLE %s SET TBLPROPERTIES('%s' '%d')", tableName, FORMAT_VERSION, formatVersion);
+ "CREATE TABLE %s (%s) USING iceberg TBLPROPERTIES ('%s' '%s', '%s' '%d')",
+ tableName, schema, DEFAULT_FILE_FORMAT, fileFormat.name(), FORMAT_VERSION, formatVersion);
}
}
diff --git a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java
index e399852285..79b755872d 100644
--- a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java
+++ b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java
@@ -28,6 +28,7 @@ import static org.apache.spark.sql.functions.lit;
import java.io.IOException;
import java.util.List;
+import java.util.Map;
import java.util.stream.Collectors;
import org.apache.iceberg.AssertHelpers;
import org.apache.iceberg.FileFormat;
@@ -39,13 +40,13 @@ import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.TableOperations;
-import org.apache.iceberg.UpdateProperties;
import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.spark.SparkTestBase;
import org.apache.iceberg.types.Types;
import org.apache.spark.sql.Dataset;
@@ -72,6 +73,7 @@ public class TestSparkMetadataColumns extends SparkTestBase {
Types.NestedField.optional(1, "id", Types.LongType.get()),
Types.NestedField.optional(2, "category", Types.StringType.get()),
Types.NestedField.optional(3, "data", Types.StringType.get()));
+ private static final PartitionSpec SPEC = PartitionSpec.unpartitioned();
private static final PartitionSpec UNKNOWN_SPEC =
PartitionSpecParser.fromJson(
SCHEMA,
@@ -264,25 +266,22 @@ public class TestSparkMetadataColumns extends SparkTestBase {
}
private void createAndInitTable() throws IOException {
- this.table =
- TestTables.create(temp.newFolder(), TABLE_NAME, SCHEMA, PartitionSpec.unpartitioned());
-
- UpdateProperties updateProperties = table.updateProperties();
- updateProperties.set(FORMAT_VERSION, String.valueOf(formatVersion));
- updateProperties.set(DEFAULT_FILE_FORMAT, fileFormat.name());
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(FORMAT_VERSION, String.valueOf(formatVersion));
+ properties.put(DEFAULT_FILE_FORMAT, fileFormat.name());
switch (fileFormat) {
case PARQUET:
- updateProperties.set(PARQUET_VECTORIZATION_ENABLED, String.valueOf(vectorized));
+ properties.put(PARQUET_VECTORIZATION_ENABLED, String.valueOf(vectorized));
break;
case ORC:
- updateProperties.set(ORC_VECTORIZATION_ENABLED, String.valueOf(vectorized));
+ properties.put(ORC_VECTORIZATION_ENABLED, String.valueOf(vectorized));
break;
default:
Preconditions.checkState(
!vectorized, "File format %s does not support vectorized reads", fileFormat);
}
- updateProperties.commit();
+ this.table = TestTables.create(temp.newFolder(), TABLE_NAME, SCHEMA, SPEC, properties);
}
}
diff --git a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestTables.java b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestTables.java
index 0650cb9738..b54bb315c5 100644
--- a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestTables.java
+++ b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestTables.java
@@ -44,12 +44,16 @@ class TestTables {
private TestTables() {}
static TestTable create(File temp, String name, Schema schema, PartitionSpec spec) {
+ return create(temp, name, schema, spec, ImmutableMap.of());
+ }
+
+ static TestTable create(
+ File temp, String name, Schema schema, PartitionSpec spec, Map<String, String> properties) {
TestTableOperations ops = new TestTableOperations(name);
if (ops.current() != null) {
throw new AlreadyExistsException("Table %s already exists at location: %s", name, temp);
}
- ops.commit(
- null, TableMetadata.newTableMetadata(schema, spec, temp.toString(), ImmutableMap.of()));
+ ops.commit(null, TableMetadata.newTableMetadata(schema, spec, temp.toString(), properties));
return new TestTable(ops, name);
}
diff --git a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTableAsSelect.java b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTableAsSelect.java
index 2581c0fd3c..1c08b1fd5a 100644
--- a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTableAsSelect.java
+++ b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTableAsSelect.java
@@ -388,7 +388,6 @@ public class TestCreateTableAsSelect extends SparkCatalogTestBase {
PartitionSpec expectedSpec =
PartitionSpec.builderFor(expectedSchema)
- .alwaysNull("part", "part_1000")
.identity("part")
.identity("id")
.withSpecId(2) // The Spec is new
diff --git a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java
index a4928e6552..a43f2a041b 100644
--- a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java
+++ b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java
@@ -18,9 +18,11 @@
*/
package org.apache.iceberg.spark.extensions;
-import java.util.Map;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.PartitionSpecParser;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.spark.SparkCatalogConfig;
import org.apache.iceberg.spark.source.SparkTable;
import org.apache.spark.sql.connector.catalog.CatalogManager;
import org.apache.spark.sql.connector.catalog.Identifier;
@@ -28,11 +30,23 @@ import org.apache.spark.sql.connector.catalog.TableCatalog;
import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
+import org.junit.runners.Parameterized;
public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
- public TestAlterTablePartitionFields(
- String catalogName, String implementation, Map<String, String> config) {
- super(catalogName, implementation, config);
+
+ @Parameterized.Parameters(name = "catalogConfig = {0}, formatVersion = {1}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {SparkCatalogConfig.HIVE, 1},
+ {SparkCatalogConfig.SPARK, 2}
+ };
+ }
+
+ private final int formatVersion;
+
+ public TestAlterTablePartitionFields(SparkCatalogConfig catalogConfig, int formatVersion) {
+ super(catalogConfig.catalogName(), catalogConfig.implementation(), catalogConfig.properties());
+ this.formatVersion = formatVersion;
}
@After
@@ -42,9 +56,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddIdentityPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -61,9 +73,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddBucketPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -83,9 +93,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddTruncatePartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -105,9 +113,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddYearsPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -124,9 +130,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddMonthsPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -143,9 +147,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddDaysPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -162,9 +164,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddHoursPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -181,9 +181,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testAddNamedPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -200,9 +198,7 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
@Test
public void testDropIdentityPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg PARTITIONED BY (category)",
- tableName);
+ createTable("id bigint NOT NULL, category string, data string", "category");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertEquals(
@@ -212,20 +208,21 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
table.refresh();
- PartitionSpec expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(1)
- .alwaysNull("category", "category")
- .build();
-
- Assert.assertEquals("Should have new spec field", expected, table.spec());
+ if (formatVersion == 1) {
+ PartitionSpec expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(1)
+ .alwaysNull("category", "category")
+ .build();
+ Assert.assertEquals("Should have new spec field", expected, table.spec());
+ } else {
+ Assert.assertTrue("New spec must be unpartitioned", table.spec().isUnpartitioned());
+ }
}
@Test
public void testDropDaysPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, ts timestamp, data string) USING iceberg PARTITIONED BY (days(ts))",
- tableName);
+ createTable("id bigint NOT NULL, ts timestamp, data string", "days(ts)");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertEquals(
@@ -235,17 +232,18 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
table.refresh();
- PartitionSpec expected =
- PartitionSpec.builderFor(table.schema()).withSpecId(1).alwaysNull("ts", "ts_day").build();
-
- Assert.assertEquals("Should have new spec field", expected, table.spec());
+ if (formatVersion == 1) {
+ PartitionSpec expected =
+ PartitionSpec.builderFor(table.schema()).withSpecId(1).alwaysNull("ts", "ts_day").build();
+ Assert.assertEquals("Should have new spec field", expected, table.spec());
+ } else {
+ Assert.assertTrue("New spec must be unpartitioned", table.spec().isUnpartitioned());
+ }
}
@Test
public void testDropBucketPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg PARTITIONED BY (bucket(16, id))",
- tableName);
+ createTable("id bigint NOT NULL, data string", "bucket(16, id)");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertEquals(
@@ -255,20 +253,21 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
table.refresh();
- PartitionSpec expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(1)
- .alwaysNull("id", "id_bucket")
- .build();
-
- Assert.assertEquals("Should have new spec field", expected, table.spec());
+ if (formatVersion == 1) {
+ PartitionSpec expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(1)
+ .alwaysNull("id", "id_bucket")
+ .build();
+ Assert.assertEquals("Should have new spec field", expected, table.spec());
+ } else {
+ Assert.assertTrue("New spec must be unpartitioned", table.spec().isUnpartitioned());
+ }
}
@Test
public void testDropPartitionByName() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -284,17 +283,18 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
table.refresh();
- PartitionSpec expected =
- PartitionSpec.builderFor(table.schema()).withSpecId(2).alwaysNull("id", "shard").build();
-
- Assert.assertEquals("Should have new spec field", expected, table.spec());
+ if (formatVersion == 1) {
+ PartitionSpec expected =
+ PartitionSpec.builderFor(table.schema()).withSpecId(2).alwaysNull("id", "shard").build();
+ Assert.assertEquals("Should have new spec field", expected, table.spec());
+ } else {
+ Assert.assertTrue("New spec must be unpartitioned", table.spec().isUnpartitioned());
+ }
}
@Test
public void testReplacePartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -306,21 +306,34 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
sql("ALTER TABLE %s REPLACE PARTITION FIELD days(ts) WITH hours(ts)", tableName);
table.refresh();
- expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(2)
- .alwaysNull("ts", "ts_day")
- .hour("ts")
- .build();
+ if (formatVersion == 1) {
+ expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(2)
+ .alwaysNull("ts", "ts_day")
+ .hour("ts")
+ .build();
+ } else {
+ expected =
+ PartitionSpecParser.fromJson(
+ table.schema(),
+ "{\n"
+ + " \"spec-id\" : 2,\n"
+ + " \"fields\" : [ {\n"
+ + " \"name\" : \"ts_hour\",\n"
+ + " \"transform\" : \"hour\",\n"
+ + " \"source-id\" : 3,\n"
+ + " \"field-id\" : 1001\n"
+ + " } ]\n"
+ + "}");
+ }
Assert.assertEquals(
"Should changed from daily to hourly partitioned field", expected, table.spec());
}
@Test
public void testReplacePartitionAndRename() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -332,21 +345,34 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
sql("ALTER TABLE %s REPLACE PARTITION FIELD days(ts) WITH hours(ts) AS hour_col", tableName);
table.refresh();
- expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(2)
- .alwaysNull("ts", "ts_day")
- .hour("ts", "hour_col")
- .build();
+ if (formatVersion == 1) {
+ expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(2)
+ .alwaysNull("ts", "ts_day")
+ .hour("ts", "hour_col")
+ .build();
+ } else {
+ expected =
+ PartitionSpecParser.fromJson(
+ table.schema(),
+ "{\n"
+ + " \"spec-id\" : 2,\n"
+ + " \"fields\" : [ {\n"
+ + " \"name\" : \"hour_col\",\n"
+ + " \"transform\" : \"hour\",\n"
+ + " \"source-id\" : 3,\n"
+ + " \"field-id\" : 1001\n"
+ + " } ]\n"
+ + "}");
+ }
Assert.assertEquals(
"Should changed from daily to hourly partitioned field", expected, table.spec());
}
@Test
public void testReplaceNamedPartition() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -358,21 +384,34 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
sql("ALTER TABLE %s REPLACE PARTITION FIELD day_col WITH hours(ts)", tableName);
table.refresh();
- expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(2)
- .alwaysNull("ts", "day_col")
- .hour("ts")
- .build();
+ if (formatVersion == 1) {
+ expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(2)
+ .alwaysNull("ts", "day_col")
+ .hour("ts")
+ .build();
+ } else {
+ expected =
+ PartitionSpecParser.fromJson(
+ table.schema(),
+ "{\n"
+ + " \"spec-id\" : 2,\n"
+ + " \"fields\" : [ {\n"
+ + " \"name\" : \"ts_hour\",\n"
+ + " \"transform\" : \"hour\",\n"
+ + " \"source-id\" : 3,\n"
+ + " \"field-id\" : 1001\n"
+ + " } ]\n"
+ + "}");
+ }
Assert.assertEquals(
"Should changed from daily to hourly partitioned field", expected, table.spec());
}
@Test
public void testReplaceNamedPartitionAndRenameDifferently() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, ts timestamp, data string) USING iceberg",
- tableName);
+ createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);
Assert.assertTrue("Table should start unpartitioned", table.spec().isUnpartitioned());
@@ -384,19 +423,34 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
sql("ALTER TABLE %s REPLACE PARTITION FIELD day_col WITH hours(ts) AS hour_col", tableName);
table.refresh();
- expected =
- PartitionSpec.builderFor(table.schema())
- .withSpecId(2)
- .alwaysNull("ts", "day_col")
- .hour("ts", "hour_col")
- .build();
+ if (formatVersion == 1) {
+ expected =
+ PartitionSpec.builderFor(table.schema())
+ .withSpecId(2)
+ .alwaysNull("ts", "day_col")
+ .hour("ts", "hour_col")
+ .build();
+ } else {
+ expected =
+ PartitionSpecParser.fromJson(
+ table.schema(),
+ "{\n"
+ + " \"spec-id\" : 2,\n"
+ + " \"fields\" : [ {\n"
+ + " \"name\" : \"hour_col\",\n"
+ + " \"transform\" : \"hour\",\n"
+ + " \"source-id\" : 3,\n"
+ + " \"field-id\" : 1001\n"
+ + " } ]\n"
+ + "}");
+ }
Assert.assertEquals(
"Should changed from daily to hourly partitioned field", expected, table.spec());
}
@Test
public void testSparkTableAddDropPartitions() throws Exception {
- sql("CREATE TABLE %s (id bigint NOT NULL, ts timestamp, data string) USING iceberg", tableName);
+ createTable("id bigint NOT NULL, ts timestamp, data string");
Assert.assertEquals(
"spark table partition should be empty", 0, sparkTable().partitioning().length);
@@ -459,4 +513,20 @@ public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {
Identifier identifier = Identifier.of(tableIdent.namespace().levels(), tableIdent.name());
return (SparkTable) catalog.loadTable(identifier);
}
+
+ private void createTable(String schema) {
+ createTable(schema, null);
+ }
+
+ private void createTable(String schema, String spec) {
+ if (spec == null) {
+ sql(
+ "CREATE TABLE %s (%s) USING iceberg TBLPROPERTIES ('%s' '%d')",
+ tableName, schema, TableProperties.FORMAT_VERSION, formatVersion);
+ } else {
+ sql(
+ "CREATE TABLE %s (%s) USING iceberg PARTITIONED BY (%s) TBLPROPERTIES ('%s' '%d')",
+ tableName, schema, spec, TableProperties.FORMAT_VERSION, formatVersion);
+ }
+ }
}
diff --git a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java
index 0a4a1073c3..015f2ad5fd 100644
--- a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java
+++ b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java
@@ -49,19 +49,16 @@ public class TestCreateChangelogViewProcedure extends SparkExtensionsTestBase {
public void createTableWithTwoColumns() {
sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName);
- sql("ALTER TABLE %s SET TBLPROPERTIES ('format-version'='%d')", tableName, 1);
sql("ALTER TABLE %s ADD PARTITION FIELD data", tableName);
}
private void createTableWithThreeColumns() {
sql("CREATE TABLE %s (id INT, data STRING, age INT) USING iceberg", tableName);
- sql("ALTER TABLE %s SET TBLPROPERTIES ('format-version'='%d')", tableName, 1);
sql("ALTER TABLE %s ADD PARTITION FIELD id", tableName);
}
private void createTableWithIdentifierField() {
sql("CREATE TABLE %s (id INT NOT NULL, data STRING) USING iceberg", tableName);
- sql("ALTER TABLE %s SET TBLPROPERTIES ('format-version'='%d')", tableName, 1);
sql("ALTER TABLE %s SET IDENTIFIER FIELDS id", tableName);
}
diff --git a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
index 744c47f3d5..1c7f533d86 100644
--- a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
+++ b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
@@ -1187,10 +1187,7 @@ public abstract class TestDelete extends SparkRowLevelOperationsTestBase {
Snapshot currentSnapshot = SnapshotUtil.latestSnapshot(table, branch);
if (mode(table) == COPY_ON_WRITE) {
- // copy-on-write is tested against v1 and such tables have different partition evolution
- // behavior
- // that's why the number of changed partitions is 4 for copy-on-write
- validateCopyOnWrite(currentSnapshot, "4", "4", "1");
+ validateCopyOnWrite(currentSnapshot, "3", "4", "1");
} else {
validateMergeOnRead(currentSnapshot, "3", "3", null);
}
diff --git a/spark/v3.4/spark-runtime/src/integration/java/org/apache/iceberg/spark/SmokeTest.java b/spark/v3.4/spark-runtime/src/integration/java/org/apache/iceberg/spark/SmokeTest.java
index 510108af8c..d6fab897d6 100644
--- a/spark/v3.4/spark-runtime/src/integration/java/org/apache/iceberg/spark/SmokeTest.java
+++ b/spark/v3.4/spark-runtime/src/integration/java/org/apache/iceberg/spark/SmokeTest.java
@@ -117,11 +117,7 @@ public class SmokeTest extends SparkExtensionsTestBase {
sql("ALTER TABLE %s DROP PARTITION FIELD shard", tableName);
table = getTable();
- Assert.assertEquals("Table should have 4 partition fields", 4, table.spec().fields().size());
- // VoidTransform is package private so we can't reach it here, just checking name
- Assert.assertTrue(
- "All transforms should be void",
- table.spec().fields().stream().allMatch(pf -> pf.transform().toString().equals("void")));
+ Assert.assertTrue("Table should be unpartitioned", table.spec().isUnpartitioned());
// Sort order examples
sql("ALTER TABLE %s WRITE ORDERED BY category, id", tableName);
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestCreateActions.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestCreateActions.java
index 96950e8227..3dccd79527 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestCreateActions.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestCreateActions.java
@@ -33,6 +33,7 @@ import org.apache.commons.io.filefilter.TrueFileFilter;
import org.apache.hadoop.fs.Path;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
import org.apache.iceberg.actions.MigrateTable;
import org.apache.iceberg.actions.SnapshotTable;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -531,7 +532,9 @@ public class TestCreateActions extends SparkCatalogTestBase {
String source = sourceName("test_reserved_properties_table");
String dest = destName(destTableName);
createSourceTable(CREATE_PARQUET, source);
- assertSnapshotFileCount(SparkActions.get().snapshotTable(source).as(dest), source, dest);
+ SnapshotTableSparkAction action = SparkActions.get().snapshotTable(source).as(dest);
+ action.tableProperty(TableProperties.FORMAT_VERSION, "1");
+ assertSnapshotFileCount(action, source, dest);
SparkTable table = loadTable(dest);
// set sort orders
table.table().replaceSortOrder().asc("id").desc("data").commit();
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestDeleteReachableFilesAction.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestDeleteReachableFilesAction.java
index 18de71ce8e..3844052655 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestDeleteReachableFilesAction.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestDeleteReachableFilesAction.java
@@ -267,8 +267,6 @@ public class TestDeleteReachableFilesAction extends SparkTestBase {
@Test
public void testPositionDeleteFiles() {
- table.updateProperties().set(TableProperties.FORMAT_VERSION, "2").commit();
-
table.newAppend().appendFile(FILE_A).commit();
table.newAppend().appendFile(FILE_B).commit();
@@ -277,13 +275,11 @@ public class TestDeleteReachableFilesAction extends SparkTestBase {
DeleteReachableFiles baseRemoveFilesSparkAction =
sparkActions().deleteReachableFiles(metadataLocation(table)).io(table.io());
- checkRemoveFilesResults(2, 1, 0, 3, 3, 6, baseRemoveFilesSparkAction.execute());
+ checkRemoveFilesResults(2, 1, 0, 3, 3, 5, baseRemoveFilesSparkAction.execute());
}
@Test
public void testEqualityDeleteFiles() {
- table.updateProperties().set(TableProperties.FORMAT_VERSION, "2").commit();
-
table.newAppend().appendFile(FILE_A).commit();
table.newAppend().appendFile(FILE_B).commit();
@@ -292,7 +288,7 @@ public class TestDeleteReachableFilesAction extends SparkTestBase {
DeleteReachableFiles baseRemoveFilesSparkAction =
sparkActions().deleteReachableFiles(metadataLocation(table)).io(table.io());
- checkRemoveFilesResults(2, 0, 1, 3, 3, 6, baseRemoveFilesSparkAction.execute());
+ checkRemoveFilesResults(2, 0, 1, 3, 3, 5, baseRemoveFilesSparkAction.execute());
}
@Test
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
index ae6a204d2e..1581e58364 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
@@ -83,6 +83,7 @@ import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -381,7 +382,8 @@ public class TestRewriteDataFilesAction extends SparkTestBase {
@Test
public void testBinPackWithStartingSequenceNumberV1Compatibility() {
- Table table = createTablePartitioned(4, 2);
+ Map<String, String> properties = ImmutableMap.of(TableProperties.FORMAT_VERSION, "1");
+ Table table = createTablePartitioned(4, 2, SCALE, properties);
shouldHaveFiles(table, 8);
List<Object[]> expectedRecords = currentData();
table.refresh();
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java
index 9bd7220b90..746415818c 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java
@@ -19,6 +19,7 @@
package org.apache.iceberg.spark.source;
import java.io.File;
+import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
@@ -43,11 +44,9 @@ public class TestIcebergSourceHadoopTables extends TestIcebergSourceTablesBase {
}
@Override
- public Table createTable(TableIdentifier ident, Schema schema, PartitionSpec spec) {
- if (spec.equals(PartitionSpec.unpartitioned())) {
- return TABLES.create(schema, tableLocation);
- }
- return TABLES.create(schema, spec, tableLocation);
+ public Table createTable(
+ TableIdentifier ident, Schema schema, PartitionSpec spec, Map<String, String> properties) {
+ return TABLES.create(schema, spec, properties, tableLocation);
}
@Override
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHiveTables.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHiveTables.java
index 6292a2c1a8..2a264b74b0 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHiveTables.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHiveTables.java
@@ -19,6 +19,7 @@
package org.apache.iceberg.spark.source;
import java.io.IOException;
+import java.util.Map;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.iceberg.PartitionSpec;
@@ -51,9 +52,10 @@ public class TestIcebergSourceHiveTables extends TestIcebergSourceTablesBase {
}
@Override
- public Table createTable(TableIdentifier ident, Schema schema, PartitionSpec spec) {
+ public Table createTable(
+ TableIdentifier ident, Schema schema, PartitionSpec spec, Map<String, String> properties) {
TestIcebergSourceHiveTables.currentIdentifier = ident;
- return TestIcebergSourceHiveTables.catalog.createTable(ident, schema, spec);
+ return TestIcebergSourceHiveTables.catalog.createTable(ident, schema, spec, properties);
}
@Override
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java
index 61bed7464b..486713e52e 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java
@@ -31,6 +31,7 @@ import java.time.ZoneOffset;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
+import java.util.Map;
import java.util.StringJoiner;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -118,7 +119,8 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
@Rule public TemporaryFolder temp = new TemporaryFolder();
- public abstract Table createTable(TableIdentifier ident, Schema schema, PartitionSpec spec);
+ public abstract Table createTable(
+ TableIdentifier ident, Schema schema, PartitionSpec spec, Map<String, String> properties);
public abstract Table loadTable(TableIdentifier ident, String entriesSuffix);
@@ -133,6 +135,10 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
spark.sql("DROP TABLE IF EXISTS parquet_table");
}
+ private Table createTable(TableIdentifier ident, Schema schema, PartitionSpec spec) {
+ return createTable(ident, schema, spec, ImmutableMap.of());
+ }
+
@Test
public synchronized void testTablesSupport() {
TableIdentifier tableIdentifier = TableIdentifier.of("db", "table");
@@ -191,8 +197,8 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
// each row must inherit snapshot_id and sequence_number
rows.forEach(
row -> {
- row.put(2, 0L); // data sequence number
- row.put(3, 0L); // file sequence number
+ row.put(2, 1L); // data sequence number
+ row.put(3, 1L); // file sequence number
GenericData.Record file = (GenericData.Record) row.get("data_file");
TestHelpers.asMetadataRecord(file);
expected.add(row);
@@ -386,8 +392,13 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
// each row must inherit snapshot_id and sequence_number
rows.forEach(
row -> {
- row.put(2, 0L); // data sequence number
- row.put(3, 0L); // file sequence number
+ if (row.get("snapshot_id").equals(table.currentSnapshot().snapshotId())) {
+ row.put(2, 3L); // data sequence number
+ row.put(3, 3L); // file sequence number
+ } else {
+ row.put(2, 1L); // data sequence number
+ row.put(3, 1L); // file sequence number
+ }
GenericData.Record file = (GenericData.Record) row.get("data_file");
TestHelpers.asMetadataRecord(file);
expected.add(row);
@@ -544,10 +555,10 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
}
@Test
- public void testEntriesTableWithSnapshotIdInheritance() throws Exception {
+ public void testV1EntriesTableWithSnapshotIdInheritance() throws Exception {
TableIdentifier tableIdentifier = TableIdentifier.of("db", "entries_inheritance_test");
- PartitionSpec spec = SPEC;
- Table table = createTable(tableIdentifier, SCHEMA, spec);
+ Map<String, String> properties = ImmutableMap.of(TableProperties.FORMAT_VERSION, "1");
+ Table table = createTable(tableIdentifier, SCHEMA, SPEC, properties);
table.updateProperties().set(TableProperties.SNAPSHOT_ID_INHERITANCE_ENABLED, "true").commit();
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java
index 1b96245019..b4b60dc02d 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java
@@ -28,6 +28,7 @@ import static org.apache.iceberg.MetadataTableType.FILES;
import static org.apache.iceberg.MetadataTableType.PARTITIONS;
import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
import static org.apache.iceberg.TableProperties.FORMAT_VERSION;
+import static org.apache.iceberg.TableProperties.MANIFEST_MERGE_ENABLED;
import java.util.Arrays;
import java.util.List;
@@ -139,10 +140,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testFilesMetadataTable() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'a1', 'b1')", tableName);
@@ -203,11 +201,8 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testFilesMetadataTableFilter() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg "
- + "TBLPROPERTIES ('commit.manifest-merge.enabled' 'false')",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
+ sql("ALTER TABLE %s SET TBLPROPERTIES ('%s' 'false')", tableName, MANIFEST_MERGE_ENABLED);
sql("INSERT INTO TABLE %s VALUES (1, 'c1', 'd1')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'c2', 'd2')", tableName);
@@ -297,10 +292,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testEntriesMetadataTable() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'a1', 'b1')", tableName);
@@ -361,10 +353,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testPartitionsTableAddRemoveFields() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg ",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'c1', 'd1')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'c2', 'd2')", tableName);
@@ -419,10 +408,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testPartitionsTableRenameFields() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
Table table = validationCatalog.loadTable(tableIdent);
@@ -449,10 +435,8 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testPartitionsTableSwitchFields() throws Exception {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
+
Table table = validationCatalog.loadTable(tableIdent);
// verify the metadata tables after re-adding the first dropped column in the second location
@@ -514,10 +498,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testPartitionTableFilterAddRemoveFields() throws ParseException {
// Create un-partitioned table
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'c1', 'd1')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'c2', 'd2')", tableName);
@@ -576,10 +557,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
// In V2, re-added field currently conflicts with its deleted form
Assume.assumeTrue(formatVersion == 1);
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'c1', 'd1')", tableName);
sql("INSERT INTO TABLE %s VALUES (2, 'c2', 'd2')", tableName);
Table table = validationCatalog.loadTable(tableIdent);
@@ -619,10 +597,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testPartitionsTableFilterRenameFields() throws ParseException {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
Table table = validationCatalog.loadTable(tableIdent);
@@ -645,10 +620,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
@Test
public void testMetadataTablesWithUnknownTransforms() {
- sql(
- "CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg",
- tableName);
- initTable();
+ createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'a1', 'b1')", tableName);
@@ -748,10 +720,9 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
return spark.read().format("iceberg").load(tableName + "." + tableType.name());
}
- private void initTable() {
+ private void createTable(String schema) {
sql(
- "ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
- tableName, DEFAULT_FILE_FORMAT, fileFormat.name());
- sql("ALTER TABLE %s SET TBLPROPERTIES('%s' '%d')", tableName, FORMAT_VERSION, formatVersion);
+ "CREATE TABLE %s (%s) USING iceberg TBLPROPERTIES ('%s' '%s', '%s' '%d')",
+ tableName, schema, DEFAULT_FILE_FORMAT, fileFormat.name(), FORMAT_VERSION, formatVersion);
}
}
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java
index 1b46759590..8a2fae8117 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java
@@ -28,6 +28,7 @@ import static org.apache.spark.sql.functions.lit;
import java.io.IOException;
import java.util.List;
+import java.util.Map;
import java.util.stream.Collectors;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.HasTableOperations;
@@ -38,13 +39,13 @@ import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.TableOperations;
-import org.apache.iceberg.UpdateProperties;
import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.spark.SparkTestBase;
import org.apache.iceberg.types.Types;
import org.apache.spark.sql.Dataset;
@@ -72,6 +73,7 @@ public class TestSparkMetadataColumns extends SparkTestBase {
Types.NestedField.optional(1, "id", Types.LongType.get()),
Types.NestedField.optional(2, "category", Types.StringType.get()),
Types.NestedField.optional(3, "data", Types.StringType.get()));
+ private static final PartitionSpec SPEC = PartitionSpec.unpartitioned();
private static final PartitionSpec UNKNOWN_SPEC =
PartitionSpecParser.fromJson(
SCHEMA,
@@ -261,25 +263,22 @@ public class TestSparkMetadataColumns extends SparkTestBase {
}
private void createAndInitTable() throws IOException {
- this.table =
- TestTables.create(temp.newFolder(), TABLE_NAME, SCHEMA, PartitionSpec.unpartitioned());
-
- UpdateProperties updateProperties = table.updateProperties();
- updateProperties.set(FORMAT_VERSION, String.valueOf(formatVersion));
- updateProperties.set(DEFAULT_FILE_FORMAT, fileFormat.name());
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(FORMAT_VERSION, String.valueOf(formatVersion));
+ properties.put(DEFAULT_FILE_FORMAT, fileFormat.name());
switch (fileFormat) {
case PARQUET:
- updateProperties.set(PARQUET_VECTORIZATION_ENABLED, String.valueOf(vectorized));
+ properties.put(PARQUET_VECTORIZATION_ENABLED, String.valueOf(vectorized));
break;
case ORC:
- updateProperties.set(ORC_VECTORIZATION_ENABLED, String.valueOf(vectorized));
+ properties.put(ORC_VECTORIZATION_ENABLED, String.valueOf(vectorized));
break;
default:
Preconditions.checkState(
!vectorized, "File format %s does not support vectorized reads", fileFormat);
}
- updateProperties.commit();
+ this.table = TestTables.create(temp.newFolder(), TABLE_NAME, SCHEMA, SPEC, properties);
}
}
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestTables.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestTables.java
index 0650cb9738..b54bb315c5 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestTables.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestTables.java
@@ -44,12 +44,16 @@ class TestTables {
private TestTables() {}
static TestTable create(File temp, String name, Schema schema, PartitionSpec spec) {
+ return create(temp, name, schema, spec, ImmutableMap.of());
+ }
+
+ static TestTable create(
+ File temp, String name, Schema schema, PartitionSpec spec, Map<String, String> properties) {
TestTableOperations ops = new TestTableOperations(name);
if (ops.current() != null) {
throw new AlreadyExistsException("Table %s already exists at location: %s", name, temp);
}
- ops.commit(
- null, TableMetadata.newTableMetadata(schema, spec, temp.toString(), ImmutableMap.of()));
+ ops.commit(null, TableMetadata.newTableMetadata(schema, spec, temp.toString(), properties));
return new TestTable(ops, name);
}
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTableAsSelect.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTableAsSelect.java
index 2581c0fd3c..1c08b1fd5a 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTableAsSelect.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTableAsSelect.java
@@ -388,7 +388,6 @@ public class TestCreateTableAsSelect extends SparkCatalogTestBase {
PartitionSpec expectedSpec =
PartitionSpec.builderFor(expectedSchema)
- .alwaysNull("part", "part_1000")
.identity("part")
.identity("id")
.withSpecId(2) // The Spec is new