You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by et...@apache.org on 2025/05/08 05:31:34 UTC
(iceberg) branch main updated: Spark 3.4: Migrate source and spark tests to JUnit5 (#12998)
This is an automated email from the ASF dual-hosted git repository.
etudenhoefner pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new ada3d12b4e Spark 3.4: Migrate source and spark tests to JUnit5 (#12998)
ada3d12b4e is described below
commit ada3d12b4e02f8c9579c708fffb4711c52bc434c
Author: Tom Tanaka <43...@users.noreply.github.com>
AuthorDate: Thu May 8 14:31:29 2025 +0900
Spark 3.4: Migrate source and spark tests to JUnit5 (#12998)
---
.../iceberg/spark/TestFileRewriteCoordinator.java | 53 +++--
.../iceberg/spark/TestSparkCachedTableCatalog.java | 29 ++-
.../iceberg/spark/TestSparkCatalogOperations.java | 52 ++--
.../apache/iceberg/spark/TestSparkWriteConf.java | 65 ++---
.../spark/actions/TestMigrateTableAction.java | 30 ++-
.../spark/source/TestDataSourceOptions.java | 92 ++++----
.../source/TestIcebergSourceHadoopTables.java | 13 +-
.../spark/source/TestIcebergSourceHiveTables.java | 8 +-
.../spark/source/TestIcebergSourceTablesBase.java | 261 +++++++++++----------
.../source/TestMetadataTableReadableMetrics.java | 45 ++--
.../TestMetadataTablesWithPartitionEvolution.java | 89 ++++---
.../iceberg/spark/TestFileRewriteCoordinator.java | 24 +-
.../iceberg/spark/TestSparkCachedTableCatalog.java | 3 +
.../iceberg/spark/TestSparkCatalogOperations.java | 8 +-
.../apache/iceberg/spark/TestSparkWriteConf.java | 3 +
.../spark/source/TestDataSourceOptions.java | 13 +-
.../source/TestIcebergSourceHadoopTables.java | 2 +-
.../spark/source/TestIcebergSourceTablesBase.java | 71 +++---
.../source/TestMetadataTableReadableMetrics.java | 7 +-
19 files changed, 442 insertions(+), 426 deletions(-)
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestFileRewriteCoordinator.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestFileRewriteCoordinator.java
index 20a2d5c93a..ae94f35019 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestFileRewriteCoordinator.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestFileRewriteCoordinator.java
@@ -18,6 +18,8 @@
*/
package org.apache.iceberg.spark;
+import static org.assertj.core.api.Assertions.assertThat;
+
import java.io.IOException;
import java.util.List;
import java.util.Map;
@@ -26,34 +28,30 @@ import java.util.UUID;
import java.util.stream.Collectors;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Table;
import org.apache.iceberg.io.CloseableIterable;
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;
import org.apache.iceberg.spark.source.SimpleRecord;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Test;
-
-public class TestFileRewriteCoordinator extends SparkCatalogTestBase {
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
- public TestFileRewriteCoordinator(
- String catalogName, String implementation, Map<String, String> config) {
- super(catalogName, implementation, config);
- }
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestFileRewriteCoordinator extends CatalogTestBase {
- @After
+ @AfterEach
public void removeTables() {
sql("DROP TABLE IF EXISTS %s", tableName);
}
- @Test
+ @TestTemplate
public void testBinPackRewrite() throws NoSuchTableException, IOException {
sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName);
@@ -64,7 +62,7 @@ public class TestFileRewriteCoordinator extends SparkCatalogTestBase {
df.coalesce(1).writeTo(tableName).append();
Table table = validationCatalog.loadTable(tableIdent);
- Assert.assertEquals("Should produce 4 snapshots", 4, Iterables.size(table.snapshots()));
+ assertThat(table.snapshots()).as("Should produce 4 snapshots").hasSize(4);
Dataset<Row> fileDF =
spark.read().format("iceberg").load(tableName(tableIdent.name() + ".files"));
@@ -106,14 +104,15 @@ public class TestFileRewriteCoordinator extends SparkCatalogTestBase {
table.refresh();
Map<String, String> summary = table.currentSnapshot().summary();
- Assert.assertEquals("Deleted files count must match", "4", summary.get("deleted-data-files"));
- Assert.assertEquals("Added files count must match", "2", summary.get("added-data-files"));
+ assertThat(summary)
+ .containsEntry("deleted-data-files", "4")
+ .containsEntry("added-data-files", "2");
Object rowCount = scalarSql("SELECT count(*) FROM %s", tableName);
- Assert.assertEquals("Row count must match", 4000L, rowCount);
+ assertThat(rowCount).as("Row count must match").isEqualTo(4000L);
}
- @Test
+ @TestTemplate
public void testSortRewrite() throws NoSuchTableException, IOException {
sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName);
@@ -124,7 +123,7 @@ public class TestFileRewriteCoordinator extends SparkCatalogTestBase {
df.coalesce(1).writeTo(tableName).append();
Table table = validationCatalog.loadTable(tableIdent);
- Assert.assertEquals("Should produce 4 snapshots", 4, Iterables.size(table.snapshots()));
+ assertThat(table.snapshots()).as("Should produce 4 snapshots").hasSize(4);
try (CloseableIterable<FileScanTask> fileScanTasks = table.newScan().planFiles()) {
String fileSetID = UUID.randomUUID().toString();
@@ -176,14 +175,15 @@ public class TestFileRewriteCoordinator extends SparkCatalogTestBase {
table.refresh();
Map<String, String> summary = table.currentSnapshot().summary();
- Assert.assertEquals("Deleted files count must match", "4", summary.get("deleted-data-files"));
- Assert.assertEquals("Added files count must match", "2", summary.get("added-data-files"));
+ assertThat(summary)
+ .containsEntry("deleted-data-files", "4")
+ .containsEntry("added-data-files", "2");
Object rowCount = scalarSql("SELECT count(*) FROM %s", tableName);
- Assert.assertEquals("Row count must match", 4000L, rowCount);
+ assertThat(rowCount).as("Row count must match").isEqualTo(4000L);
}
- @Test
+ @TestTemplate
public void testCommitMultipleRewrites() throws NoSuchTableException, IOException {
sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName);
@@ -253,14 +253,15 @@ public class TestFileRewriteCoordinator extends SparkCatalogTestBase {
table.refresh();
- Assert.assertEquals("Should produce 5 snapshots", 5, Iterables.size(table.snapshots()));
+ assertThat(table.snapshots()).as("Should produce 5 snapshots").hasSize(5);
Map<String, String> summary = table.currentSnapshot().summary();
- Assert.assertEquals("Deleted files count must match", "4", summary.get("deleted-data-files"));
- Assert.assertEquals("Added files count must match", "2", summary.get("added-data-files"));
+ assertThat(summary)
+ .containsEntry("deleted-data-files", "4")
+ .containsEntry("added-data-files", "2");
Object rowCount = scalarSql("SELECT count(*) FROM %s", tableName);
- Assert.assertEquals("Row count must match", 4000L, rowCount);
+ assertThat(rowCount).as("Row count must match").isEqualTo(4000L);
}
private Dataset<Row> newDF(int numRecords) {
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestSparkCachedTableCatalog.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestSparkCachedTableCatalog.java
index 23e8717fb8..228bf43b89 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestSparkCachedTableCatalog.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestSparkCachedTableCatalog.java
@@ -18,32 +18,43 @@
*/
package org.apache.iceberg.spark;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
-public class TestSparkCachedTableCatalog extends SparkTestBaseWithCatalog {
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestSparkCachedTableCatalog extends TestBaseWithCatalog {
private static final SparkTableCache TABLE_CACHE = SparkTableCache.get();
- @BeforeClass
+ @BeforeAll
public static void setupCachedTableCatalog() {
spark.conf().set("spark.sql.catalog.testcache", SparkCachedTableCatalog.class.getName());
}
- @AfterClass
+ @AfterAll
public static void unsetCachedTableCatalog() {
spark.conf().unset("spark.sql.catalog.testcache");
}
- public TestSparkCachedTableCatalog() {
- super(SparkCatalogConfig.HIVE);
+ @Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ protected static Object[][] parameters() {
+ return new Object[][] {
+ {
+ SparkCatalogConfig.HIVE.catalogName(),
+ SparkCatalogConfig.HIVE.implementation(),
+ SparkCatalogConfig.HIVE.properties()
+ },
+ };
}
- @Test
+ @TestTemplate
public void testTimeTravel() {
sql("CREATE TABLE %s (id INT, dep STRING) USING iceberg", tableName);
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestSparkCatalogOperations.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestSparkCatalogOperations.java
index 0836271a7c..8c76cbc3e8 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestSparkCatalogOperations.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestSparkCatalogOperations.java
@@ -18,37 +18,37 @@
*/
package org.apache.iceberg.spark;
-import java.util.Map;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.atIndex;
+
+import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Schema;
import org.apache.iceberg.catalog.Catalog;
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.connector.catalog.Column;
import org.apache.spark.sql.connector.catalog.Identifier;
import org.apache.spark.sql.connector.catalog.Table;
import org.apache.spark.sql.connector.catalog.TableChange;
import org.apache.spark.sql.types.DataTypes;
-import org.apache.spark.sql.types.StructField;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
-public class TestSparkCatalogOperations extends SparkCatalogTestBase {
- public TestSparkCatalogOperations(
- String catalogName, String implementation, Map<String, String> config) {
- super(catalogName, implementation, config);
- }
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestSparkCatalogOperations extends CatalogTestBase {
- @Before
+ @BeforeEach
public void createTable() {
sql("CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg", tableName);
}
- @After
+ @AfterEach
public void removeTable() {
sql("DROP TABLE IF EXISTS %s", tableName);
}
- @Test
+ @TestTemplate
public void testAlterTable() throws NoSuchTableException {
BaseCatalog catalog = (BaseCatalog) spark.sessionState().catalogManager().catalog(catalogName);
Identifier identifier = Identifier.of(tableIdent.namespace().levels(), tableIdent.name());
@@ -62,24 +62,20 @@ public class TestSparkCatalogOperations extends SparkCatalogTestBase {
TableChange.addColumn(new String[] {fieldName}, DataTypes.StringType, true),
TableChange.setProperty(propsKey, propsValue));
- Assert.assertNotNull("Should return updated table", table);
+ assertThat(table).as("Should return updated table").isNotNull();
- StructField expectedField = DataTypes.createStructField(fieldName, DataTypes.StringType, true);
- Assert.assertEquals(
- "Adding a column to a table should return the updated table with the new column",
- table.schema().fields()[2],
- expectedField);
+ Column expectedField = Column.create(fieldName, DataTypes.StringType, true);
+ assertThat(table.columns())
+ .as("Adding a column to a table should return the updated table with the new column")
+ .contains(expectedField, atIndex(2));
- Assert.assertTrue(
- "Adding a property to a table should return the updated table with the new property",
- table.properties().containsKey(propsKey));
- Assert.assertEquals(
- "Altering a table to add a new property should add the correct value",
- propsValue,
- table.properties().get(propsKey));
+ assertThat(table.properties())
+ .as(
+ "Adding a property to a table should return the updated table with the new property with the new correct value")
+ .containsEntry(propsKey, propsValue);
}
- @Test
+ @TestTemplate
public void testInvalidateTable() {
// load table to CachingCatalog
sql("SELECT count(1) FROM %s", tableName);
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java
index 00faecf9a8..ae7ed3af65 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java
@@ -51,21 +51,24 @@ import java.time.Duration;
import java.util.List;
import java.util.Map;
import org.apache.iceberg.DistributionMode;
+import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.UpdateProperties;
import org.apache.iceberg.deletes.DeleteGranularity;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
-public class TestSparkWriteConf extends SparkTestBaseWithCatalog {
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestSparkWriteConf extends TestBaseWithCatalog {
- @Before
+ @BeforeEach
public void before() {
+ super.before();
sql(
"CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) "
+ "USING iceberg "
@@ -73,12 +76,12 @@ public class TestSparkWriteConf extends SparkTestBaseWithCatalog {
tableName);
}
- @After
+ @AfterEach
public void after() {
sql("DROP TABLE IF EXISTS %s", tableName);
}
- @Test
+ @TestTemplate
public void testDurationConf() {
Table table = validationCatalog.loadTable(tableIdent);
String confName = "spark.sql.iceberg.some-duration-conf";
@@ -100,7 +103,7 @@ public class TestSparkWriteConf extends SparkTestBaseWithCatalog {
});
}
- @Test
+ @TestTemplate
public void testDeleteGranularityDefault() {
Table table = validationCatalog.loadTable(tableIdent);
SparkWriteConf writeConf = new SparkWriteConf(spark, table, ImmutableMap.of());
@@ -109,7 +112,7 @@ public class TestSparkWriteConf extends SparkTestBaseWithCatalog {
assertThat(value).isEqualTo(DeleteGranularity.PARTITION);
}
- @Test
+ @TestTemplate
public void testDeleteGranularityTableProperty() {
Table table = validationCatalog.loadTable(tableIdent);
@@ -124,7 +127,7 @@ public class TestSparkWriteConf extends SparkTestBaseWithCatalog {
assertThat(value).isEqualTo(DeleteGranularity.FILE);
}
- @Test
+ @TestTemplate
public void testDeleteGranularityWriteOption() {
Table table = validationCatalog.loadTable(tableIdent);
@@ -142,7 +145,7 @@ public class TestSparkWriteConf extends SparkTestBaseWithCatalog {
assertThat(value).isEqualTo(DeleteGranularity.FILE);
}
- @Test
+ @TestTemplate
public void testDeleteGranularityInvalidValue() {
Table table = validationCatalog.loadTable(tableIdent);
@@ -155,7 +158,7 @@ public class TestSparkWriteConf extends SparkTestBaseWithCatalog {
.hasMessageContaining("Unknown delete granularity");
}
- @Test
+ @TestTemplate
public void testSparkWriteConfDistributionDefault() {
Table table = validationCatalog.loadTable(tableIdent);
@@ -164,7 +167,7 @@ public class TestSparkWriteConf extends SparkTestBaseWithCatalog {
checkMode(DistributionMode.HASH, writeConf);
}
- @Test
+ @TestTemplate
public void testSparkWriteConfDistributionModeWithWriteOption() {
Table table = validationCatalog.loadTable(tableIdent);
@@ -175,7 +178,7 @@ public class TestSparkWriteConf extends SparkTestBaseWithCatalog {
checkMode(DistributionMode.NONE, writeConf);
}
- @Test
+ @TestTemplate
public void testSparkWriteConfDistributionModeWithSessionConfig() {
withSQLConf(
ImmutableMap.of(SparkSQLProperties.DISTRIBUTION_MODE, DistributionMode.NONE.modeName()),
@@ -186,7 +189,7 @@ public class TestSparkWriteConf extends SparkTestBaseWithCatalog {
});
}
- @Test
+ @TestTemplate
public void testSparkWriteConfDistributionModeWithTableProperties() {
Table table = validationCatalog.loadTable(tableIdent);
@@ -202,7 +205,7 @@ public class TestSparkWriteConf extends SparkTestBaseWithCatalog {
checkMode(DistributionMode.NONE, writeConf);
}
- @Test
+ @TestTemplate
public void testSparkWriteConfDistributionModeWithTblPropAndSessionConfig() {
withSQLConf(
ImmutableMap.of(SparkSQLProperties.DISTRIBUTION_MODE, DistributionMode.NONE.modeName()),
@@ -223,7 +226,7 @@ public class TestSparkWriteConf extends SparkTestBaseWithCatalog {
});
}
- @Test
+ @TestTemplate
public void testSparkWriteConfDistributionModeWithWriteOptionAndSessionConfig() {
withSQLConf(
ImmutableMap.of(SparkSQLProperties.DISTRIBUTION_MODE, DistributionMode.RANGE.modeName()),
@@ -240,7 +243,7 @@ public class TestSparkWriteConf extends SparkTestBaseWithCatalog {
});
}
- @Test
+ @TestTemplate
public void testSparkWriteConfDistributionModeWithEverything() {
withSQLConf(
ImmutableMap.of(SparkSQLProperties.DISTRIBUTION_MODE, DistributionMode.RANGE.modeName()),
@@ -265,7 +268,7 @@ public class TestSparkWriteConf extends SparkTestBaseWithCatalog {
});
}
- @Test
+ @TestTemplate
public void testSparkConfOverride() {
List<List<Map<String, String>>> propertiesSuites =
Lists.newArrayList(
@@ -338,7 +341,7 @@ public class TestSparkWriteConf extends SparkTestBaseWithCatalog {
}
}
- @Test
+ @TestTemplate
public void testDataPropsDefaultsAsDeleteProps() {
List<List<Map<String, String>>> propertiesSuites =
Lists.newArrayList(
@@ -407,7 +410,7 @@ public class TestSparkWriteConf extends SparkTestBaseWithCatalog {
}
}
- @Test
+ @TestTemplate
public void testDeleteFileWriteConf() {
List<List<Map<String, String>>> propertiesSuites =
Lists.newArrayList(
@@ -502,9 +505,9 @@ public class TestSparkWriteConf extends SparkTestBaseWithCatalog {
SparkWriteConf writeConf = new SparkWriteConf(spark, table, ImmutableMap.of());
Map<String, String> writeProperties = writeConf.writeProperties();
Map<String, String> expectedProperties = propertiesSuite.get(2);
- Assert.assertEquals(expectedProperties.size(), writeConf.writeProperties().size());
+ assertThat(writeConf.writeProperties()).hasSameSizeAs(expectedProperties);
for (Map.Entry<String, String> entry : writeProperties.entrySet()) {
- Assert.assertEquals(entry.getValue(), expectedProperties.get(entry.getKey()));
+ assertThat(expectedProperties).containsEntry(entry.getKey(), entry.getValue());
}
table.refresh();
@@ -518,12 +521,12 @@ public class TestSparkWriteConf extends SparkTestBaseWithCatalog {
}
private void checkMode(DistributionMode expectedMode, SparkWriteConf writeConf) {
- Assert.assertEquals(expectedMode, writeConf.distributionMode());
- Assert.assertEquals(expectedMode, writeConf.copyOnWriteDistributionMode(DELETE));
- Assert.assertEquals(expectedMode, writeConf.positionDeltaDistributionMode(DELETE));
- Assert.assertEquals(expectedMode, writeConf.copyOnWriteDistributionMode(UPDATE));
- Assert.assertEquals(expectedMode, writeConf.positionDeltaDistributionMode(UPDATE));
- Assert.assertEquals(expectedMode, writeConf.copyOnWriteDistributionMode(MERGE));
- Assert.assertEquals(expectedMode, writeConf.positionDeltaDistributionMode(MERGE));
+ assertThat(writeConf.distributionMode()).isEqualTo(expectedMode);
+ assertThat(writeConf.copyOnWriteDistributionMode(DELETE)).isEqualTo(expectedMode);
+ assertThat(writeConf.positionDeltaDistributionMode(DELETE)).isEqualTo(expectedMode);
+ assertThat(writeConf.copyOnWriteDistributionMode(UPDATE)).isEqualTo(expectedMode);
+ assertThat(writeConf.positionDeltaDistributionMode(UPDATE)).isEqualTo(expectedMode);
+ assertThat(writeConf.copyOnWriteDistributionMode(MERGE)).isEqualTo(expectedMode);
+ assertThat(writeConf.positionDeltaDistributionMode(MERGE)).isEqualTo(expectedMode);
}
}
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestMigrateTableAction.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestMigrateTableAction.java
index 7bed72b7cc..94afa50cf4 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestMigrateTableAction.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestMigrateTableAction.java
@@ -18,34 +18,32 @@
*/
package org.apache.iceberg.spark.actions;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assumptions.assumeThat;
import java.io.IOException;
-import java.util.Map;
+import java.nio.file.Files;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.iceberg.spark.SparkCatalogTestBase;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Test;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.spark.CatalogTestBase;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
-public class TestMigrateTableAction extends SparkCatalogTestBase {
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestMigrateTableAction extends CatalogTestBase {
- public TestMigrateTableAction(
- String catalogName, String implementation, Map<String, String> config) {
- super(catalogName, implementation, config);
- assumeThat(catalogName).isEqualToIgnoringCase("spark_catalog");
- }
-
- @After
+ @AfterEach
public void removeTables() {
sql("DROP TABLE IF EXISTS %s", tableName);
sql("DROP TABLE IF EXISTS %s_BACKUP_", tableName);
}
- @Test
+ @TestTemplate
public void testMigrateWithParallelTasks() throws IOException {
- String location = temp.newFolder().toURI().toString();
+ assumeThat(catalogName).isEqualToIgnoringCase("spark_catalog");
+ String location = Files.createTempDirectory(temp, "junit").toFile().toString();
sql(
"CREATE TABLE %s (id bigint NOT NULL, data string) USING parquet LOCATION '%s'",
tableName, location);
@@ -65,6 +63,6 @@ public class TestMigrateTableAction extends SparkCatalogTestBase {
return thread;
}))
.execute();
- Assert.assertEquals(migrationThreadsIndex.get(), 2);
+ assertThat(migrationThreadsIndex.get()).isEqualTo(2);
}
}
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestDataSourceOptions.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestDataSourceOptions.java
index 013b8d4386..99d3f38ee7 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestDataSourceOptions.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestDataSourceOptions.java
@@ -31,6 +31,7 @@ import org.apache.iceberg.DataFile;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
@@ -45,8 +46,8 @@ import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.math.LongMath;
import org.apache.iceberg.spark.CommitMetadata;
import org.apache.iceberg.spark.SparkReadOptions;
-import org.apache.iceberg.spark.SparkTestBaseWithCatalog;
import org.apache.iceberg.spark.SparkWriteOptions;
+import org.apache.iceberg.spark.TestBaseWithCatalog;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.SnapshotUtil;
import org.apache.spark.sql.Column;
@@ -57,14 +58,13 @@ import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
import org.apache.spark.sql.functions;
-import org.junit.AfterClass;
-import org.junit.Assert;
-import org.junit.BeforeClass;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
-public class TestDataSourceOptions extends SparkTestBaseWithCatalog {
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestDataSourceOptions extends TestBaseWithCatalog {
private static final Configuration CONF = new Configuration();
private static final Schema SCHEMA =
@@ -72,23 +72,21 @@ public class TestDataSourceOptions extends SparkTestBaseWithCatalog {
optional(1, "id", Types.IntegerType.get()), optional(2, "data", Types.StringType.get()));
private static SparkSession spark = null;
- @Rule public TemporaryFolder temp = new TemporaryFolder();
-
- @BeforeClass
+ @BeforeAll
public static void startSpark() {
TestDataSourceOptions.spark = SparkSession.builder().master("local[2]").getOrCreate();
}
- @AfterClass
+ @AfterAll
public static void stopSpark() {
SparkSession currentSpark = TestDataSourceOptions.spark;
TestDataSourceOptions.spark = null;
currentSpark.stop();
}
- @Test
+ @TestTemplate
public void testWriteFormatOptionOverridesTableProperties() throws IOException {
- String tableLocation = temp.newFolder("iceberg-table").toString();
+ String tableLocation = temp.resolve("iceberg-table").toFile().toString();
HadoopTables tables = new HadoopTables(CONF);
PartitionSpec spec = PartitionSpec.unpartitioned();
@@ -111,14 +109,14 @@ public class TestDataSourceOptions extends SparkTestBaseWithCatalog {
tasks.forEach(
task -> {
FileFormat fileFormat = FileFormat.fromFileName(task.file().location());
- Assert.assertEquals(FileFormat.PARQUET, fileFormat);
+ assertThat(fileFormat).isEqualTo(FileFormat.PARQUET);
});
}
}
- @Test
+ @TestTemplate
public void testNoWriteFormatOption() throws IOException {
- String tableLocation = temp.newFolder("iceberg-table").toString();
+ String tableLocation = temp.resolve("iceberg-table").toFile().toString();
HadoopTables tables = new HadoopTables(CONF);
PartitionSpec spec = PartitionSpec.unpartitioned();
@@ -136,14 +134,14 @@ public class TestDataSourceOptions extends SparkTestBaseWithCatalog {
tasks.forEach(
task -> {
FileFormat fileFormat = FileFormat.fromFileName(task.file().location());
- Assert.assertEquals(FileFormat.AVRO, fileFormat);
+ assertThat(fileFormat).isEqualTo(FileFormat.AVRO);
});
}
}
- @Test
+ @TestTemplate
public void testHadoopOptions() throws IOException {
- String tableLocation = temp.newFolder("iceberg-table").toString();
+ String tableLocation = temp.resolve("iceberg-table").toFile().toString();
Configuration sparkHadoopConf = spark.sessionState().newHadoopConf();
String originalDefaultFS = sparkHadoopConf.get("fs.default.name");
@@ -177,15 +175,15 @@ public class TestDataSourceOptions extends SparkTestBaseWithCatalog {
List<SimpleRecord> resultRecords =
resultDf.orderBy("id").as(Encoders.bean(SimpleRecord.class)).collectAsList();
- Assert.assertEquals("Records should match", expectedRecords, resultRecords);
+ assertThat(resultRecords).as("Records should match").isEqualTo(expectedRecords);
} finally {
sparkHadoopConf.set("fs.default.name", originalDefaultFS);
}
}
- @Test
+ @TestTemplate
public void testSplitOptionsOverridesTableProperties() throws IOException {
- String tableLocation = temp.newFolder("iceberg-table").toString();
+ String tableLocation = temp.resolve("iceberg-table").toFile().toString();
HadoopTables tables = new HadoopTables(CONF);
PartitionSpec spec = PartitionSpec.unpartitioned();
@@ -209,7 +207,7 @@ public class TestDataSourceOptions extends SparkTestBaseWithCatalog {
List<DataFile> files =
Lists.newArrayList(icebergTable.currentSnapshot().addedDataFiles(icebergTable.io()));
- Assert.assertEquals("Should have written 1 file", 1, files.size());
+ assertThat(files).as("Should have written 1 file").hasSize(1);
long fileSize = files.get(0).fileSizeInBytes();
long splitSize = LongMath.divide(fileSize, 2, RoundingMode.CEILING);
@@ -221,12 +219,14 @@ public class TestDataSourceOptions extends SparkTestBaseWithCatalog {
.option(SparkReadOptions.SPLIT_SIZE, String.valueOf(splitSize))
.load(tableLocation);
- Assert.assertEquals("Spark partitions should match", 2, resultDf.javaRDD().getNumPartitions());
+ assertThat(resultDf.javaRDD().getNumPartitions())
+ .as("Spark partitions should match")
+ .isEqualTo(2);
}
- @Test
+ @TestTemplate
public void testIncrementalScanOptions() throws IOException {
- String tableLocation = temp.newFolder("iceberg-table").toString();
+ String tableLocation = temp.resolve("iceberg-table").toFile().toString();
HadoopTables tables = new HadoopTables(CONF);
PartitionSpec spec = PartitionSpec.unpartitioned();
@@ -330,9 +330,9 @@ public class TestDataSourceOptions extends SparkTestBaseWithCatalog {
assertThat(row2.getInt(1)).as("max value should match").isEqualTo(3);
}
- @Test
+ @TestTemplate
public void testMetadataSplitSizeOptionOverrideTableProperties() throws IOException {
- String tableLocation = temp.newFolder("iceberg-table").toString();
+ String tableLocation = temp.resolve("iceberg-table").toFile().toString();
HadoopTables tables = new HadoopTables(CONF);
PartitionSpec spec = PartitionSpec.unpartitioned();
@@ -349,7 +349,7 @@ public class TestDataSourceOptions extends SparkTestBaseWithCatalog {
List<ManifestFile> manifests = table.currentSnapshot().allManifests(table.io());
- Assert.assertEquals("Must be 2 manifests", 2, manifests.size());
+ assertThat(manifests).as("Must be 2 manifests").hasSize(2);
// set the target metadata split size so each manifest ends up in a separate split
table
@@ -358,7 +358,7 @@ public class TestDataSourceOptions extends SparkTestBaseWithCatalog {
.commit();
Dataset<Row> entriesDf = spark.read().format("iceberg").load(tableLocation + "#entries");
- Assert.assertEquals("Num partitions must match", 2, entriesDf.javaRDD().getNumPartitions());
+ assertThat(entriesDf.javaRDD().getNumPartitions()).as("Num partitions must match").isEqualTo(2);
// override the table property using options
entriesDf =
@@ -367,12 +367,12 @@ public class TestDataSourceOptions extends SparkTestBaseWithCatalog {
.format("iceberg")
.option(SparkReadOptions.SPLIT_SIZE, String.valueOf(128 * 1024 * 1024))
.load(tableLocation + "#entries");
- Assert.assertEquals("Num partitions must match", 1, entriesDf.javaRDD().getNumPartitions());
+ assertThat(entriesDf.javaRDD().getNumPartitions()).as("Num partitions must match").isEqualTo(1);
}
- @Test
+ @TestTemplate
public void testDefaultMetadataSplitSize() throws IOException {
- String tableLocation = temp.newFolder("iceberg-table").toString();
+ String tableLocation = temp.resolve("iceberg-table").toFile().toString();
HadoopTables tables = new HadoopTables(CONF);
PartitionSpec spec = PartitionSpec.unpartitioned();
@@ -401,12 +401,12 @@ public class TestDataSourceOptions extends SparkTestBaseWithCatalog {
Dataset<Row> metadataDf = spark.read().format("iceberg").load(tableLocation + "#entries");
int partitionNum = metadataDf.javaRDD().getNumPartitions();
- Assert.assertEquals("Spark partitions should match", expectedSplits, partitionNum);
+ assertThat(partitionNum).as("Spark partitions should match").isEqualTo(expectedSplits);
}
- @Test
+ @TestTemplate
public void testExtraSnapshotMetadata() throws IOException {
- String tableLocation = temp.newFolder("iceberg-table").toString();
+ String tableLocation = temp.resolve("iceberg-table").toFile().toString();
HadoopTables tables = new HadoopTables(CONF);
tables.create(SCHEMA, PartitionSpec.unpartitioned(), Maps.newHashMap(), tableLocation);
@@ -424,13 +424,14 @@ public class TestDataSourceOptions extends SparkTestBaseWithCatalog {
Table table = tables.load(tableLocation);
- Assert.assertTrue(table.currentSnapshot().summary().get("extra-key").equals("someValue"));
- Assert.assertTrue(table.currentSnapshot().summary().get("another-key").equals("anotherValue"));
+ assertThat(table.currentSnapshot().summary())
+ .containsEntry("extra-key", "someValue")
+ .containsEntry("another-key", "anotherValue");
}
- @Test
+ @TestTemplate
public void testExtraSnapshotMetadataWithSQL() throws InterruptedException, IOException {
- String tableLocation = temp.newFolder("iceberg-table").toString();
+ String tableLocation = temp.resolve("iceberg-table").toFile().toString();
HadoopTables tables = new HadoopTables(CONF);
Table table =
@@ -465,15 +466,15 @@ public class TestDataSourceOptions extends SparkTestBaseWithCatalog {
writerThread.join();
List<Snapshot> snapshots = Lists.newArrayList(table.snapshots());
- Assert.assertEquals(2, snapshots.size());
- Assert.assertNull(snapshots.get(0).summary().get("writer-thread"));
+ assertThat(snapshots).hasSize(2);
+ assertThat(snapshots.get(0).summary()).doesNotContainKey("writer-thread");
assertThat(snapshots.get(1).summary())
.containsEntry("writer-thread", "test-extra-commit-message-writer-thread")
.containsEntry("extra-key", "someValue")
.containsEntry("another-key", "anotherValue");
}
- @Test
+ @TestTemplate
public void testExtraSnapshotMetadataWithDelete()
throws InterruptedException, NoSuchTableException {
spark.sessionState().conf().setConfString("spark.sql.shuffle.partitions", "1");
@@ -508,8 +509,7 @@ public class TestDataSourceOptions extends SparkTestBaseWithCatalog {
Table table = validationCatalog.loadTable(tableIdent);
List<Snapshot> snapshots = Lists.newArrayList(table.snapshots());
- Assert.assertEquals(2, snapshots.size());
- Assert.assertNull(snapshots.get(0).summary().get("writer-thread"));
+ assertThat(snapshots.get(0).summary()).doesNotContainKey("writer-thread");
assertThat(snapshots.get(1).summary())
.containsEntry("writer-thread", "test-extra-commit-message-delete-thread")
.containsEntry("extra-key", "someValue")
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 746415818c..701c421a25 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
@@ -26,20 +26,17 @@ import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.hadoop.HadoopTables;
-import org.junit.Before;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.io.TempDir;
public class TestIcebergSourceHadoopTables extends TestIcebergSourceTablesBase {
private static final HadoopTables TABLES = new HadoopTables(new Configuration());
-
- File tableDir = null;
+ @TempDir private File tableDir;
String tableLocation = null;
- @Before
- public void setupTable() throws Exception {
- this.tableDir = temp.newFolder();
- tableDir.delete(); // created by table create
-
+ @BeforeEach
+ public void setupTable() {
this.tableLocation = tableDir.toURI().toString();
}
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 2a264b74b0..9120bbcc35 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
@@ -27,14 +27,14 @@ import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
-import org.junit.After;
-import org.junit.BeforeClass;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
public class TestIcebergSourceHiveTables extends TestIcebergSourceTablesBase {
private static TableIdentifier currentIdentifier;
- @BeforeClass
+ @BeforeAll
public static void start() {
Namespace db = Namespace.of("db");
if (!catalog.namespaceExists(db)) {
@@ -42,7 +42,7 @@ public class TestIcebergSourceHiveTables extends TestIcebergSourceTablesBase {
}
}
- @After
+ @AfterEach
public void dropTable() throws IOException {
if (!catalog.tableExists(currentIdentifier)) {
return;
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 19e8e8146d..09219963f7 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
@@ -79,8 +79,8 @@ import org.apache.iceberg.spark.SparkReadOptions;
import org.apache.iceberg.spark.SparkSQLProperties;
import org.apache.iceberg.spark.SparkSchemaUtil;
import org.apache.iceberg.spark.SparkTableUtil;
-import org.apache.iceberg.spark.SparkTestBase;
import org.apache.iceberg.spark.SparkWriteOptions;
+import org.apache.iceberg.spark.TestBase;
import org.apache.iceberg.spark.actions.SparkActions;
import org.apache.iceberg.spark.data.TestHelpers;
import org.apache.iceberg.types.Types;
@@ -96,13 +96,11 @@ import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.functions;
import org.apache.spark.sql.internal.SQLConf;
import org.apache.spark.sql.types.StructType;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
-public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
+public abstract class TestIcebergSourceTablesBase extends TestBase {
private static final Schema SCHEMA =
new Schema(
@@ -121,7 +119,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
private static final PartitionSpec SPEC = PartitionSpec.builderFor(SCHEMA).identity("id").build();
- @Rule public TemporaryFolder temp = new TemporaryFolder();
+ @TempDir protected Path temp;
public abstract Table createTable(
TableIdentifier ident, Schema schema, PartitionSpec spec, Map<String, String> properties);
@@ -134,7 +132,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
public abstract void dropTable(TableIdentifier ident) throws IOException;
- @After
+ @AfterEach
public void removeTable() {
spark.sql("DROP TABLE IF EXISTS parquet_table");
}
@@ -164,7 +162,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
List<SimpleRecord> actualRecords =
resultDf.orderBy("id").as(Encoders.bean(SimpleRecord.class)).collectAsList();
- Assert.assertEquals("Records should match", expectedRecords, actualRecords);
+ assertThat(actualRecords).as("Records should match").isEqualTo(expectedRecords);
}
@Test
@@ -191,8 +189,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
Snapshot snapshot = table.currentSnapshot();
- Assert.assertEquals(
- "Should only contain one manifest", 1, snapshot.allManifests(table.io()).size());
+ assertThat(snapshot.allManifests(table.io())).as("Should only contain one manifest").hasSize(1);
InputFile manifest = table.io().newInputFile(snapshot.allManifests(table.io()).get(0).path());
List<GenericData.Record> expected = Lists.newArrayList();
@@ -209,8 +206,8 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
});
}
- Assert.assertEquals("Entries table should have one row", 1, expected.size());
- Assert.assertEquals("Actual results should have one row", 1, actual.size());
+ assertThat(expected).as("Entries table should have one row").hasSize(1);
+ assertThat(actual).as("Actual results should have one row").hasSize(1);
TestHelpers.assertEqualsSafe(
TestHelpers.nonDerivedSchema(entriesTableDs), expected.get(0), actual.get(0));
}
@@ -240,8 +237,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.select("status")
.collectAsList();
- Assert.assertEquals("Results should contain only one status", 1, actual.size());
- Assert.assertEquals("That status should be Added (1)", 1, actual.get(0).getInt(0));
+ assertThat(actual).singleElement().satisfies(row -> assertThat(row.getInt(0)).isEqualTo(1));
}
@Test
@@ -412,8 +408,8 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
expected.sort(Comparator.comparing(o -> (Long) o.get("snapshot_id")));
- Assert.assertEquals("Entries table should have 3 rows", 3, expected.size());
- Assert.assertEquals("Actual results should have 3 rows", 3, actual.size());
+ assertThat(expected).as("Entries table should have 3 rows").hasSize(3);
+ assertThat(actual).as("Actual results should have 3 rows").hasSize(3);
for (int i = 0; i < expected.size(); i += 1) {
TestHelpers.assertEqualsSafe(
TestHelpers.nonDerivedSchema(entriesTableDs), expected.get(i), actual.get(i));
@@ -438,16 +434,20 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
final int expectedEntryCount = 1;
// count entries
- Assert.assertEquals(
- "Count should return " + expectedEntryCount,
- expectedEntryCount,
- spark.read().format("iceberg").load(loadLocation(tableIdentifier, "entries")).count());
+ assertThat(
+ spark.read().format("iceberg").load(loadLocation(tableIdentifier, "entries")).count())
+ .as("Count should return " + expectedEntryCount)
+ .isEqualTo(expectedEntryCount);
// count all_entries
- Assert.assertEquals(
- "Count should return " + expectedEntryCount,
- expectedEntryCount,
- spark.read().format("iceberg").load(loadLocation(tableIdentifier, "all_entries")).count());
+ assertThat(
+ spark
+ .read()
+ .format("iceberg")
+ .load(loadLocation(tableIdentifier, "all_entries"))
+ .count())
+ .as("Count should return " + expectedEntryCount)
+ .isEqualTo(expectedEntryCount);
}
@Test
@@ -496,8 +496,8 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
}
}
- Assert.assertEquals("Files table should have one row", 1, expected.size());
- Assert.assertEquals("Actual results should have one row", 1, actual.size());
+ assertThat(expected).as("Files table should have one row").hasSize(1);
+ assertThat(actual).as("Actual results should have one row").hasSize(1);
TestHelpers.assertEqualsSafe(
TestHelpers.nonDerivedSchema(filesTableDs), expected.get(0), actual.get(0));
@@ -514,7 +514,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
String.format(
"CREATE TABLE parquet_table (data string, id int) "
+ "USING parquet PARTITIONED BY (id) LOCATION '%s'",
- temp.newFolder()));
+ temp.toFile()));
List<SimpleRecord> records =
Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "b"));
@@ -552,8 +552,8 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
}
Types.StructType struct = TestHelpers.nonDerivedSchema(filesTableDs);
- Assert.assertEquals("Files table should have one row", 2, expected.size());
- Assert.assertEquals("Actual results should have one row", 2, actual.size());
+ assertThat(expected).as("Files table should have 2 rows").hasSize(2);
+ assertThat(actual).as("Actual results should have 2 rows").hasSize(2);
TestHelpers.assertEqualsSafe(struct, expected.get(0), actual.get(0));
TestHelpers.assertEqualsSafe(struct, expected.get(1), actual.get(1));
}
@@ -570,7 +570,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
String.format(
"CREATE TABLE parquet_table (data string, id int) "
+ "USING parquet PARTITIONED BY (id) LOCATION '%s'",
- temp.newFolder()));
+ temp.toFile()));
List<SimpleRecord> records =
Lists.newArrayList(new SimpleRecord(1, "a"), new SimpleRecord(2, "b"));
@@ -597,11 +597,21 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
long snapshotId = table.currentSnapshot().snapshotId();
- Assert.assertEquals("Entries table should have 2 rows", 2, actual.size());
- Assert.assertEquals("Sequence number must match", 0, actual.get(0).getLong(0));
- Assert.assertEquals("Snapshot id must match", snapshotId, actual.get(0).getLong(1));
- Assert.assertEquals("Sequence number must match", 0, actual.get(1).getLong(0));
- Assert.assertEquals("Snapshot id must match", snapshotId, actual.get(1).getLong(1));
+ assertThat(actual).as("Entries table should have 2 rows").hasSize(2);
+ assertThat(actual)
+ .first()
+ .satisfies(
+ row -> {
+ assertThat(row.getLong(0)).isEqualTo(0);
+ assertThat(row.getLong(1)).isEqualTo(snapshotId);
+ });
+ assertThat(actual)
+ .element(1)
+ .satisfies(
+ row -> {
+ assertThat(row.getLong(0)).isEqualTo(0);
+ assertThat(row.getLong(1)).isEqualTo(snapshotId);
+ });
}
@Test
@@ -654,8 +664,8 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
}
}
- Assert.assertEquals("Files table should have one row", 1, expected.size());
- Assert.assertEquals("Actual results should have one row", 1, actual.size());
+ assertThat(expected).as("Files table should have one row").hasSize(1);
+ assertThat(actual).as("Actual results should have one row").hasSize(1);
TestHelpers.assertEqualsSafe(
TestHelpers.nonDerivedSchema(filesTableDs), expected.get(0), actual.get(0));
}
@@ -706,12 +716,11 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.load(loadLocation(tableIdentifier, "all_entries"))
.collectAsList();
- Assert.assertTrue(
- "Stage table should have some snapshots", table.snapshots().iterator().hasNext());
- Assert.assertNull("Stage table should have null currentSnapshot", table.currentSnapshot());
- Assert.assertEquals("Actual results should have two rows", 2, actualAllData.size());
- Assert.assertEquals("Actual results should have two rows", 2, actualAllManifests.size());
- Assert.assertEquals("Actual results should have two rows", 2, actualAllEntries.size());
+ assertThat(table.snapshots().iterator()).as("Stage table should have some snapshots").hasNext();
+ assertThat(table.currentSnapshot()).as("Stage table should have null currentSnapshot").isNull();
+ assertThat(actualAllData).as("Actual results should have two rows").hasSize(2);
+ assertThat(actualAllManifests).as("Actual results should have two rows").hasSize(2);
+ assertThat(actualAllEntries).as("Actual results should have two rows").hasSize(2);
}
@Test
@@ -769,8 +778,8 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
expected.sort(Comparator.comparing(o -> o.get("file_path").toString()));
- Assert.assertEquals("Files table should have two rows", 2, expected.size());
- Assert.assertEquals("Actual results should have two rows", 2, actual.size());
+ assertThat(expected).as("Files table should have two rows").hasSize(2);
+ assertThat(actual).as("Actual results should have two rows").hasSize(2);
for (int i = 0; i < expected.size(); i += 1) {
TestHelpers.assertEqualsSafe(
TestHelpers.nonDerivedSchema(filesTableDs), expected.get(i), actual.get(i));
@@ -861,7 +870,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.set("is_current_ancestor", true)
.build());
- Assert.assertEquals("History table should have a row for each commit", 4, actual.size());
+ assertThat(actual).as("History table should have a row for each commit").hasSize(4);
TestHelpers.assertEqualsSafe(historyTable.schema().asStruct(), expected.get(0), actual.get(0));
TestHelpers.assertEqualsSafe(historyTable.schema().asStruct(), expected.get(1), actual.get(1));
TestHelpers.assertEqualsSafe(historyTable.schema().asStruct(), expected.get(2), actual.get(2));
@@ -940,7 +949,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
"total-data-files", "0"))
.build());
- Assert.assertEquals("Snapshots table should have a row for each snapshot", 2, actual.size());
+ assertThat(actual).as("Snapshots table should have a row for each snapshot").hasSize(2);
TestHelpers.assertEqualsSafe(snapTable.schema().asStruct(), expected.get(0), actual.get(0));
TestHelpers.assertEqualsSafe(snapTable.schema().asStruct(), expected.get(1), actual.get(1));
}
@@ -1013,7 +1022,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
"total-data-files", "0"))
.build());
- Assert.assertEquals("Snapshots table should have a row for each snapshot", 2, actual.size());
+ assertThat(actual).as("Snapshots table should have a row for each snapshot").hasSize(2);
TestHelpers.assertEqualsSafe(projectedSchema.asStruct(), expected.get(0), actual.get(0));
TestHelpers.assertEqualsSafe(projectedSchema.asStruct(), expected.get(1), actual.get(1));
}
@@ -1098,7 +1107,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.build()))
.build());
- Assert.assertEquals("Manifests table should have two manifest rows", 2, actual.size());
+ assertThat(actual).as("Manifests table should have two manifest rows").hasSize(2);
TestHelpers.assertEqualsSafe(manifestTable.schema().asStruct(), expected.get(0), actual.get(0));
TestHelpers.assertEqualsSafe(manifestTable.schema().asStruct(), expected.get(1), actual.get(1));
}
@@ -1179,7 +1188,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.build()))
.build());
- Assert.assertEquals("Manifests table should have one manifest row", 1, actual.size());
+ assertThat(actual).as("Manifests table should have one manifest row").hasSize(1);
TestHelpers.assertEqualsSafe(projectedSchema.asStruct(), expected.get(0), actual.get(0));
}
@@ -1231,7 +1240,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.sorted(Comparator.comparing(o -> o.get("path").toString()))
.collect(Collectors.toList());
- Assert.assertEquals("Manifests table should have 5 manifest rows", 5, actual.size());
+ assertThat(actual).as("Manifests table should have 5 manifest rows").hasSize(5);
for (int i = 0; i < expected.size(); i += 1) {
TestHelpers.assertEqualsSafe(
manifestTable.schema().asStruct(), expected.get(i), actual.get(i));
@@ -1294,10 +1303,9 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
Table partitionsTable = loadTable(tableIdentifier, "partitions");
- Assert.assertEquals(
- "Schema should not have partition field",
- expectedSchema,
- partitionsTable.schema().asStruct());
+ assertThat(expectedSchema)
+ .as("Schema should not have partition field")
+ .isEqualTo(partitionsTable.schema().asStruct());
GenericRecordBuilder builder =
new GenericRecordBuilder(AvroSchemaUtil.convert(partitionsTable.schema(), "partitions"));
@@ -1323,7 +1331,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.load(loadLocation(tableIdentifier, "partitions"))
.collectAsList();
- Assert.assertEquals("Unpartitioned partitions table should have one row", 1, actual.size());
+ assertThat(actual).as("Unpartitioned partitions table should have one row").hasSize(1);
TestHelpers.assertEqualsSafe(expectedSchema, expectedRow, actual.get(0));
}
@@ -1404,8 +1412,8 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.set("last_updated_snapshot_id", secondCommitId)
.build());
- Assert.assertEquals("Partitions table should have two rows", 2, expected.size());
- Assert.assertEquals("Actual results should have two rows", 2, actual.size());
+ assertThat(expected).as("Partitions table should have two rows").hasSize(2);
+ assertThat(actual).as("Actual results should have two rows").hasSize(2);
for (int i = 0; i < 2; i += 1) {
TestHelpers.assertEqualsSafe(
partitionsTable.schema().asStruct(), expected.get(i), actual.get(i));
@@ -1421,7 +1429,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.orderBy("partition.id")
.collectAsList();
- Assert.assertEquals("Actual results should have one row", 1, actualAfterFirstCommit.size());
+ assertThat(actualAfterFirstCommit).as("Actual results should have one row").hasSize(1);
TestHelpers.assertEqualsSafe(
partitionsTable.schema().asStruct(), expected.get(0), actualAfterFirstCommit.get(0));
@@ -1433,7 +1441,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.load(loadLocation(tableIdentifier, "partitions"))
.filter("partition.id < 2")
.collectAsList();
- Assert.assertEquals("Actual results should have one row", 1, filtered.size());
+ assertThat(filtered).as("Actual results should have one row").hasSize(1);
TestHelpers.assertEqualsSafe(
partitionsTable.schema().asStruct(), expected.get(0), filtered.get(0));
@@ -1444,7 +1452,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.load(loadLocation(tableIdentifier, "partitions"))
.filter("partition.id < 2 or record_count=1")
.collectAsList();
- Assert.assertEquals("Actual results should have two row", 2, nonFiltered.size());
+ assertThat(nonFiltered).as("Actual results should have two rows").hasSize(2);
for (int i = 0; i < 2; i += 1) {
TestHelpers.assertEqualsSafe(
partitionsTable.schema().asStruct(), expected.get(i), actual.get(i));
@@ -1485,12 +1493,10 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
// check if rewrite manifest does not override metadata about data file's creating snapshot
RewriteManifests.Result rewriteManifestResult =
SparkActions.get().rewriteManifests(table).execute();
- Assert.assertEquals(
- "rewrite replaced 2 manifests",
- 2,
- Iterables.size(rewriteManifestResult.rewrittenManifests()));
- Assert.assertEquals(
- "rewrite added 1 manifests", 1, Iterables.size(rewriteManifestResult.addedManifests()));
+ assertThat(rewriteManifestResult.rewrittenManifests())
+ .as("rewrite replaced 2 manifests")
+ .hasSize(2);
+ assertThat(rewriteManifestResult.addedManifests()).as("rewrite added 1 manifests").hasSize(1);
List<Row> actual =
spark
@@ -1542,8 +1548,8 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.set("last_updated_snapshot_id", secondCommitId)
.build());
- Assert.assertEquals("Partitions table should have two rows", 2, expected.size());
- Assert.assertEquals("Actual results should have two rows", 2, actual.size());
+ assertThat(expected).as("Partitions table should have two rows").hasSize(2);
+ assertThat(actual).as("Actual results should have two rows").hasSize(2);
for (int i = 0; i < 2; i += 1) {
TestHelpers.assertEqualsSafe(
partitionsTable.schema().asStruct(), expected.get(i), actual.get(i));
@@ -1557,7 +1563,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.load(loadLocation(tableIdentifier, "partitions"))
.filter("partition.id < 2")
.collectAsList();
- Assert.assertEquals("Actual results should have one row", 1, filtered.size());
+ assertThat(filtered).as("Actual results should have one row").hasSize(1);
TestHelpers.assertEqualsSafe(
partitionsTable.schema().asStruct(), expected.get(0), filtered.get(0));
@@ -1588,8 +1594,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.format("iceberg")
.load(loadLocation(tableIdentifier, "partitions"))
.collectAsList();
- Assert.assertEquals(
- "Actual results should have two row", 2, actualAfterSnapshotExpiration.size());
+ assertThat(actualAfterSnapshotExpiration).as("Actual results should have two rows").hasSize(2);
for (int i = 0; i < 2; i += 1) {
TestHelpers.assertEqualsSafe(
partitionsTable.schema().asStruct(),
@@ -1645,7 +1650,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.load(loadLocation(tableIdentifier, "partitions"))
.orderBy("partition.id")
.collectAsList();
- Assert.assertEquals("Actual results should have two rows", 2, actual.size());
+ assertThat(actual).as("Actual results should have two rows").hasSize(2);
GenericRecordBuilder builder =
new GenericRecordBuilder(AvroSchemaUtil.convert(partitionsTable.schema(), "partitions"));
@@ -1705,7 +1710,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.load(loadLocation(tableIdentifier, "partitions"))
.orderBy("partition.id")
.collectAsList();
- Assert.assertEquals("Actual results should have two rows", 2, actual.size());
+ assertThat(actual).as("Actual results should have two rows").hasSize(2);
expected.remove(0);
expected.add(
0,
@@ -1747,8 +1752,9 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
table.refresh();
Dataset<Row> resultDf = spark.read().format("iceberg").load(loadLocation(tableIdentifier));
- Assert.assertEquals(
- "Records should match", originalRecords, resultDf.orderBy("id").collectAsList());
+ assertThat(resultDf.orderBy("id").collectAsList())
+ .as("Records should match")
+ .containsExactlyElementsOf(originalRecords);
Snapshot snapshotBeforeAddColumn = table.currentSnapshot();
@@ -1777,8 +1783,9 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
RowFactory.create(5, "xyz", "C"));
Dataset<Row> resultDf2 = spark.read().format("iceberg").load(loadLocation(tableIdentifier));
- Assert.assertEquals(
- "Records should match", updatedRecords, resultDf2.orderBy("id").collectAsList());
+ assertThat(resultDf2.orderBy("id").collectAsList())
+ .as("Records should match")
+ .containsExactlyElementsOf(updatedRecords);
Dataset<Row> resultDf3 =
spark
@@ -1786,9 +1793,10 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.format("iceberg")
.option(SparkReadOptions.SNAPSHOT_ID, snapshotBeforeAddColumn.snapshotId())
.load(loadLocation(tableIdentifier));
- Assert.assertEquals(
- "Records should match", originalRecords, resultDf3.orderBy("id").collectAsList());
- Assert.assertEquals("Schemas should match", originalSparkSchema, resultDf3.schema());
+ assertThat(resultDf3.orderBy("id").collectAsList())
+ .as("Records should match")
+ .containsExactlyElementsOf(originalRecords);
+ assertThat(resultDf3.schema()).as("Schemas should match").isEqualTo(originalSparkSchema);
}
@Test
@@ -1814,8 +1822,9 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
table.refresh();
Dataset<Row> resultDf = spark.read().format("iceberg").load(loadLocation(tableIdentifier));
- Assert.assertEquals(
- "Records should match", originalRecords, resultDf.orderBy("id").collectAsList());
+ assertThat(resultDf.orderBy("id").collectAsList())
+ .as("Records should match")
+ .containsExactlyElementsOf(originalRecords);
long tsBeforeDropColumn = waitUntilAfter(System.currentTimeMillis());
table.updateSchema().deleteColumn("data").commit();
@@ -1843,8 +1852,9 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
RowFactory.create(5, "C"));
Dataset<Row> resultDf2 = spark.read().format("iceberg").load(loadLocation(tableIdentifier));
- Assert.assertEquals(
- "Records should match", updatedRecords, resultDf2.orderBy("id").collectAsList());
+ assertThat(resultDf2.orderBy("id").collectAsList())
+ .as("Records should match")
+ .containsExactlyElementsOf(updatedRecords);
Dataset<Row> resultDf3 =
spark
@@ -1852,9 +1862,10 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.format("iceberg")
.option(SparkReadOptions.AS_OF_TIMESTAMP, tsBeforeDropColumn)
.load(loadLocation(tableIdentifier));
- Assert.assertEquals(
- "Records should match", originalRecords, resultDf3.orderBy("id").collectAsList());
- Assert.assertEquals("Schemas should match", originalSparkSchema, resultDf3.schema());
+ assertThat(resultDf3.orderBy("id").collectAsList())
+ .as("Records should match")
+ .containsExactlyElementsOf(originalRecords);
+ assertThat(resultDf3.schema()).as("Schemas should match").isEqualTo(originalSparkSchema);
// At tsAfterDropColumn, there has been a schema change, but no new snapshot,
// so the snapshot as of tsAfterDropColumn is the same as that as of tsBeforeDropColumn.
@@ -1864,9 +1875,10 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.format("iceberg")
.option(SparkReadOptions.AS_OF_TIMESTAMP, tsAfterDropColumn)
.load(loadLocation(tableIdentifier));
- Assert.assertEquals(
- "Records should match", originalRecords, resultDf4.orderBy("id").collectAsList());
- Assert.assertEquals("Schemas should match", originalSparkSchema, resultDf4.schema());
+ assertThat(resultDf4.orderBy("id").collectAsList())
+ .as("Records should match")
+ .containsExactlyElementsOf(originalRecords);
+ assertThat(resultDf4.schema()).as("Schemas should match").isEqualTo(originalSparkSchema);
}
@Test
@@ -1890,8 +1902,9 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
table.refresh();
Dataset<Row> resultDf = spark.read().format("iceberg").load(loadLocation(tableIdentifier));
- Assert.assertEquals(
- "Records should match", originalRecords, resultDf.orderBy("id").collectAsList());
+ assertThat(resultDf.orderBy("id").collectAsList())
+ .as("Records should match")
+ .containsExactlyElementsOf(originalRecords);
Snapshot snapshotBeforeAddColumn = table.currentSnapshot();
@@ -1920,8 +1933,9 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
RowFactory.create(5, "xyz", "C"));
Dataset<Row> resultDf2 = spark.read().format("iceberg").load(loadLocation(tableIdentifier));
- Assert.assertEquals(
- "Records should match", updatedRecords, resultDf2.orderBy("id").collectAsList());
+ assertThat(resultDf2.orderBy("id").collectAsList())
+ .as("Records should match")
+ .containsExactlyElementsOf(updatedRecords);
table.updateSchema().deleteColumn("data").commit();
@@ -1934,8 +1948,9 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
RowFactory.create(5, "C"));
Dataset<Row> resultDf3 = spark.read().format("iceberg").load(loadLocation(tableIdentifier));
- Assert.assertEquals(
- "Records should match", recordsAfterDropColumn, resultDf3.orderBy("id").collectAsList());
+ assertThat(resultDf3.orderBy("id").collectAsList())
+ .as("Records should match")
+ .containsExactlyElementsOf(recordsAfterDropColumn);
Dataset<Row> resultDf4 =
spark
@@ -1943,9 +1958,10 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.format("iceberg")
.option(SparkReadOptions.SNAPSHOT_ID, snapshotBeforeAddColumn.snapshotId())
.load(loadLocation(tableIdentifier));
- Assert.assertEquals(
- "Records should match", originalRecords, resultDf4.orderBy("id").collectAsList());
- Assert.assertEquals("Schemas should match", originalSparkSchema, resultDf4.schema());
+ assertThat(resultDf4.orderBy("id").collectAsList())
+ .as("Records should match")
+ .containsExactlyElementsOf(originalRecords);
+ assertThat(resultDf4.schema()).as("Schemas should match").isEqualTo(originalSparkSchema);
}
@Test
@@ -1976,19 +1992,16 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.location(table.location() + "/metadata")
.olderThan(System.currentTimeMillis())
.execute();
- Assert.assertTrue(
- "Should not delete any metadata files", Iterables.isEmpty(result1.orphanFileLocations()));
+ assertThat(result1.orphanFileLocations()).as("Should not delete any metadata files").isEmpty();
DeleteOrphanFiles.Result result2 =
actions.deleteOrphanFiles(table).olderThan(System.currentTimeMillis()).execute();
- Assert.assertEquals(
- "Should delete 1 data file", 1, Iterables.size(result2.orphanFileLocations()));
+ assertThat(result2.orphanFileLocations()).as("Should delete 1 data file").hasSize(1);
Dataset<Row> resultDF = spark.read().format("iceberg").load(loadLocation(tableIdentifier));
List<SimpleRecord> actualRecords =
resultDF.as(Encoders.bean(SimpleRecord.class)).collectAsList();
-
- Assert.assertEquals("Rows must match", records, actualRecords);
+ assertThat(actualRecords).as("Rows must match").containsExactlyInAnyOrderElementsOf(records);
}
@Test
@@ -2033,7 +2046,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.map(r -> (Integer) r.getAs(DataFile.SPEC_ID.name()))
.collect(Collectors.toList());
- Assert.assertEquals("Should have two partition specs", ImmutableList.of(spec0, spec1), actual);
+ assertThat(actual).as("Should have two partition specs").containsExactly(spec0, spec1);
}
@Test
@@ -2067,7 +2080,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
table.refresh();
Snapshot snapshot2 = table.currentSnapshot();
- Assert.assertEquals("Should have two manifests", 2, snapshot2.allManifests(table.io()).size());
+ assertThat(snapshot2.allManifests(table.io())).as("Should have two manifests").hasSize(2);
snapshotIdToManifests.addAll(
snapshot2.allManifests(table.io()).stream()
.map(manifest -> Pair.of(snapshot2.snapshotId(), manifest))
@@ -2109,7 +2122,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
.sorted(Comparator.comparing(o -> o.get("path").toString()))
.collect(Collectors.toList());
- Assert.assertEquals("Manifests table should have 3 manifest rows", 3, actual.size());
+ assertThat(actual).as("Manifests table should have 3 manifest rows").hasSize(3);
for (int i = 0; i < expected.size(); i += 1) {
TestHelpers.assertEqualsSafe(
manifestTable.schema().asStruct(), expected.get(i), actual.get(i));
@@ -2118,7 +2131,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
@Test
public void testTableWithInt96Timestamp() throws IOException {
- File parquetTableDir = temp.newFolder("table_timestamp_int96");
+ File parquetTableDir = temp.resolve("table_timestamp_int96").toFile();
String parquetTableLocation = parquetTableDir.toURI().toString();
Schema schema =
new Schema(
@@ -2178,8 +2191,8 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
TableIdentifier tableIdentifier = TableIdentifier.of("db", "missing_files_test");
Table table = createTable(tableIdentifier, SCHEMA, SPEC);
- File parquetTablePath = temp.newFolder("table_missing_files");
- String parquetTableLocation = parquetTablePath.toURI().toString();
+ File parquetTableDir = temp.resolve("table_missing_files").toFile();
+ String parquetTableLocation = parquetTableDir.toURI().toString();
spark.sql(
String.format(
"CREATE TABLE parquet_table (data string, id int) "
@@ -2194,7 +2207,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
// Add a Spark partition of which location is missing
spark.sql("ALTER TABLE parquet_table ADD PARTITION (id = 1234)");
- Path partitionLocationPath = parquetTablePath.toPath().resolve("id=1234");
+ Path partitionLocationPath = parquetTableDir.toPath().resolve("id=1234");
java.nio.file.Files.delete(partitionLocationPath);
String stagingLocation = table.location() + "/metadata";
@@ -2217,7 +2230,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
TableIdentifier tableIdentifier = TableIdentifier.of("db", "missing_files_test");
Table table = createTable(tableIdentifier, SCHEMA, SPEC);
- File parquetTableDir = temp.newFolder("table_missing_files");
+ File parquetTableDir = temp.resolve("table_missing_files").toFile();
String parquetTableLocation = parquetTableDir.toURI().toString();
spark.sql(
String.format(
@@ -2415,7 +2428,7 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
try {
return FileHelpers.writeDeleteFile(
table,
- Files.localOutput(temp.newFile()),
+ Files.localOutput(File.createTempFile("junit", null, temp.toFile())),
org.apache.iceberg.TestHelpers.Row.of(1),
deletes,
deleteRowSchema);
@@ -2430,16 +2443,14 @@ public abstract class TestIcebergSourceTablesBase extends SparkTestBase {
private void assertDataFilePartitions(
List<DataFile> dataFiles, List<Integer> expectedPartitionIds) {
- Assert.assertEquals(
- "Table should have " + expectedPartitionIds.size() + " data files",
- expectedPartitionIds.size(),
- dataFiles.size());
+ assertThat(dataFiles)
+ .as("Table should have " + expectedPartitionIds.size() + " data files")
+ .hasSameSizeAs(expectedPartitionIds);
for (int i = 0; i < dataFiles.size(); ++i) {
- Assert.assertEquals(
- "Data file should have partition of id " + expectedPartitionIds.get(i),
- expectedPartitionIds.get(i).intValue(),
- dataFiles.get(i).partition().get(0, Integer.class).intValue());
+ assertThat(dataFiles.get(i).partition().get(0, Integer.class).intValue())
+ .as("Data file should have partition of id " + expectedPartitionIds.get(i))
+ .isEqualTo(expectedPartitionIds.get(i).intValue());
}
}
}
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTableReadableMetrics.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTableReadableMetrics.java
index 9075257fa9..c21ccd0100 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTableReadableMetrics.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTableReadableMetrics.java
@@ -29,6 +29,8 @@ import java.util.List;
import java.util.Map;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.Files;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
@@ -41,19 +43,17 @@ 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.spark.SparkCatalogConfig;
-import org.apache.iceberg.spark.SparkTestBaseWithCatalog;
+import org.apache.iceberg.spark.TestBaseWithCatalog;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.Pair;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
-import org.junit.After;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
-public class TestMetadataTableReadableMetrics extends SparkTestBaseWithCatalog {
-
- @Rule public TemporaryFolder temp = new TemporaryFolder();
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestMetadataTableReadableMetrics extends TestBaseWithCatalog {
private static final Types.StructType LEAF_STRUCT_TYPE =
Types.StructType.of(
@@ -78,9 +78,16 @@ public class TestMetadataTableReadableMetrics extends SparkTestBaseWithCatalog {
optional(8, "fixedCol", Types.FixedType.ofLength(3)),
optional(9, "binaryCol", Types.BinaryType.get()));
- public TestMetadataTableReadableMetrics() {
- // only SparkCatalog supports metadata table sql queries
- super(SparkCatalogConfig.HIVE);
+ @Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ protected static Object[][] parameters() {
+ return new Object[][] {
+ {
+ // only SparkCatalog supports metadata table sql queries
+ SparkCatalogConfig.HIVE.catalogName(),
+ SparkCatalogConfig.HIVE.implementation(),
+ SparkCatalogConfig.HIVE.properties()
+ },
+ };
}
protected String tableName() {
@@ -124,8 +131,7 @@ public class TestMetadataTableReadableMetrics extends SparkTestBaseWithCatalog {
createPrimitiveRecord(
false, 2, 2L, Float.NaN, 2.0D, new BigDecimal("2.00"), "2", null, null));
- DataFile dataFile =
- FileHelpers.writeDataFile(table, Files.localOutput(temp.newFile()), records);
+ DataFile dataFile = FileHelpers.writeDataFile(table, Files.localOutput(temp.toFile()), records);
table.newAppend().appendFile(dataFile).commit();
return table;
}
@@ -143,13 +149,12 @@ public class TestMetadataTableReadableMetrics extends SparkTestBaseWithCatalog {
createNestedRecord(0L, 0.0),
createNestedRecord(1L, Double.NaN),
createNestedRecord(null, null));
- DataFile dataFile =
- FileHelpers.writeDataFile(table, Files.localOutput(temp.newFile()), records);
+ DataFile dataFile = FileHelpers.writeDataFile(table, Files.localOutput(temp.toFile()), records);
table.newAppend().appendFile(dataFile).commit();
return Pair.of(table, dataFile);
}
- @After
+ @AfterEach
public void dropTable() {
sql("DROP TABLE %s", tableName);
}
@@ -192,7 +197,7 @@ public class TestMetadataTableReadableMetrics extends SparkTestBaseWithCatalog {
return record;
}
- @Test
+ @TestTemplate
public void testPrimitiveColumns() throws Exception {
Table table = createPrimitiveTable();
DataFile dataFile = table.currentSnapshot().addedDataFiles(table.io()).iterator().next();
@@ -291,7 +296,7 @@ public class TestMetadataTableReadableMetrics extends SparkTestBaseWithCatalog {
assertEquals("Row should match for entries table", expected, entriesReadableMetrics);
}
- @Test
+ @TestTemplate
public void testSelectPrimitiveValues() throws Exception {
createPrimitiveTable();
@@ -330,7 +335,7 @@ public class TestMetadataTableReadableMetrics extends SparkTestBaseWithCatalog {
sql("SELECT readable_metrics.longCol.value_count, status FROM %s.entries", tableName));
}
- @Test
+ @TestTemplate
public void testSelectNestedValues() throws Exception {
createNestedTable();
@@ -351,7 +356,7 @@ public class TestMetadataTableReadableMetrics extends SparkTestBaseWithCatalog {
entriesReadableMetrics);
}
- @Test
+ @TestTemplate
public void testNestedValues() throws Exception {
Pair<Table, DataFile> table = createNestedTable();
int longColId =
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 13abdad9ef..a417454b45 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
@@ -29,14 +29,18 @@ 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 static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assumptions.assumeThat;
import java.util.Arrays;
import java.util.List;
-import java.util.Map;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
@@ -46,24 +50,20 @@ import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.CatalogTestBase;
import org.apache.iceberg.spark.SparkCatalog;
-import org.apache.iceberg.spark.SparkCatalogTestBase;
import org.apache.iceberg.spark.SparkSessionCatalog;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.catalyst.parser.ParseException;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.StructType;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Assume;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-import org.junit.runners.Parameterized.Parameters;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
-@RunWith(Parameterized.class)
-public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBase {
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestMetadataTablesWithPartitionEvolution extends CatalogTestBase {
@Parameters(name = "catalog = {0}, impl = {1}, conf = {2}, fileFormat = {3}, formatVersion = {4}")
public static Object[][] parameters() {
@@ -119,26 +119,18 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
};
}
- private final FileFormat fileFormat;
- private final int formatVersion;
-
- public TestMetadataTablesWithPartitionEvolution(
- String catalogName,
- String implementation,
- Map<String, String> config,
- FileFormat fileFormat,
- int formatVersion) {
- super(catalogName, implementation, config);
- this.fileFormat = fileFormat;
- this.formatVersion = formatVersion;
- }
+ @Parameter(index = 3)
+ private FileFormat fileFormat;
+
+ @Parameter(index = 4)
+ private int formatVersion;
- @After
+ @AfterEach
public void removeTable() {
sql("DROP TABLE IF EXISTS %s", tableName);
}
- @Test
+ @TestTemplate
public void testFilesMetadataTable() throws ParseException {
createTable("id bigint NOT NULL, category string, data string");
@@ -147,8 +139,9 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
// verify the metadata tables while the current spec is still unpartitioned
for (MetadataTableType tableType : Arrays.asList(FILES, ALL_DATA_FILES)) {
Dataset<Row> df = loadMetadataTable(tableType);
- Assert.assertTrue(
- "Partition must be skipped", df.schema().getFieldIndex("partition").isEmpty());
+ assertThat(df.schema().getFieldIndex("partition").isEmpty())
+ .as("Partition must be skipped")
+ .isTrue();
}
Table table = validationCatalog.loadTable(tableIdent);
@@ -199,7 +192,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
}
}
- @Test
+ @TestTemplate
public void testFilesMetadataTableFilter() throws ParseException {
createTable("id bigint NOT NULL, category string, data string");
sql("ALTER TABLE %s SET TBLPROPERTIES ('%s' 'false')", tableName, MANIFEST_MERGE_ENABLED);
@@ -210,8 +203,9 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
// verify the metadata tables while the current spec is still unpartitioned
for (MetadataTableType tableType : Arrays.asList(FILES, ALL_DATA_FILES)) {
Dataset<Row> df = loadMetadataTable(tableType);
- Assert.assertTrue(
- "Partition must be skipped", df.schema().getFieldIndex("partition").isEmpty());
+ assertThat(df.schema().getFieldIndex("partition").isEmpty())
+ .as("Partition must be skipped")
+ .isTrue();
}
Table table = validationCatalog.loadTable(tableIdent);
@@ -290,7 +284,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
}
}
- @Test
+ @TestTemplate
public void testEntriesMetadataTable() throws ParseException {
createTable("id bigint NOT NULL, category string, data string");
@@ -300,7 +294,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
for (MetadataTableType tableType : Arrays.asList(ENTRIES, ALL_ENTRIES)) {
Dataset<Row> df = loadMetadataTable(tableType);
StructType dataFileType = (StructType) df.schema().apply("data_file").dataType();
- Assert.assertTrue("Partition must be skipped", dataFileType.getFieldIndex("").isEmpty());
+ assertThat(dataFileType.getFieldIndex("").isEmpty()).as("Partition must be skipped").isTrue();
}
Table table = validationCatalog.loadTable(tableIdent);
@@ -351,7 +345,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
}
}
- @Test
+ @TestTemplate
public void testPartitionsTableAddRemoveFields() throws ParseException {
createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'c1', 'd1')", tableName);
@@ -359,8 +353,9 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
// verify the metadata tables while the current spec is still unpartitioned
Dataset<Row> df = loadMetadataTable(PARTITIONS);
- Assert.assertTrue(
- "Partition must be skipped", df.schema().getFieldIndex("partition").isEmpty());
+ assertThat(df.schema().getFieldIndex("partition").isEmpty())
+ .as("Partition must be skipped")
+ .isTrue();
Table table = validationCatalog.loadTable(tableIdent);
@@ -406,7 +401,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
PARTITIONS);
}
- @Test
+ @TestTemplate
public void testPartitionsTableRenameFields() throws ParseException {
createTable("id bigint NOT NULL, category string, data string");
@@ -433,7 +428,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
PARTITIONS);
}
- @Test
+ @TestTemplate
public void testPartitionsTableSwitchFields() throws Exception {
createTable("id bigint NOT NULL, category string, data string");
@@ -495,7 +490,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
}
}
- @Test
+ @TestTemplate
public void testPartitionTableFilterAddRemoveFields() throws ParseException {
// Create un-partitioned table
createTable("id bigint NOT NULL, category string, data string");
@@ -549,13 +544,13 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
"partition.category = 'c2'");
}
- @Test
+ @TestTemplate
public void testPartitionTableFilterSwitchFields() throws Exception {
// Re-added partition fields currently not re-associated:
// https://github.com/apache/iceberg/issues/4292
// In V1, dropped partition fields show separately when field is re-added
// In V2, re-added field currently conflicts with its deleted form
- Assume.assumeTrue(formatVersion == 1);
+ assumeThat(formatVersion).isEqualTo(1);
createTable("id bigint NOT NULL, category string, data string");
sql("INSERT INTO TABLE %s VALUES (1, 'c1', 'd1')", tableName);
@@ -595,7 +590,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
"partition.data = 'd1'");
}
- @Test
+ @TestTemplate
public void testPartitionsTableFilterRenameFields() throws ParseException {
createTable("id bigint NOT NULL, category string, data string");
@@ -618,7 +613,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
"partition.category_another_name = 'c1'");
}
- @Test
+ @TestTemplate
public void testMetadataTablesWithUnknownTransforms() {
createTable("id bigint NOT NULL, category string, data string");
@@ -647,7 +642,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
}
}
- @Test
+ @TestTemplate
public void testPartitionColumnNamedPartition() {
sql(
"CREATE TABLE %s (id int, partition int) USING iceberg PARTITIONED BY (partition)",
@@ -655,7 +650,7 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
sql("INSERT INTO %s VALUES (1, 1), (2, 1), (3, 2), (2, 2)", tableName);
List<Object[]> expected = ImmutableList.of(row(1, 1), row(2, 1), row(3, 2), row(2, 2));
assertEquals("Should return all expected rows", expected, sql("SELECT * FROM %s", tableName));
- Assert.assertEquals(2, sql("SELECT * FROM %s.files", tableName).size());
+ assertThat(sql("SELECT * FROM %s.files", tableName)).hasSize(2);
}
private void assertPartitions(
@@ -681,14 +676,14 @@ public class TestMetadataTablesWithPartitionEvolution extends SparkCatalogTestBa
case FILES:
case ALL_DATA_FILES:
DataType actualFilesType = df.schema().apply("partition").dataType();
- Assert.assertEquals("Partition type must match", expectedType, actualFilesType);
+ assertThat(actualFilesType).as("Partition type must match").isEqualTo(expectedType);
break;
case ENTRIES:
case ALL_ENTRIES:
StructType dataFileType = (StructType) df.schema().apply("data_file").dataType();
DataType actualEntriesType = dataFileType.apply("partition").dataType();
- Assert.assertEquals("Partition type must match", expectedType, actualEntriesType);
+ assertThat(actualEntriesType).as("Partition type must match").isEqualTo(expectedType);
break;
default:
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestFileRewriteCoordinator.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestFileRewriteCoordinator.java
index 666634a06c..085eedf45d 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestFileRewriteCoordinator.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestFileRewriteCoordinator.java
@@ -28,6 +28,7 @@ import java.util.UUID;
import java.util.stream.Collectors;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Table;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
@@ -41,7 +42,9 @@ import org.apache.spark.sql.Row;
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+@ExtendWith(ParameterizedTestExtension.class)
public class TestFileRewriteCoordinator extends CatalogTestBase {
@AfterEach
@@ -102,10 +105,9 @@ public class TestFileRewriteCoordinator extends CatalogTestBase {
table.refresh();
Map<String, String> summary = table.currentSnapshot().summary();
- assertThat(summary.get("deleted-data-files"))
- .as("Deleted files count must match")
- .isEqualTo("4");
- assertThat(summary.get("added-data-files")).as("Added files count must match").isEqualTo("2");
+ assertThat(summary)
+ .containsEntry("deleted-data-files", "4")
+ .containsEntry("added-data-files", "2");
Object rowCount = scalarSql("SELECT count(*) FROM %s", tableName);
assertThat(rowCount).as("Row count must match").isEqualTo(4000L);
@@ -174,10 +176,9 @@ public class TestFileRewriteCoordinator extends CatalogTestBase {
table.refresh();
Map<String, String> summary = table.currentSnapshot().summary();
- assertThat(summary.get("deleted-data-files"))
- .as("Deleted files count must match")
- .isEqualTo("4");
- assertThat(summary.get("added-data-files")).as("Added files count must match").isEqualTo("2");
+ assertThat(summary)
+ .containsEntry("deleted-data-files", "4")
+ .containsEntry("added-data-files", "2");
Object rowCount = scalarSql("SELECT count(*) FROM %s", tableName);
assertThat(rowCount).as("Row count must match").isEqualTo(4000L);
@@ -256,10 +257,9 @@ public class TestFileRewriteCoordinator extends CatalogTestBase {
assertThat(table.snapshots()).as("Should produce 5 snapshots").hasSize(5);
Map<String, String> summary = table.currentSnapshot().summary();
- assertThat(summary.get("deleted-data-files"))
- .as("Deleted files count must match")
- .isEqualTo("4");
- assertThat(summary.get("added-data-files")).as("Added files count must match").isEqualTo("2");
+ assertThat(summary)
+ .containsEntry("deleted-data-files", "4")
+ .containsEntry("added-data-files", "2");
Object rowCount = scalarSql("SELECT count(*) FROM %s", tableName);
assertThat(rowCount).as("Row count must match").isEqualTo(4000L);
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkCachedTableCatalog.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkCachedTableCatalog.java
index eaf2308659..228bf43b89 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkCachedTableCatalog.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkCachedTableCatalog.java
@@ -18,6 +18,7 @@
*/
package org.apache.iceberg.spark;
+import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Parameters;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
@@ -25,7 +26,9 @@ import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+@ExtendWith(ParameterizedTestExtension.class)
public class TestSparkCachedTableCatalog extends TestBaseWithCatalog {
private static final SparkTableCache TABLE_CACHE = SparkTableCache.get();
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkCatalogOperations.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkCatalogOperations.java
index 5b460a0b52..d7ead52880 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkCatalogOperations.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkCatalogOperations.java
@@ -19,8 +19,10 @@
package org.apache.iceberg.spark;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.atIndex;
import java.util.concurrent.ThreadLocalRandom;
+import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Parameters;
import org.apache.iceberg.Schema;
import org.apache.iceberg.catalog.Catalog;
@@ -36,7 +38,9 @@ import org.apache.spark.sql.types.DataTypes;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+@ExtendWith(ParameterizedTestExtension.class)
public class TestSparkCatalogOperations extends CatalogTestBase {
private static final boolean USE_NULLABLE_QUERY_SCHEMA =
ThreadLocalRandom.current().nextBoolean();
@@ -108,9 +112,9 @@ public class TestSparkCatalogOperations extends CatalogTestBase {
assertThat(table).as("Should return updated table").isNotNull();
Column expectedField = Column.create(fieldName, DataTypes.StringType, true);
- assertThat(table.columns()[2])
+ assertThat(table.columns())
.as("Adding a column to a table should return the updated table with the new column")
- .isEqualTo(expectedField);
+ .contains(expectedField, atIndex(2));
assertThat(table.properties())
.as(
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java
index 6041c22e24..a9b5d1a237 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java
@@ -52,6 +52,7 @@ import java.util.List;
import java.util.Map;
import org.apache.iceberg.DistributionMode;
import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.UpdateProperties;
@@ -62,7 +63,9 @@ import org.apache.spark.sql.internal.SQLConf;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+@ExtendWith(ParameterizedTestExtension.class)
public class TestSparkWriteConf extends TestBaseWithCatalog {
@BeforeEach
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestDataSourceOptions.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestDataSourceOptions.java
index c4ba96e634..a7702b169a 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestDataSourceOptions.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestDataSourceOptions.java
@@ -24,7 +24,6 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.IOException;
import java.math.RoundingMode;
-import java.nio.file.Path;
import java.util.List;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
@@ -32,6 +31,7 @@ import org.apache.iceberg.DataFile;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
@@ -61,8 +61,9 @@ import org.apache.spark.sql.functions;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.TestTemplate;
-import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.api.extension.ExtendWith;
+@ExtendWith(ParameterizedTestExtension.class)
public class TestDataSourceOptions extends TestBaseWithCatalog {
private static final Configuration CONF = new Configuration();
@@ -71,8 +72,6 @@ public class TestDataSourceOptions extends TestBaseWithCatalog {
optional(1, "id", Types.IntegerType.get()), optional(2, "data", Types.StringType.get()));
private static SparkSession spark = null;
- @TempDir private Path temp;
-
@BeforeAll
public static void startSpark() {
TestDataSourceOptions.spark = SparkSession.builder().master("local[2]").getOrCreate();
@@ -372,7 +371,7 @@ public class TestDataSourceOptions extends TestBaseWithCatalog {
}
@TestTemplate
- public void testDefaultMetadataSplitSize() throws IOException {
+ public void testDefaultMetadataSplitSize() {
String tableLocation = temp.resolve("iceberg-table").toFile().toString();
HadoopTables tables = new HadoopTables(CONF);
@@ -468,7 +467,7 @@ public class TestDataSourceOptions extends TestBaseWithCatalog {
List<Snapshot> snapshots = Lists.newArrayList(table.snapshots());
assertThat(snapshots).hasSize(2);
- assertThat(snapshots.get(0).summary().get("writer-thread")).isNull();
+ assertThat(snapshots.get(0).summary()).doesNotContainKey("writer-thread");
assertThat(snapshots.get(1).summary())
.containsEntry("writer-thread", "test-extra-commit-message-writer-thread")
.containsEntry("extra-key", "someValue")
@@ -512,7 +511,7 @@ public class TestDataSourceOptions extends TestBaseWithCatalog {
List<Snapshot> snapshots = Lists.newArrayList(table.snapshots());
assertThat(snapshots).hasSize(2);
- assertThat(snapshots.get(0).summary().get("writer-thread")).isNull();
+ assertThat(snapshots.get(0).summary()).doesNotContainKey("writer-thread");
assertThat(snapshots.get(1).summary())
.containsEntry("writer-thread", "test-extra-commit-message-delete-thread")
.containsEntry("extra-key", "someValue")
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java
index 35d6e119e8..701c421a25 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceHadoopTables.java
@@ -36,7 +36,7 @@ public class TestIcebergSourceHadoopTables extends TestIcebergSourceTablesBase {
String tableLocation = null;
@BeforeEach
- public void setupTable() throws Exception {
+ public void setupTable() {
this.tableLocation = tableDir.toURI().toString();
}
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java
index 225c9540a0..63994726ca 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java
@@ -237,8 +237,7 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.select("status")
.collectAsList();
- assertThat(actual).as("Results should contain only one status").hasSize(1);
- assertThat(actual.get(0).getInt(0)).as("That status should be Added (1)").isEqualTo(1);
+ assertThat(actual).singleElement().satisfies(row -> assertThat(row.getInt(0)).isEqualTo(1));
}
@Test
@@ -600,10 +599,20 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
long snapshotId = table.currentSnapshot().snapshotId();
assertThat(actual).as("Entries table should have 2 rows").hasSize(2);
- assertThat(actual.get(0).getLong(0)).as("Sequence number must match").isEqualTo(0);
- assertThat(actual.get(0).getLong(1)).as("Snapshot id must match").isEqualTo(snapshotId);
- assertThat(actual.get(1).getLong(0)).as("Sequence number must match").isEqualTo(0);
- assertThat(actual.get(1).getLong(1)).as("Snapshot id must match").isEqualTo(snapshotId);
+ assertThat(actual)
+ .first()
+ .satisfies(
+ row -> {
+ assertThat(row.getLong(0)).isEqualTo(0);
+ assertThat(row.getLong(1)).isEqualTo(snapshotId);
+ });
+ assertThat(actual)
+ .element(1)
+ .satisfies(
+ row -> {
+ assertThat(row.getLong(0)).isEqualTo(0);
+ assertThat(row.getLong(1)).isEqualTo(snapshotId);
+ });
}
@Test
@@ -1490,7 +1499,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
assertThat(rewriteManifestResult.rewrittenManifests())
.as("rewrite replaced 2 manifests")
.hasSize(2);
-
assertThat(rewriteManifestResult.addedManifests()).as("rewrite added 1 manifests").hasSize(1);
List<Row> actual =
@@ -1747,9 +1755,9 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
table.refresh();
Dataset<Row> resultDf = spark.read().format("iceberg").load(loadLocation(tableIdentifier));
- assertThat(originalRecords)
+ assertThat(resultDf.orderBy("id").collectAsList())
.as("Records should match")
- .isEqualTo(resultDf.orderBy("id").collectAsList());
+ .containsExactlyElementsOf(originalRecords);
Snapshot snapshotBeforeAddColumn = table.currentSnapshot();
@@ -1778,9 +1786,9 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
RowFactory.create(5, "xyz", "C"));
Dataset<Row> resultDf2 = spark.read().format("iceberg").load(loadLocation(tableIdentifier));
- assertThat(updatedRecords)
+ assertThat(resultDf2.orderBy("id").collectAsList())
.as("Records should match")
- .isEqualTo(resultDf2.orderBy("id").collectAsList());
+ .containsExactlyElementsOf(updatedRecords);
Dataset<Row> resultDf3 =
spark
@@ -1788,11 +1796,9 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.format("iceberg")
.option(SparkReadOptions.SNAPSHOT_ID, snapshotBeforeAddColumn.snapshotId())
.load(loadLocation(tableIdentifier));
-
- assertThat(originalRecords)
+ assertThat(resultDf3.orderBy("id").collectAsList())
.as("Records should match")
- .isEqualTo(resultDf3.orderBy("id").collectAsList());
-
+ .containsExactlyElementsOf(originalRecords);
assertThat(resultDf3.schema()).as("Schemas should match").isEqualTo(originalSparkSchema);
}
@@ -1819,10 +1825,9 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
table.refresh();
Dataset<Row> resultDf = spark.read().format("iceberg").load(loadLocation(tableIdentifier));
-
assertThat(resultDf.orderBy("id").collectAsList())
.as("Records should match")
- .isEqualTo(originalRecords);
+ .containsExactlyElementsOf(originalRecords);
long tsBeforeDropColumn = waitUntilAfter(System.currentTimeMillis());
table.updateSchema().deleteColumn("data").commit();
@@ -1852,7 +1857,7 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
Dataset<Row> resultDf2 = spark.read().format("iceberg").load(loadLocation(tableIdentifier));
assertThat(resultDf2.orderBy("id").collectAsList())
.as("Records should match")
- .isEqualTo(updatedRecords);
+ .containsExactlyElementsOf(updatedRecords);
Dataset<Row> resultDf3 =
spark
@@ -1860,11 +1865,9 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.format("iceberg")
.option(SparkReadOptions.AS_OF_TIMESTAMP, tsBeforeDropColumn)
.load(loadLocation(tableIdentifier));
-
assertThat(resultDf3.orderBy("id").collectAsList())
.as("Records should match")
- .isEqualTo(originalRecords);
-
+ .containsExactlyElementsOf(originalRecords);
assertThat(resultDf3.schema()).as("Schemas should match").isEqualTo(originalSparkSchema);
// At tsAfterDropColumn, there has been a schema change, but no new snapshot,
@@ -1875,11 +1878,9 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.format("iceberg")
.option(SparkReadOptions.AS_OF_TIMESTAMP, tsAfterDropColumn)
.load(loadLocation(tableIdentifier));
-
assertThat(resultDf4.orderBy("id").collectAsList())
.as("Records should match")
- .isEqualTo(originalRecords);
-
+ .containsExactlyElementsOf(originalRecords);
assertThat(resultDf4.schema()).as("Schemas should match").isEqualTo(originalSparkSchema);
}
@@ -1904,10 +1905,9 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
table.refresh();
Dataset<Row> resultDf = spark.read().format("iceberg").load(loadLocation(tableIdentifier));
-
assertThat(resultDf.orderBy("id").collectAsList())
.as("Records should match")
- .isEqualTo(originalRecords);
+ .containsExactlyElementsOf(originalRecords);
Snapshot snapshotBeforeAddColumn = table.currentSnapshot();
@@ -1936,10 +1936,9 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
RowFactory.create(5, "xyz", "C"));
Dataset<Row> resultDf2 = spark.read().format("iceberg").load(loadLocation(tableIdentifier));
-
assertThat(resultDf2.orderBy("id").collectAsList())
.as("Records should match")
- .isEqualTo(updatedRecords);
+ .containsExactlyElementsOf(updatedRecords);
table.updateSchema().deleteColumn("data").commit();
@@ -1952,10 +1951,9 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
RowFactory.create(5, "C"));
Dataset<Row> resultDf3 = spark.read().format("iceberg").load(loadLocation(tableIdentifier));
-
assertThat(resultDf3.orderBy("id").collectAsList())
.as("Records should match")
- .isEqualTo(recordsAfterDropColumn);
+ .containsExactlyElementsOf(recordsAfterDropColumn);
Dataset<Row> resultDf4 =
spark
@@ -1963,11 +1961,9 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.format("iceberg")
.option(SparkReadOptions.SNAPSHOT_ID, snapshotBeforeAddColumn.snapshotId())
.load(loadLocation(tableIdentifier));
-
assertThat(resultDf4.orderBy("id").collectAsList())
.as("Records should match")
- .isEqualTo(originalRecords);
-
+ .containsExactlyElementsOf(originalRecords);
assertThat(resultDf4.schema()).as("Schemas should match").isEqualTo(originalSparkSchema);
}
@@ -1999,19 +1995,16 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.location(table.location() + "/metadata")
.olderThan(System.currentTimeMillis())
.execute();
-
assertThat(result1.orphanFileLocations()).as("Should not delete any metadata files").isEmpty();
DeleteOrphanFiles.Result result2 =
actions.deleteOrphanFiles(table).olderThan(System.currentTimeMillis()).execute();
-
assertThat(result2.orphanFileLocations()).as("Should delete 1 data file").hasSize(1);
Dataset<Row> resultDF = spark.read().format("iceberg").load(loadLocation(tableIdentifier));
List<SimpleRecord> actualRecords =
resultDF.as(Encoders.bean(SimpleRecord.class)).collectAsList();
-
- assertThat(actualRecords).as("Rows must match").isEqualTo(records);
+ assertThat(actualRecords).as("Rows must match").containsExactlyInAnyOrderElementsOf(records);
}
@Test
@@ -2056,9 +2049,7 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.map(r -> (Integer) r.getAs(DataFile.SPEC_ID.name()))
.collect(Collectors.toList());
- assertThat(actual)
- .as("Should have two partition specs")
- .isEqualTo(ImmutableList.of(spec0, spec1));
+ assertThat(actual).as("Should have two partition specs").containsExactly(spec0, spec1);
}
@Test
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTableReadableMetrics.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTableReadableMetrics.java
index 547ab32eac..c21ccd0100 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTableReadableMetrics.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTableReadableMetrics.java
@@ -24,12 +24,12 @@ import static org.apache.iceberg.types.Types.NestedField.required;
import java.io.IOException;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
-import java.nio.file.Path;
import java.util.Base64;
import java.util.List;
import java.util.Map;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.Files;
+import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Parameters;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
@@ -50,12 +50,11 @@ import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.TestTemplate;
-import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.api.extension.ExtendWith;
+@ExtendWith(ParameterizedTestExtension.class)
public class TestMetadataTableReadableMetrics extends TestBaseWithCatalog {
- @TempDir private Path temp;
-
private static final Types.StructType LEAF_STRUCT_TYPE =
Types.StructType.of(
optional(1, "leafLongCol", Types.LongType.get()),