From 460282433756620310fd346e519b18b74a745305 Mon Sep 17 00:00:00 2001 From: Chinmay Bhat <12948588+chinmay-bhat@users.noreply.github.com> Date: Fri, 5 Jan 2024 21:48:20 +0530 Subject: [PATCH] Spark 3.5: Migrate tests to JUnit5 (#9417) --- .../org/apache/iceberg/TaskCheckHelper.java | 99 ++++---- .../iceberg/TestDataFileSerialization.java | 23 +- .../iceberg/TestFileIOSerialization.java | 31 +-- ...TestHadoopMetricsContextSerialization.java | 2 +- .../TestManifestFileSerialization.java | 108 +++++---- .../iceberg/TestScanTaskSerialization.java | 45 ++-- .../iceberg/TestTableSerialization.java | 49 ++-- .../org/apache/iceberg/ValidationHelpers.java | 5 +- .../org/apache/iceberg/spark/TestBase.java | 6 +- .../iceberg/spark/TestBaseWithCatalog.java | 9 +- .../iceberg/spark/TestChangelogIterator.java | 16 +- .../spark/TestFileRewriteCoordinator.java | 53 +++-- .../iceberg/spark/TestFunctionCatalog.java | 84 ++++--- .../apache/iceberg/spark/TestSpark3Util.java | 109 +++++---- .../spark/TestSparkCachedTableCatalog.java | 26 +- .../spark/TestSparkCatalogOperations.java | 44 ++-- .../spark/TestSparkCompressionUtil.java | 6 +- .../TestSparkDistributionAndOrderingUtil.java | 222 +++++++++--------- .../iceberg/spark/TestSparkFilters.java | 84 +++---- .../iceberg/spark/TestSparkSchemaUtil.java | 27 +-- .../spark/TestSparkSessionCatalog.java | 66 +++--- .../iceberg/spark/TestSparkTableUtil.java | 36 ++- .../iceberg/spark/TestSparkV2Filters.java | 150 ++++++------ .../spark/TestSparkValueConverter.java | 12 +- .../iceberg/spark/TestSparkWriteConf.java | 62 ++--- 25 files changed, 685 insertions(+), 689 deletions(-) diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/TaskCheckHelper.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/TaskCheckHelper.java index c44bacf149b5..668f410091dd 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/TaskCheckHelper.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/TaskCheckHelper.java @@ -18,10 +18,11 @@ */ package org.apache.iceberg; +import static org.assertj.core.api.Assertions.assertThat; + import java.util.Comparator; import java.util.List; import java.util.stream.Collectors; -import org.junit.Assert; public final class TaskCheckHelper { private TaskCheckHelper() {} @@ -31,8 +32,9 @@ public static void assertEquals( List expectedTasks = getFileScanTasksInFilePathOrder(expected); List actualTasks = getFileScanTasksInFilePathOrder(actual); - Assert.assertEquals( - "The number of file scan tasks should match", expectedTasks.size(), actualTasks.size()); + assertThat(actualTasks) + .as("The number of file scan tasks should match") + .hasSameSizeAs(expectedTasks); for (int i = 0; i < expectedTasks.size(); i++) { FileScanTask expectedTask = expectedTasks.get(i); @@ -45,60 +47,57 @@ public static void assertEquals(FileScanTask expected, FileScanTask actual) { assertEquals(expected.file(), actual.file()); // PartitionSpec implements its own equals method - Assert.assertEquals("PartitionSpec doesn't match", expected.spec(), actual.spec()); + assertThat(actual.spec()).as("PartitionSpec doesn't match").isEqualTo(expected.spec()); - Assert.assertEquals("starting position doesn't match", expected.start(), actual.start()); + assertThat(actual.start()).as("starting position doesn't match").isEqualTo(expected.start()); - Assert.assertEquals( - "the number of bytes to scan doesn't match", expected.start(), actual.start()); + assertThat(actual.start()) + .as("the number of bytes to scan doesn't match") + .isEqualTo(expected.start()); // simplify comparison on residual expression via comparing toString - Assert.assertEquals( - "Residual expression doesn't match", - expected.residual().toString(), - actual.residual().toString()); + assertThat(actual.residual().toString()) + .as("Residual expression doesn't match") + .isEqualTo(expected.residual().toString()); } public static void assertEquals(DataFile expected, DataFile actual) { - Assert.assertEquals("Should match the serialized record path", expected.path(), actual.path()); - Assert.assertEquals( - "Should match the serialized record format", expected.format(), actual.format()); - Assert.assertEquals( - "Should match the serialized record partition", - expected.partition().get(0, Object.class), - actual.partition().get(0, Object.class)); - Assert.assertEquals( - "Should match the serialized record count", expected.recordCount(), actual.recordCount()); - Assert.assertEquals( - "Should match the serialized record size", - expected.fileSizeInBytes(), - actual.fileSizeInBytes()); - Assert.assertEquals( - "Should match the serialized record value counts", - expected.valueCounts(), - actual.valueCounts()); - Assert.assertEquals( - "Should match the serialized record null value counts", - expected.nullValueCounts(), - actual.nullValueCounts()); - Assert.assertEquals( - "Should match the serialized record lower bounds", - expected.lowerBounds(), - actual.lowerBounds()); - Assert.assertEquals( - "Should match the serialized record upper bounds", - expected.upperBounds(), - actual.upperBounds()); - Assert.assertEquals( - "Should match the serialized record key metadata", - expected.keyMetadata(), - actual.keyMetadata()); - Assert.assertEquals( - "Should match the serialized record offsets", - expected.splitOffsets(), - actual.splitOffsets()); - Assert.assertEquals( - "Should match the serialized record offsets", expected.keyMetadata(), actual.keyMetadata()); + assertThat(actual.path()) + .as("Should match the serialized record path") + .isEqualTo(expected.path()); + assertThat(actual.format()) + .as("Should match the serialized record format") + .isEqualTo(expected.format()); + assertThat(actual.partition().get(0, Object.class)) + .as("Should match the serialized record partition") + .isEqualTo(expected.partition().get(0, Object.class)); + assertThat(actual.recordCount()) + .as("Should match the serialized record count") + .isEqualTo(expected.recordCount()); + assertThat(actual.fileSizeInBytes()) + .as("Should match the serialized record size") + .isEqualTo(expected.fileSizeInBytes()); + assertThat(actual.valueCounts()) + .as("Should match the serialized record value counts") + .isEqualTo(expected.valueCounts()); + assertThat(actual.nullValueCounts()) + .as("Should match the serialized record null value counts") + .isEqualTo(expected.nullValueCounts()); + assertThat(actual.lowerBounds()) + .as("Should match the serialized record lower bounds") + .isEqualTo(expected.lowerBounds()); + assertThat(actual.upperBounds()) + .as("Should match the serialized record upper bounds") + .isEqualTo(expected.upperBounds()); + assertThat(actual.keyMetadata()) + .as("Should match the serialized record key metadata") + .isEqualTo(expected.keyMetadata()); + assertThat(actual.splitOffsets()) + .as("Should match the serialized record offsets") + .isEqualTo(expected.splitOffsets()); + assertThat(actual.keyMetadata()) + .as("Should match the serialized record offsets") + .isEqualTo(expected.keyMetadata()); } private static List getFileScanTasksInFilePathOrder( diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestDataFileSerialization.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestDataFileSerialization.java index 33b5316b72b7..57c4dc7cdf23 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestDataFileSerialization.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestDataFileSerialization.java @@ -21,6 +21,7 @@ import static org.apache.iceberg.TaskCheckHelper.assertEquals; import static org.apache.iceberg.types.Types.NestedField.optional; import static org.apache.iceberg.types.Types.NestedField.required; +import static org.assertj.core.api.Assertions.assertThat; import com.esotericsoftware.kryo.Kryo; import com.esotericsoftware.kryo.io.Input; @@ -35,6 +36,7 @@ import java.io.ObjectOutputStream; import java.nio.ByteBuffer; import java.nio.ByteOrder; +import java.nio.file.Path; import java.util.Map; import java.util.UUID; import org.apache.iceberg.io.FileAppender; @@ -48,11 +50,8 @@ import org.apache.spark.SparkConf; import org.apache.spark.serializer.KryoSerializer; import org.apache.spark.sql.catalyst.InternalRow; -import org.assertj.core.api.Assertions; -import org.junit.Assert; -import org.junit.Rule; -import org.junit.Test; -import org.junit.rules.TemporaryFolder; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; public class TestDataFileSerialization { @@ -102,12 +101,12 @@ public class TestDataFileSerialization { .withSortOrder(SortOrder.unsorted()) .build(); - @Rule public TemporaryFolder temp = new TemporaryFolder(); + @TempDir private Path temp; @Test public void testDataFileKryoSerialization() throws Exception { - File data = temp.newFile(); - Assert.assertTrue(data.delete()); + File data = File.createTempFile("junit", null, temp.toFile()); + assertThat(data.delete()).isTrue(); Kryo kryo = new KryoSerializer(new SparkConf()).newKryo(); try (Output out = new Output(new FileOutputStream(data))) { @@ -118,7 +117,7 @@ public void testDataFileKryoSerialization() throws Exception { try (Input in = new Input(new FileInputStream(data))) { for (int i = 0; i < 2; i += 1) { Object obj = kryo.readClassAndObject(in); - Assertions.assertThat(obj).as("Should be a DataFile").isInstanceOf(DataFile.class); + assertThat(obj).as("Should be a DataFile").isInstanceOf(DataFile.class); assertEquals(DATA_FILE, (DataFile) obj); } } @@ -136,7 +135,7 @@ public void testDataFileJavaSerialization() throws Exception { new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) { for (int i = 0; i < 2; i += 1) { Object obj = in.readObject(); - Assertions.assertThat(obj).as("Should be a DataFile").isInstanceOf(DataFile.class); + assertThat(obj).as("Should be a DataFile").isInstanceOf(DataFile.class); assertEquals(DATA_FILE, (DataFile) obj); } } @@ -146,7 +145,7 @@ public void testDataFileJavaSerialization() throws Exception { public void testParquetWriterSplitOffsets() throws IOException { Iterable records = RandomData.generateSpark(DATE_SCHEMA, 1, 33L); File parquetFile = - new File(temp.getRoot(), FileFormat.PARQUET.addExtension(UUID.randomUUID().toString())); + new File(temp.toFile(), FileFormat.PARQUET.addExtension(UUID.randomUUID().toString())); FileAppender writer = Parquet.write(Files.localOutput(parquetFile)) .schema(DATE_SCHEMA) @@ -161,7 +160,7 @@ public void testParquetWriterSplitOffsets() throws IOException { } Kryo kryo = new KryoSerializer(new SparkConf()).newKryo(); - File dataFile = temp.newFile(); + File dataFile = File.createTempFile("junit", null, temp.toFile()); try (Output out = new Output(new FileOutputStream(dataFile))) { kryo.writeClassAndObject(out, writer.splitOffsets()); } diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestFileIOSerialization.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestFileIOSerialization.java index c6f491ece5ad..bfdfa8deca06 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestFileIOSerialization.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestFileIOSerialization.java @@ -20,9 +20,12 @@ import static org.apache.iceberg.types.Types.NestedField.optional; import static org.apache.iceberg.types.Types.NestedField.required; +import static org.assertj.core.api.Assertions.assertThat; import java.io.File; import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.Map; import org.apache.hadoop.conf.Configuration; import org.apache.iceberg.hadoop.HadoopFileIO; @@ -32,11 +35,9 @@ import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.spark.source.SerializableTableWithSize; import org.apache.iceberg.types.Types; -import org.junit.Assert; -import org.junit.Before; -import org.junit.Rule; -import org.junit.Test; -import org.junit.rules.TemporaryFolder; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; public class TestFileIOSerialization { @@ -60,15 +61,15 @@ public class TestFileIOSerialization { CONF.set("k2", "v2"); } - @Rule public TemporaryFolder temp = new TemporaryFolder(); + @TempDir private Path temp; private Table table; - @Before + @BeforeEach public void initTable() throws IOException { Map props = ImmutableMap.of("k1", "v1", "k2", "v2"); - File tableLocation = temp.newFolder(); - Assert.assertTrue(tableLocation.delete()); + File tableLocation = Files.createTempDirectory(temp, "junit").toFile(); + assertThat(tableLocation.delete()).isTrue(); this.table = TABLES.create(SCHEMA, SPEC, SORT_ORDER, props, tableLocation.toString()); } @@ -82,9 +83,9 @@ public void testHadoopFileIOKryoSerialization() throws IOException { FileIO deserializedIO = KryoHelpers.roundTripSerialize(serializableTable.io()); Configuration actualConf = ((HadoopFileIO) deserializedIO).conf(); - Assert.assertEquals("Conf pairs must match", toMap(expectedConf), toMap(actualConf)); - Assert.assertEquals("Conf values must be present", "v1", actualConf.get("k1")); - Assert.assertEquals("Conf values must be present", "v2", actualConf.get("k2")); + assertThat(toMap(actualConf)).as("Conf pairs must match").isEqualTo(toMap(expectedConf)); + assertThat(actualConf.get("k1")).as("Conf values must be present").isEqualTo("v1"); + assertThat(actualConf.get("k2")).as("Conf values must be present").isEqualTo("v2"); } @Test @@ -96,9 +97,9 @@ public void testHadoopFileIOJavaSerialization() throws IOException, ClassNotFoun FileIO deserializedIO = TestHelpers.roundTripSerialize(serializableTable.io()); Configuration actualConf = ((HadoopFileIO) deserializedIO).conf(); - Assert.assertEquals("Conf pairs must match", toMap(expectedConf), toMap(actualConf)); - Assert.assertEquals("Conf values must be present", "v1", actualConf.get("k1")); - Assert.assertEquals("Conf values must be present", "v2", actualConf.get("k2")); + assertThat(toMap(actualConf)).as("Conf pairs must match").isEqualTo(toMap(expectedConf)); + assertThat(actualConf.get("k1")).as("Conf values must be present").isEqualTo("v1"); + assertThat(actualConf.get("k2")).as("Conf values must be present").isEqualTo("v2"); } private Map toMap(Configuration conf) { diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestHadoopMetricsContextSerialization.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestHadoopMetricsContextSerialization.java index 92d233e129e2..a4643d7a087b 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestHadoopMetricsContextSerialization.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestHadoopMetricsContextSerialization.java @@ -23,7 +23,7 @@ import org.apache.iceberg.io.FileIOMetricsContext; import org.apache.iceberg.metrics.MetricsContext; import org.apache.iceberg.relocated.com.google.common.collect.Maps; -import org.junit.Test; +import org.junit.jupiter.api.Test; public class TestHadoopMetricsContextSerialization { diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestManifestFileSerialization.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestManifestFileSerialization.java index 92a646d3861b..1e09917d0305 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestManifestFileSerialization.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestManifestFileSerialization.java @@ -20,6 +20,7 @@ import static org.apache.iceberg.types.Types.NestedField.optional; import static org.apache.iceberg.types.Types.NestedField.required; +import static org.assertj.core.api.Assertions.assertThat; import com.esotericsoftware.kryo.Kryo; import com.esotericsoftware.kryo.io.Input; @@ -34,6 +35,7 @@ import java.io.ObjectOutputStream; import java.nio.ByteBuffer; import java.nio.ByteOrder; +import java.nio.file.Path; import org.apache.hadoop.conf.Configuration; import org.apache.iceberg.ManifestFile.PartitionFieldSummary; import org.apache.iceberg.hadoop.HadoopFileIO; @@ -43,11 +45,8 @@ import org.apache.iceberg.types.Types; import org.apache.spark.SparkConf; import org.apache.spark.serializer.KryoSerializer; -import org.assertj.core.api.Assertions; -import org.junit.Assert; -import org.junit.Rule; -import org.junit.Test; -import org.junit.rules.TemporaryFolder; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; public class TestManifestFileSerialization { @@ -99,12 +98,12 @@ public class TestManifestFileSerialization { private static final FileIO FILE_IO = new HadoopFileIO(new Configuration()); - @Rule public TemporaryFolder temp = new TemporaryFolder(); + @TempDir private Path temp; @Test public void testManifestFileKryoSerialization() throws IOException { - File data = temp.newFile(); - Assert.assertTrue(data.delete()); + File data = File.createTempFile("junit", null, temp.toFile()); + assertThat(data.delete()).isTrue(); Kryo kryo = new KryoSerializer(new SparkConf()).newKryo(); @@ -119,7 +118,7 @@ public void testManifestFileKryoSerialization() throws IOException { try (Input in = new Input(new FileInputStream(data))) { for (int i = 0; i < 3; i += 1) { Object obj = kryo.readClassAndObject(in); - Assertions.assertThat(obj).as("Should be a ManifestFile").isInstanceOf(ManifestFile.class); + assertThat(obj).as("Should be a ManifestFile").isInstanceOf(ManifestFile.class); checkManifestFile(manifest, (ManifestFile) obj); } } @@ -141,62 +140,67 @@ public void testManifestFileJavaSerialization() throws Exception { new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) { for (int i = 0; i < 3; i += 1) { Object obj = in.readObject(); - Assertions.assertThat(obj).as("Should be a ManifestFile").isInstanceOf(ManifestFile.class); + assertThat(obj).as("Should be a ManifestFile").isInstanceOf(ManifestFile.class); checkManifestFile(manifest, (ManifestFile) obj); } } } private void checkManifestFile(ManifestFile expected, ManifestFile actual) { - Assert.assertEquals("Path must match", expected.path(), actual.path()); - Assert.assertEquals("Length must match", expected.length(), actual.length()); - Assert.assertEquals("Spec id must match", expected.partitionSpecId(), actual.partitionSpecId()); - Assert.assertEquals("Snapshot id must match", expected.snapshotId(), actual.snapshotId()); - Assert.assertEquals( - "Added files flag must match", expected.hasAddedFiles(), actual.hasAddedFiles()); - Assert.assertEquals( - "Added files count must match", expected.addedFilesCount(), actual.addedFilesCount()); - Assert.assertEquals( - "Added rows count must match", expected.addedRowsCount(), actual.addedRowsCount()); - Assert.assertEquals( - "Existing files flag must match", expected.hasExistingFiles(), actual.hasExistingFiles()); - Assert.assertEquals( - "Existing files count must match", - expected.existingFilesCount(), - actual.existingFilesCount()); - Assert.assertEquals( - "Existing rows count must match", expected.existingRowsCount(), actual.existingRowsCount()); - Assert.assertEquals( - "Deleted files flag must match", expected.hasDeletedFiles(), actual.hasDeletedFiles()); - Assert.assertEquals( - "Deleted files count must match", expected.deletedFilesCount(), actual.deletedFilesCount()); - Assert.assertEquals( - "Deleted rows count must match", expected.deletedRowsCount(), actual.deletedRowsCount()); + assertThat(actual.path()).as("Path must match").isEqualTo(expected.path()); + assertThat(actual.length()).as("Length must match").isEqualTo(expected.length()); + assertThat(actual.partitionSpecId()) + .as("Spec id must match") + .isEqualTo(expected.partitionSpecId()); + assertThat(actual.snapshotId()).as("Snapshot id must match").isEqualTo(expected.snapshotId()); + assertThat(actual.hasAddedFiles()) + .as("Added files flag must match") + .isEqualTo(expected.hasAddedFiles()); + assertThat(actual.addedFilesCount()) + .as("Added files count must match") + .isEqualTo(expected.addedFilesCount()); + assertThat(actual.addedRowsCount()) + .as("Added rows count must match") + .isEqualTo(expected.addedRowsCount()); + assertThat(actual.hasExistingFiles()) + .as("Existing files flag must match") + .isEqualTo(expected.hasExistingFiles()); + assertThat(actual.existingFilesCount()) + .as("Existing files count must match") + .isEqualTo(expected.existingFilesCount()); + assertThat(actual.existingRowsCount()) + .as("Existing rows count must match") + .isEqualTo(expected.existingRowsCount()); + assertThat(actual.hasDeletedFiles()) + .as("Deleted files flag must match") + .isEqualTo(expected.hasDeletedFiles()); + assertThat(actual.deletedFilesCount()) + .as("Deleted files count must match") + .isEqualTo(expected.deletedFilesCount()); + assertThat(actual.deletedRowsCount()) + .as("Deleted rows count must match") + .isEqualTo(expected.deletedRowsCount()); PartitionFieldSummary expectedPartition = expected.partitions().get(0); PartitionFieldSummary actualPartition = actual.partitions().get(0); - Assert.assertEquals( - "Null flag in partition must match", - expectedPartition.containsNull(), - actualPartition.containsNull()); - Assert.assertEquals( - "NaN flag in partition must match", - expectedPartition.containsNaN(), - actualPartition.containsNaN()); - Assert.assertEquals( - "Lower bounds in partition must match", - expectedPartition.lowerBound(), - actualPartition.lowerBound()); - Assert.assertEquals( - "Upper bounds in partition must match", - expectedPartition.upperBound(), - actualPartition.upperBound()); + assertThat(actualPartition.containsNull()) + .as("Null flag in partition must match") + .isEqualTo(expectedPartition.containsNull()); + assertThat(actualPartition.containsNaN()) + .as("NaN flag in partition must match") + .isEqualTo(expectedPartition.containsNaN()); + assertThat(actualPartition.lowerBound()) + .as("Lower bounds in partition must match") + .isEqualTo(expectedPartition.lowerBound()); + assertThat(actualPartition.upperBound()) + .as("Upper bounds in partition must match") + .isEqualTo(expectedPartition.upperBound()); } private ManifestFile writeManifest(DataFile... files) throws IOException { - File manifestFile = temp.newFile("input.m0.avro"); - Assert.assertTrue(manifestFile.delete()); + File manifestFile = File.createTempFile("input.m0", ".avro", temp.toFile()); + assertThat(manifestFile.delete()).isTrue(); OutputFile outputFile = FILE_IO.newOutputFile(manifestFile.getCanonicalPath()); ManifestWriter writer = ManifestFiles.write(SPEC, outputFile); diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestScanTaskSerialization.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestScanTaskSerialization.java index 14e9e7f605b5..bc094292afa5 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestScanTaskSerialization.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestScanTaskSerialization.java @@ -19,6 +19,7 @@ package org.apache.iceberg; import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.assertj.core.api.Assertions.assertThat; import com.esotericsoftware.kryo.Kryo; import com.esotericsoftware.kryo.io.Input; @@ -31,6 +32,7 @@ import java.io.ObjectInputStream; import java.io.ObjectOutputStream; import java.nio.file.Files; +import java.nio.file.Path; import java.util.List; import java.util.Map; import org.apache.hadoop.conf.Configuration; @@ -39,21 +41,18 @@ import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Maps; -import org.apache.iceberg.spark.SparkTestBase; +import org.apache.iceberg.spark.TestBase; import org.apache.iceberg.spark.source.ThreeColumnRecord; import org.apache.iceberg.types.Types; import org.apache.spark.SparkConf; import org.apache.spark.serializer.KryoSerializer; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; -import org.assertj.core.api.Assertions; -import org.junit.Assert; -import org.junit.Before; -import org.junit.Rule; -import org.junit.Test; -import org.junit.rules.TemporaryFolder; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; -public class TestScanTaskSerialization extends SparkTestBase { +public class TestScanTaskSerialization extends TestBase { private static final HadoopTables TABLES = new HadoopTables(new Configuration()); private static final Schema SCHEMA = @@ -62,13 +61,13 @@ public class TestScanTaskSerialization extends SparkTestBase { optional(2, "c2", Types.StringType.get()), optional(3, "c3", Types.StringType.get())); - @Rule public TemporaryFolder temp = new TemporaryFolder(); + @TempDir private Path temp; private String tableLocation = null; - @Before + @BeforeEach public void setupTableLocation() throws Exception { - File tableDir = temp.newFolder(); + File tableDir = Files.createTempDirectory(temp, "junit").toFile(); this.tableLocation = tableDir.toURI().toString(); } @@ -76,8 +75,8 @@ public void setupTableLocation() throws Exception { public void testBaseCombinedScanTaskKryoSerialization() throws Exception { BaseCombinedScanTask scanTask = prepareBaseCombinedScanTaskForSerDeTest(); - File data = temp.newFile(); - Assert.assertTrue(data.delete()); + File data = File.createTempFile("junit", null, temp.toFile()); + assertThat(data.delete()).isTrue(); Kryo kryo = new KryoSerializer(new SparkConf()).newKryo(); try (Output out = new Output(new FileOutputStream(data))) { @@ -86,7 +85,7 @@ public void testBaseCombinedScanTaskKryoSerialization() throws Exception { try (Input in = new Input(new FileInputStream(data))) { Object obj = kryo.readClassAndObject(in); - Assertions.assertThat(obj) + assertThat(obj) .as("Should be a BaseCombinedScanTask") .isInstanceOf(BaseCombinedScanTask.class); TaskCheckHelper.assertEquals(scanTask, (BaseCombinedScanTask) obj); @@ -105,7 +104,7 @@ public void testBaseCombinedScanTaskJavaSerialization() throws Exception { try (ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) { Object obj = in.readObject(); - Assertions.assertThat(obj) + assertThat(obj) .as("Should be a BaseCombinedScanTask") .isInstanceOf(BaseCombinedScanTask.class); TaskCheckHelper.assertEquals(scanTask, (BaseCombinedScanTask) obj); @@ -117,10 +116,10 @@ public void testBaseCombinedScanTaskJavaSerialization() throws Exception { public void testBaseScanTaskGroupKryoSerialization() throws Exception { BaseScanTaskGroup taskGroup = prepareBaseScanTaskGroupForSerDeTest(); - Assert.assertTrue("Task group can't be empty", !taskGroup.tasks().isEmpty()); + assertThat(taskGroup.tasks()).as("Task group can't be empty").isNotEmpty(); - File data = temp.newFile(); - Assert.assertTrue(data.delete()); + File data = File.createTempFile("junit", null, temp.toFile()); + assertThat(data.delete()).isTrue(); Kryo kryo = new KryoSerializer(new SparkConf()).newKryo(); try (Output out = new Output(Files.newOutputStream(data.toPath()))) { @@ -129,9 +128,7 @@ public void testBaseScanTaskGroupKryoSerialization() throws Exception { try (Input in = new Input(Files.newInputStream(data.toPath()))) { Object obj = kryo.readClassAndObject(in); - Assertions.assertThat(obj) - .as("should be a BaseScanTaskGroup") - .isInstanceOf(BaseScanTaskGroup.class); + assertThat(obj).as("should be a BaseScanTaskGroup").isInstanceOf(BaseScanTaskGroup.class); TaskCheckHelper.assertEquals(taskGroup, (BaseScanTaskGroup) obj); } } @@ -141,7 +138,7 @@ public void testBaseScanTaskGroupKryoSerialization() throws Exception { public void testBaseScanTaskGroupJavaSerialization() throws Exception { BaseScanTaskGroup taskGroup = prepareBaseScanTaskGroupForSerDeTest(); - Assert.assertTrue("Task group can't be empty", !taskGroup.tasks().isEmpty()); + assertThat(taskGroup.tasks()).as("Task group can't be empty").isNotEmpty(); ByteArrayOutputStream bytes = new ByteArrayOutputStream(); try (ObjectOutputStream out = new ObjectOutputStream(bytes)) { @@ -151,9 +148,7 @@ public void testBaseScanTaskGroupJavaSerialization() throws Exception { try (ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) { Object obj = in.readObject(); - Assertions.assertThat(obj) - .as("should be a BaseScanTaskGroup") - .isInstanceOf(BaseScanTaskGroup.class); + assertThat(obj).as("should be a BaseScanTaskGroup").isInstanceOf(BaseScanTaskGroup.class); TaskCheckHelper.assertEquals(taskGroup, (BaseScanTaskGroup) obj); } } diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestTableSerialization.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestTableSerialization.java index ebab094cbe84..fd6dfd07b568 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestTableSerialization.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/TestTableSerialization.java @@ -20,6 +20,7 @@ import static org.apache.iceberg.types.Types.NestedField.optional; import static org.apache.iceberg.types.Types.NestedField.required; +import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; @@ -28,6 +29,9 @@ import java.io.File; import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Arrays; import java.util.List; import java.util.Map; import org.apache.iceberg.hadoop.HadoopTables; @@ -36,29 +40,22 @@ import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.spark.source.SerializableTableWithSize; import org.apache.iceberg.types.Types; -import org.junit.Assert; -import org.junit.Before; -import org.junit.Rule; -import org.junit.Test; -import org.junit.rules.TemporaryFolder; -import org.junit.runner.RunWith; -import org.junit.runners.Parameterized; - -@RunWith(Parameterized.class) -public class TestTableSerialization { +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.io.TempDir; - public TestTableSerialization(String isObjectStoreEnabled) { - this.isObjectStoreEnabled = isObjectStoreEnabled; - } +@ExtendWith(ParameterizedTestExtension.class) +public class TestTableSerialization { - @Parameterized.Parameters(name = "isObjectStoreEnabled = {0}") - public static Object[] parameters() { - return new Object[] {"true", "false"}; + @Parameters(name = "isObjectStoreEnabled = {0}") + public static List parameters() { + return Arrays.asList("true", "false"); } private static final HadoopTables TABLES = new HadoopTables(); - private final String isObjectStoreEnabled; + @Parameter private String isObjectStoreEnabled; private static final Schema SCHEMA = new Schema( @@ -72,21 +69,21 @@ public static Object[] parameters() { private static final SortOrder SORT_ORDER = SortOrder.builderFor(SCHEMA).asc("id").build(); - @Rule public TemporaryFolder temp = new TemporaryFolder(); + @TempDir private Path temp; private Table table; - @Before + @BeforeEach public void initTable() throws IOException { Map props = ImmutableMap.of("k1", "v1", TableProperties.OBJECT_STORE_ENABLED, isObjectStoreEnabled); - File tableLocation = temp.newFolder(); - Assert.assertTrue(tableLocation.delete()); + File tableLocation = Files.createTempDirectory(temp, "junit").toFile(); + assertThat(tableLocation.delete()).isTrue(); this.table = TABLES.create(SCHEMA, SPEC, SORT_ORDER, props, tableLocation.toString()); } - @Test + @TestTemplate public void testCloseSerializableTableKryoSerialization() throws Exception { for (Table tbl : tables()) { Table spyTable = spy(tbl); @@ -107,7 +104,7 @@ public void testCloseSerializableTableKryoSerialization() throws Exception { } } - @Test + @TestTemplate public void testCloseSerializableTableJavaSerialization() throws Exception { for (Table tbl : tables()) { Table spyTable = spy(tbl); @@ -128,14 +125,14 @@ public void testCloseSerializableTableJavaSerialization() throws Exception { } } - @Test + @TestTemplate public void testSerializableTableKryoSerialization() throws IOException { Table serializableTable = SerializableTableWithSize.copyOf(table); TestHelpers.assertSerializedAndLoadedMetadata( table, KryoHelpers.roundTripSerialize(serializableTable)); } - @Test + @TestTemplate public void testSerializableMetadataTableKryoSerialization() throws IOException { for (MetadataTableType type : MetadataTableType.values()) { TableOperations ops = ((HasTableOperations) table).operations(); @@ -148,7 +145,7 @@ public void testSerializableMetadataTableKryoSerialization() throws IOException } } - @Test + @TestTemplate public void testSerializableTransactionTableKryoSerialization() throws IOException { Transaction txn = table.newTransaction(); diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/ValidationHelpers.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/ValidationHelpers.java index 70ab04f0a080..b0b3085dca70 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/ValidationHelpers.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/ValidationHelpers.java @@ -18,11 +18,12 @@ */ package org.apache.iceberg; +import static org.assertj.core.api.Assertions.assertThat; + import java.util.Arrays; import java.util.List; import java.util.stream.Collectors; import org.apache.iceberg.relocated.com.google.common.collect.Lists; -import org.assertj.core.api.Assertions; public class ValidationHelpers { @@ -72,6 +73,6 @@ public static void validateDataManifest( private static void assertSameElements(String context, List actual, List expected) { String errorMessage = String.format("%s must match", context); - Assertions.assertThat(actual).as(errorMessage).hasSameElementsAs(expected); + assertThat(actual).as(errorMessage).hasSameElementsAs(expected); } } diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestBase.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestBase.java index a456fdcf4450..e7d5a0f0398a 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestBase.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestBase.java @@ -19,6 +19,7 @@ package org.apache.iceberg.spark; import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTOREURIS; +import static org.assertj.core.api.Assertions.assertThat; import java.io.IOException; import java.io.UncheckedIOException; @@ -53,7 +54,6 @@ import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec; import org.apache.spark.sql.internal.SQLConf; import org.apache.spark.sql.util.QueryExecutionListener; -import org.assertj.core.api.Assertions; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; @@ -127,9 +127,9 @@ protected List sql(String query, Object... args) { protected Object scalarSql(String query, Object... args) { List rows = sql(query, args); - Assertions.assertThat(rows.size()).as("Scalar SQL should return one row").isEqualTo(1); + assertThat(rows.size()).as("Scalar SQL should return one row").isEqualTo(1); Object[] row = Iterables.getOnlyElement(rows); - Assertions.assertThat(row.length).as("Scalar SQL should return one value").isEqualTo(1); + assertThat(row.length).as("Scalar SQL should return one value").isEqualTo(1); return row[0]; } diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestBaseWithCatalog.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestBaseWithCatalog.java index dd00405dc87c..be0d8c6439a2 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestBaseWithCatalog.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestBaseWithCatalog.java @@ -18,6 +18,8 @@ */ package org.apache.iceberg.spark; +import static org.assertj.core.api.Assertions.assertThat; + import java.io.File; import java.io.IOException; import java.util.Map; @@ -35,7 +37,6 @@ import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.hadoop.HadoopCatalog; import org.apache.iceberg.util.PropertyUtil; -import org.assertj.core.api.Assertions; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; @@ -60,7 +61,7 @@ protected static Object[][] parameters() { @BeforeAll public static void createWarehouse() throws IOException { TestBaseWithCatalog.warehouse = File.createTempFile("warehouse", null); - Assertions.assertThat(warehouse.delete()).isTrue(); + assertThat(warehouse.delete()).isTrue(); } @AfterAll @@ -68,9 +69,7 @@ public static void dropWarehouse() throws IOException { if (warehouse != null && warehouse.exists()) { Path warehousePath = new Path(warehouse.getAbsolutePath()); FileSystem fs = warehousePath.getFileSystem(hiveConf); - Assertions.assertThat(fs.delete(warehousePath, true)) - .as("Failed to delete " + warehousePath) - .isTrue(); + assertThat(fs.delete(warehousePath, true)).as("Failed to delete " + warehousePath).isTrue(); } } diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestChangelogIterator.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestChangelogIterator.java index 0539598f147e..bd9832f7d674 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestChangelogIterator.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestChangelogIterator.java @@ -18,7 +18,8 @@ */ package org.apache.iceberg.spark; -import static org.junit.Assert.assertThrows; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.util.Arrays; import java.util.Collections; @@ -33,8 +34,7 @@ import org.apache.spark.sql.types.Metadata; import org.apache.spark.sql.types.StructField; import org.apache.spark.sql.types.StructType; -import org.junit.Assert; -import org.junit.Test; +import org.junit.jupiter.api.Test; public class TestChangelogIterator extends SparkTestHelperBase { private static final String DELETE = ChangelogOperation.DELETE.name(); @@ -78,7 +78,7 @@ public void testIterator() { Arrays.asList(RowType.DELETED, RowType.INSERTED, RowType.CARRY_OVER, RowType.UPDATED), 0, permutations); - Assert.assertEquals(24, permutations.size()); + assertThat(permutations).hasSize(24); for (Object[] permutation : permutations) { validate(permutation); @@ -196,10 +196,10 @@ public void testUpdatedRowsWithDuplication() { Iterator iterator = ChangelogIterator.computeUpdates(rowsWithDuplication.iterator(), SCHEMA, IDENTIFIER_FIELDS); - assertThrows( - "Cannot compute updates because there are multiple rows with the same identifier fields([id, name]). Please make sure the rows are unique.", - IllegalStateException.class, - () -> Lists.newArrayList(iterator)); + assertThatThrownBy(() -> Lists.newArrayList(iterator)) + .isInstanceOf(IllegalStateException.class) + .hasMessage( + "Cannot compute updates because there are multiple rows with the same identifier fields([id,name]). Please make sure the rows are unique."); // still allow extra insert rows rowsWithDuplication = 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 20a2d5c93a01..3955d0395474 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 @@ -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; @@ -30,30 +32,23 @@ 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; - public TestFileRewriteCoordinator( - String catalogName, String implementation, Map config) { - super(catalogName, implementation, config); - } +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 +59,7 @@ public void testBinPackRewrite() throws NoSuchTableException, IOException { 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 fileDF = spark.read().format("iceberg").load(tableName(tableIdent.name() + ".files")); @@ -106,14 +101,16 @@ public void testBinPackRewrite() throws NoSuchTableException, IOException { table.refresh(); Map 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.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"); 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 +121,7 @@ public void testSortRewrite() throws NoSuchTableException, IOException { 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 fileScanTasks = table.newScan().planFiles()) { String fileSetID = UUID.randomUUID().toString(); @@ -176,14 +173,16 @@ public void testSortRewrite() throws NoSuchTableException, IOException { table.refresh(); Map 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.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"); 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 +252,16 @@ public void testCommitMultipleRewrites() throws NoSuchTableException, IOExceptio table.refresh(); - Assert.assertEquals("Should produce 5 snapshots", 5, Iterables.size(table.snapshots())); + assertThat(table.snapshots()).as("Should produce 5 snapshots").hasSize(5); Map 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.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"); 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 newDF(int numRecords) { diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestFunctionCatalog.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestFunctionCatalog.java index be91c8d6370b..5789b43e7d52 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestFunctionCatalog.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestFunctionCatalog.java @@ -18,6 +18,9 @@ */ package org.apache.iceberg.spark; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + import org.apache.iceberg.IcebergBuild; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.spark.functions.IcebergVersionFunction; @@ -27,64 +30,59 @@ import org.apache.spark.sql.connector.catalog.FunctionCatalog; import org.apache.spark.sql.connector.catalog.Identifier; import org.apache.spark.sql.connector.catalog.functions.UnboundFunction; -import org.assertj.core.api.Assertions; -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; -public class TestFunctionCatalog extends SparkTestBaseWithCatalog { +public class TestFunctionCatalog extends TestBaseWithCatalog { private static final String[] EMPTY_NAMESPACE = new String[] {}; private static final String[] SYSTEM_NAMESPACE = new String[] {"system"}; private static final String[] DEFAULT_NAMESPACE = new String[] {"default"}; private static final String[] DB_NAMESPACE = new String[] {"db"}; - private final FunctionCatalog asFunctionCatalog; + private FunctionCatalog asFunctionCatalog; - public TestFunctionCatalog() { + @BeforeEach + public void before() { + super.before(); this.asFunctionCatalog = castToFunctionCatalog(catalogName); - } - - @Before - public void createDefaultNamespace() { sql("CREATE NAMESPACE IF NOT EXISTS %s", catalogName + ".default"); } - @After + @AfterEach public void dropDefaultNamespace() { sql("DROP NAMESPACE IF EXISTS %s", catalogName + ".default"); } - @Test + @TestTemplate public void testListFunctionsViaCatalog() throws NoSuchNamespaceException { - Assertions.assertThat(asFunctionCatalog.listFunctions(EMPTY_NAMESPACE)) + assertThat(asFunctionCatalog.listFunctions(EMPTY_NAMESPACE)) .anyMatch(func -> "iceberg_version".equals(func.name())); - Assertions.assertThat(asFunctionCatalog.listFunctions(SYSTEM_NAMESPACE)) + assertThat(asFunctionCatalog.listFunctions(SYSTEM_NAMESPACE)) .anyMatch(func -> "iceberg_version".equals(func.name())); - Assert.assertArrayEquals( - "Listing functions in an existing namespace that's not system should not throw", - new Identifier[0], - asFunctionCatalog.listFunctions(DEFAULT_NAMESPACE)); + assertThat(asFunctionCatalog.listFunctions(DEFAULT_NAMESPACE)) + .as("Listing functions in an existing namespace that's not system should not throw") + .isEqualTo(new Identifier[0]); - Assertions.assertThatThrownBy(() -> asFunctionCatalog.listFunctions(DB_NAMESPACE)) + assertThatThrownBy(() -> asFunctionCatalog.listFunctions(DB_NAMESPACE)) .isInstanceOf(NoSuchNamespaceException.class) .hasMessageStartingWith("[SCHEMA_NOT_FOUND] The schema `db` cannot be found."); } - @Test + @TestTemplate public void testLoadFunctions() throws NoSuchFunctionException { for (String[] namespace : ImmutableList.of(EMPTY_NAMESPACE, SYSTEM_NAMESPACE)) { Identifier identifier = Identifier.of(namespace, "iceberg_version"); UnboundFunction func = asFunctionCatalog.loadFunction(identifier); - Assertions.assertThat(func) + assertThat(func) .isNotNull() .isInstanceOf(UnboundFunction.class) .isExactlyInstanceOf(IcebergVersionFunction.class); } - Assertions.assertThatThrownBy( + assertThatThrownBy( () -> asFunctionCatalog.loadFunction(Identifier.of(DEFAULT_NAMESPACE, "iceberg_version"))) .isInstanceOf(NoSuchFunctionException.class) @@ -92,42 +90,42 @@ public void testLoadFunctions() throws NoSuchFunctionException { "[ROUTINE_NOT_FOUND] The function default.iceberg_version cannot be found."); Identifier undefinedFunction = Identifier.of(SYSTEM_NAMESPACE, "undefined_function"); - Assertions.assertThatThrownBy(() -> asFunctionCatalog.loadFunction(undefinedFunction)) + assertThatThrownBy(() -> asFunctionCatalog.loadFunction(undefinedFunction)) .isInstanceOf(NoSuchFunctionException.class) .hasMessageStartingWith( "[ROUTINE_NOT_FOUND] The function system.undefined_function cannot be found."); - Assertions.assertThatThrownBy(() -> sql("SELECT undefined_function(1, 2)")) + assertThatThrownBy(() -> sql("SELECT undefined_function(1, 2)")) .isInstanceOf(AnalysisException.class) .hasMessageStartingWith( "[UNRESOLVED_ROUTINE] Cannot resolve function `undefined_function` on search path"); } - @Test + @TestTemplate public void testCallingFunctionInSQLEndToEnd() { String buildVersion = IcebergBuild.version(); - Assert.assertEquals( - "Should be able to use the Iceberg version function from the fully qualified system namespace", - buildVersion, - scalarSql("SELECT %s.system.iceberg_version()", catalogName)); + assertThat(scalarSql("SELECT %s.system.iceberg_version()", catalogName)) + .as( + "Should be able to use the Iceberg version function from the fully qualified system namespace") + .isEqualTo(buildVersion); - Assert.assertEquals( - "Should be able to use the Iceberg version function when fully qualified without specifying a namespace", - buildVersion, - scalarSql("SELECT %s.iceberg_version()", catalogName)); + assertThat(scalarSql("SELECT %s.iceberg_version()", catalogName)) + .as( + "Should be able to use the Iceberg version function when fully qualified without specifying a namespace") + .isEqualTo(buildVersion); sql("USE %s", catalogName); - Assert.assertEquals( - "Should be able to call iceberg_version from system namespace without fully qualified name when using Iceberg catalog", - buildVersion, - scalarSql("SELECT system.iceberg_version()")); + assertThat(scalarSql("SELECT system.iceberg_version()")) + .as( + "Should be able to call iceberg_version from system namespace without fully qualified name when using Iceberg catalog") + .isEqualTo(buildVersion); - Assert.assertEquals( - "Should be able to call iceberg_version from empty namespace without fully qualified name when using Iceberg catalog", - buildVersion, - scalarSql("SELECT iceberg_version()")); + assertThat(scalarSql("SELECT iceberg_version()")) + .as( + "Should be able to call iceberg_version from empty namespace without fully qualified name when using Iceberg catalog") + .isEqualTo(buildVersion); } private FunctionCatalog castToFunctionCatalog(String name) { diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSpark3Util.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSpark3Util.java index ce11f0c05ffd..6f900ffebb10 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSpark3Util.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSpark3Util.java @@ -36,6 +36,7 @@ import static org.apache.iceberg.expressions.Expressions.year; import static org.apache.iceberg.types.Types.NestedField.optional; import static org.apache.iceberg.types.Types.NestedField.required; +import static org.assertj.core.api.Assertions.assertThat; import org.apache.iceberg.CachingCatalog; import org.apache.iceberg.Schema; @@ -45,11 +46,9 @@ import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.expressions.Expression; import org.apache.iceberg.types.Types; -import org.assertj.core.api.Assertions; -import org.junit.Assert; -import org.junit.Test; +import org.junit.jupiter.api.Test; -public class TestSpark3Util extends SparkTestBase { +public class TestSpark3Util extends TestBase { @Test public void testDescribeSortOrder() { Schema schema = @@ -57,46 +56,44 @@ public void testDescribeSortOrder() { required(1, "data", Types.StringType.get()), required(2, "time", Types.TimestampType.withoutZone())); - Assert.assertEquals( - "Sort order isn't correct.", - "data DESC NULLS FIRST", - Spark3Util.describe(buildSortOrder("Identity", schema, 1))); - Assert.assertEquals( - "Sort order isn't correct.", - "bucket(1, data) DESC NULLS FIRST", - Spark3Util.describe(buildSortOrder("bucket[1]", schema, 1))); - Assert.assertEquals( - "Sort order isn't correct.", - "truncate(data, 3) DESC NULLS FIRST", - Spark3Util.describe(buildSortOrder("truncate[3]", schema, 1))); - Assert.assertEquals( - "Sort order isn't correct.", - "years(time) DESC NULLS FIRST", - Spark3Util.describe(buildSortOrder("year", schema, 2))); - Assert.assertEquals( - "Sort order isn't correct.", - "months(time) DESC NULLS FIRST", - Spark3Util.describe(buildSortOrder("month", schema, 2))); - Assert.assertEquals( - "Sort order isn't correct.", - "days(time) DESC NULLS FIRST", - Spark3Util.describe(buildSortOrder("day", schema, 2))); - Assert.assertEquals( - "Sort order isn't correct.", - "hours(time) DESC NULLS FIRST", - Spark3Util.describe(buildSortOrder("hour", schema, 2))); - Assert.assertEquals( - "Sort order isn't correct.", - "unknown(data) DESC NULLS FIRST", - Spark3Util.describe(buildSortOrder("unknown", schema, 1))); + assertThat(Spark3Util.describe(buildSortOrder("Identity", schema, 1))) + .as("Sort order isn't correct.") + .isEqualTo("data DESC NULLS FIRST"); + + assertThat(Spark3Util.describe(buildSortOrder("bucket[1]", schema, 1))) + .as("Sort order isn't correct.") + .isEqualTo("bucket(1, data) DESC NULLS FIRST"); + + assertThat(Spark3Util.describe(buildSortOrder("truncate[3]", schema, 1))) + .as("Sort order isn't correct.") + .isEqualTo("truncate(data, 3) DESC NULLS FIRST"); + + assertThat(Spark3Util.describe(buildSortOrder("year", schema, 2))) + .as("Sort order isn't correct.") + .isEqualTo("years(time) DESC NULLS FIRST"); + + assertThat(Spark3Util.describe(buildSortOrder("month", schema, 2))) + .as("Sort order isn't correct.") + .isEqualTo("months(time) DESC NULLS FIRST"); + + assertThat(Spark3Util.describe(buildSortOrder("day", schema, 2))) + .as("Sort order isn't correct.") + .isEqualTo("days(time) DESC NULLS FIRST"); + + assertThat(Spark3Util.describe(buildSortOrder("hour", schema, 2))) + .as("Sort order isn't correct.") + .isEqualTo("hours(time) DESC NULLS FIRST"); + + assertThat(Spark3Util.describe(buildSortOrder("unknown", schema, 1))) + .as("Sort order isn't correct.") + .isEqualTo("unknown(data) DESC NULLS FIRST"); // multiple sort orders SortOrder multiOrder = SortOrder.builderFor(schema).asc("time", NULLS_FIRST).asc("data", NULLS_LAST).build(); - Assert.assertEquals( - "Sort order isn't correct.", - "time ASC NULLS FIRST, data ASC NULLS LAST", - Spark3Util.describe(multiOrder)); + assertThat(Spark3Util.describe(multiOrder)) + .as("Sort order isn't correct.") + .isEqualTo("time ASC NULLS FIRST, data ASC NULLS LAST"); } @Test @@ -110,10 +107,10 @@ public void testDescribeSchema() { Types.MapType.ofOptional(4, 5, Types.StringType.get(), Types.LongType.get())), required(6, "time", Types.TimestampType.withoutZone())); - Assert.assertEquals( - "Schema description isn't correct.", - "struct not null,pairs: map,time: timestamp not null>", - Spark3Util.describe(schema)); + assertThat(Spark3Util.describe(schema)) + .as("Schema description isn't correct.") + .isEqualTo( + "struct not null,pairs: map,time: timestamp not null>"); } @Test @@ -126,7 +123,7 @@ public void testLoadIcebergTable() throws Exception { sql("CREATE TABLE %s (c1 bigint, c2 string, c3 string) USING iceberg", tableFullName); Table table = Spark3Util.loadIcebergTable(spark, tableFullName); - Assert.assertTrue(table.name().equals(tableFullName)); + assertThat(table.name()).isEqualTo(tableFullName); } @Test @@ -134,37 +131,37 @@ public void testLoadIcebergCatalog() throws Exception { spark.conf().set("spark.sql.catalog.test_cat", SparkCatalog.class.getName()); spark.conf().set("spark.sql.catalog.test_cat.type", "hive"); Catalog catalog = Spark3Util.loadIcebergCatalog(spark, "test_cat"); - Assert.assertTrue( - "Should retrieve underlying catalog class", catalog instanceof CachingCatalog); + assertThat(catalog) + .as("Should retrieve underlying catalog class") + .isInstanceOf(CachingCatalog.class); } @Test public void testDescribeExpression() { Expression refExpression = equal("id", 1); - Assertions.assertThat(Spark3Util.describe(refExpression)).isEqualTo("id = 1"); + assertThat(Spark3Util.describe(refExpression)).isEqualTo("id = 1"); Expression yearExpression = greaterThan(year("ts"), 10); - Assertions.assertThat(Spark3Util.describe(yearExpression)).isEqualTo("year(ts) > 10"); + assertThat(Spark3Util.describe(yearExpression)).isEqualTo("year(ts) > 10"); Expression monthExpression = greaterThanOrEqual(month("ts"), 10); - Assertions.assertThat(Spark3Util.describe(monthExpression)).isEqualTo("month(ts) >= 10"); + assertThat(Spark3Util.describe(monthExpression)).isEqualTo("month(ts) >= 10"); Expression dayExpression = lessThan(day("ts"), 10); - Assertions.assertThat(Spark3Util.describe(dayExpression)).isEqualTo("day(ts) < 10"); + assertThat(Spark3Util.describe(dayExpression)).isEqualTo("day(ts) < 10"); Expression hourExpression = lessThanOrEqual(hour("ts"), 10); - Assertions.assertThat(Spark3Util.describe(hourExpression)).isEqualTo("hour(ts) <= 10"); + assertThat(Spark3Util.describe(hourExpression)).isEqualTo("hour(ts) <= 10"); Expression bucketExpression = in(bucket("id", 5), 3); - Assertions.assertThat(Spark3Util.describe(bucketExpression)).isEqualTo("bucket[5](id) IN (3)"); + assertThat(Spark3Util.describe(bucketExpression)).isEqualTo("bucket[5](id) IN (3)"); Expression truncateExpression = notIn(truncate("name", 3), "abc"); - Assertions.assertThat(Spark3Util.describe(truncateExpression)) + assertThat(Spark3Util.describe(truncateExpression)) .isEqualTo("truncate[3](name) NOT IN ('abc')"); Expression andExpression = and(refExpression, yearExpression); - Assertions.assertThat(Spark3Util.describe(andExpression)) - .isEqualTo("(id = 1 AND year(ts) > 10)"); + assertThat(Spark3Util.describe(andExpression)).isEqualTo("(id = 1 AND year(ts) > 10)"); } private SortOrder buildSortOrder(String transform, Schema schema, int sourceId) { 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 23e8717fb8c3..eaf230865957 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,32 +18,40 @@ */ package org.apache.iceberg.spark; +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; -public class TestSparkCachedTableCatalog extends SparkTestBaseWithCatalog { +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.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 0836271a7c22..0f29faf274dd 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 @@ -18,7 +18,8 @@ */ package org.apache.iceberg.spark; -import java.util.Map; +import static org.assertj.core.api.Assertions.assertThat; + import org.apache.iceberg.Schema; import org.apache.iceberg.catalog.Catalog; import org.apache.spark.sql.catalyst.analysis.NoSuchTableException; @@ -27,28 +28,23 @@ 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; -public class TestSparkCatalogOperations extends SparkCatalogTestBase { - public TestSparkCatalogOperations( - String catalogName, String implementation, Map config) { - super(catalogName, implementation, config); - } +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 +58,20 @@ public void testAlterTable() throws NoSuchTableException { 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); + assertThat(table.schema().fields()[2]) + .as("Adding a column to a table should return the updated table with the new column") + .isEqualTo(expectedField); - 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.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkCompressionUtil.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkCompressionUtil.java index a2b48edffae4..aa329efbbad5 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkCompressionUtil.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkCompressionUtil.java @@ -34,15 +34,15 @@ import org.apache.spark.SparkContext; import org.apache.spark.internal.config.package$; import org.apache.spark.sql.SparkSession; -import org.junit.Before; -import org.junit.Test; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; public class TestSparkCompressionUtil { private SparkSession spark; private SparkConf sparkConf; - @Before + @BeforeEach public void initSpark() { this.spark = mock(SparkSession.class); this.sparkConf = mock(SparkConf.class); diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java index 7ed34d4016ba..39ef72c6bb1d 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java @@ -29,6 +29,7 @@ import static org.apache.spark.sql.connector.write.RowLevelOperation.Command.DELETE; import static org.apache.spark.sql.connector.write.RowLevelOperation.Command.MERGE; import static org.apache.spark.sql.connector.write.RowLevelOperation.Command.UPDATE; +import static org.assertj.core.api.Assertions.assertThat; import org.apache.iceberg.MetadataColumns; import org.apache.iceberg.Table; @@ -40,11 +41,10 @@ import org.apache.spark.sql.connector.expressions.SortDirection; import org.apache.spark.sql.connector.expressions.SortOrder; import org.apache.spark.sql.connector.write.RowLevelOperation.Command; -import org.junit.After; -import org.junit.Assert; -import org.junit.Test; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.TestTemplate; -public class TestSparkDistributionAndOrderingUtil extends SparkTestBaseWithCatalog { +public class TestSparkDistributionAndOrderingUtil extends TestBaseWithCatalog { private static final Distribution UNSPECIFIED_DISTRIBUTION = Distributions.unspecified(); private static final Distribution FILE_CLUSTERED_DISTRIBUTION = @@ -100,7 +100,7 @@ public class TestSparkDistributionAndOrderingUtil extends SparkTestBaseWithCatal Expressions.column(MetadataColumns.ROW_POSITION.name()), SortDirection.ASCENDING) }; - @After + @AfterEach public void dropTable() { sql("DROP TABLE IF EXISTS %s", tableName); } @@ -141,7 +141,7 @@ public void dropTable() { // write mode is HASH -> CLUSTER BY date + LOCALLY ORDER BY date, id // write mode is RANGE -> ORDER BY date, id - @Test + @TestTemplate public void testDefaultWriteUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -150,7 +150,7 @@ public void testDefaultWriteUnpartitionedUnsortedTable() { checkWriteDistributionAndOrdering(table, UNSPECIFIED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testHashWriteUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -161,7 +161,7 @@ public void testHashWriteUnpartitionedUnsortedTable() { checkWriteDistributionAndOrdering(table, UNSPECIFIED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testRangeWriteUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -172,7 +172,7 @@ public void testRangeWriteUnpartitionedUnsortedTable() { checkWriteDistributionAndOrdering(table, UNSPECIFIED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testDefaultWriteUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -191,7 +191,7 @@ public void testDefaultWriteUnpartitionedSortedTable() { checkWriteDistributionAndOrdering(table, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testHashWriteUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -210,7 +210,7 @@ public void testHashWriteUnpartitionedSortedTable() { checkWriteDistributionAndOrdering(table, UNSPECIFIED_DISTRIBUTION, expectedOrdering); } - @Test + @TestTemplate public void testRangeWriteUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -231,7 +231,7 @@ public void testRangeWriteUnpartitionedSortedTable() { checkWriteDistributionAndOrdering(table, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testDefaultWritePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -260,7 +260,7 @@ public void testDefaultWritePartitionedUnsortedTable() { checkWriteDistributionAndOrdering(table, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testHashWritePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -291,7 +291,7 @@ public void testHashWritePartitionedUnsortedTable() { checkWriteDistributionAndOrdering(table, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testRangeWritePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -320,7 +320,7 @@ public void testRangeWritePartitionedUnsortedTable() { checkWriteDistributionAndOrdering(table, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testDefaultWritePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -343,7 +343,7 @@ public void testDefaultWritePartitionedSortedTable() { checkWriteDistributionAndOrdering(table, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testHashWritePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -371,7 +371,7 @@ public void testHashWritePartitionedSortedTable() { checkWriteDistributionAndOrdering(table, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testRangeWritePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -436,7 +436,7 @@ public void testRangeWritePartitionedSortedTable() { // delete mode is HASH -> CLUSTER BY date + LOCALLY ORDER BY date, id // delete mode is RANGE -> ORDER BY date, id - @Test + @TestTemplate public void testDefaultCopyOnWriteDeleteUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -446,7 +446,7 @@ public void testDefaultCopyOnWriteDeleteUnpartitionedUnsortedTable() { table, DELETE, FILE_CLUSTERED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testNoneCopyOnWriteDeleteUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -458,7 +458,7 @@ public void testNoneCopyOnWriteDeleteUnpartitionedUnsortedTable() { table, DELETE, UNSPECIFIED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testHashCopyOnWriteDeleteUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -470,7 +470,7 @@ public void testHashCopyOnWriteDeleteUnpartitionedUnsortedTable() { table, DELETE, FILE_CLUSTERED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testRangeCopyOnWriteDeleteUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -483,7 +483,7 @@ public void testRangeCopyOnWriteDeleteUnpartitionedUnsortedTable() { checkCopyOnWriteDistributionAndOrdering(table, DELETE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testDefaultCopyOnWriteDeleteUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -501,7 +501,7 @@ public void testDefaultCopyOnWriteDeleteUnpartitionedSortedTable() { table, DELETE, FILE_CLUSTERED_DISTRIBUTION, expectedOrdering); } - @Test + @TestTemplate public void testNoneCopyOnWriteDeleteUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -521,7 +521,7 @@ public void testNoneCopyOnWriteDeleteUnpartitionedSortedTable() { table, DELETE, UNSPECIFIED_DISTRIBUTION, expectedOrdering); } - @Test + @TestTemplate public void testHashCopyOnWriteDeleteUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -541,7 +541,7 @@ public void testHashCopyOnWriteDeleteUnpartitionedSortedTable() { table, DELETE, FILE_CLUSTERED_DISTRIBUTION, expectedOrdering); } - @Test + @TestTemplate public void testRangeCopyOnWriteDeleteUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -562,7 +562,7 @@ public void testRangeCopyOnWriteDeleteUnpartitionedSortedTable() { checkCopyOnWriteDistributionAndOrdering(table, DELETE, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testDefaultCopyOnWriteDeletePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -591,7 +591,7 @@ public void testDefaultCopyOnWriteDeletePartitionedUnsortedTable() { checkCopyOnWriteDistributionAndOrdering(table, DELETE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testNoneCopyOnWriteDeletePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -620,7 +620,7 @@ public void testNoneCopyOnWriteDeletePartitionedUnsortedTable() { table, DELETE, UNSPECIFIED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testHashCopyOnWriteDeletePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -651,7 +651,7 @@ public void testHashCopyOnWriteDeletePartitionedUnsortedTable() { checkCopyOnWriteDistributionAndOrdering(table, DELETE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testRangeCopyOnWriteDeletePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -680,7 +680,7 @@ public void testRangeCopyOnWriteDeletePartitionedUnsortedTable() { checkCopyOnWriteDistributionAndOrdering(table, DELETE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testDefaultCopyOnWriteDeletePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -704,7 +704,7 @@ public void testDefaultCopyOnWriteDeletePartitionedSortedTable() { checkCopyOnWriteDistributionAndOrdering(table, DELETE, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testNoneCopyOnWriteDeletePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -728,7 +728,7 @@ public void testNoneCopyOnWriteDeletePartitionedSortedTable() { table, DELETE, UNSPECIFIED_DISTRIBUTION, expectedOrdering); } - @Test + @TestTemplate public void testHashCopyOnWriteDeletePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -756,7 +756,7 @@ public void testHashCopyOnWriteDeletePartitionedSortedTable() { checkCopyOnWriteDistributionAndOrdering(table, DELETE, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testRangeCopyOnWriteDeletePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -817,7 +817,7 @@ public void testRangeCopyOnWriteDeletePartitionedSortedTable() { // update mode is HASH -> CLUSTER BY date + LOCALLY ORDER BY date, id // update mode is RANGE -> ORDER BY date, id - @Test + @TestTemplate public void testDefaultCopyOnWriteUpdateUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -827,7 +827,7 @@ public void testDefaultCopyOnWriteUpdateUnpartitionedUnsortedTable() { table, UPDATE, FILE_CLUSTERED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testNoneCopyOnWriteUpdateUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -839,7 +839,7 @@ public void testNoneCopyOnWriteUpdateUnpartitionedUnsortedTable() { table, UPDATE, UNSPECIFIED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testHashCopyOnWriteUpdateUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -851,7 +851,7 @@ public void testHashCopyOnWriteUpdateUnpartitionedUnsortedTable() { table, UPDATE, FILE_CLUSTERED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testRangeCopyOnWriteUpdateUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -864,7 +864,7 @@ public void testRangeCopyOnWriteUpdateUnpartitionedUnsortedTable() { checkCopyOnWriteDistributionAndOrdering(table, UPDATE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testDefaultCopyOnWriteUpdateUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -882,7 +882,7 @@ public void testDefaultCopyOnWriteUpdateUnpartitionedSortedTable() { table, UPDATE, FILE_CLUSTERED_DISTRIBUTION, expectedOrdering); } - @Test + @TestTemplate public void testNoneCopyOnWriteUpdateUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -902,7 +902,7 @@ public void testNoneCopyOnWriteUpdateUnpartitionedSortedTable() { table, UPDATE, UNSPECIFIED_DISTRIBUTION, expectedOrdering); } - @Test + @TestTemplate public void testHashCopyOnWriteUpdateUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -922,7 +922,7 @@ public void testHashCopyOnWriteUpdateUnpartitionedSortedTable() { table, UPDATE, FILE_CLUSTERED_DISTRIBUTION, expectedOrdering); } - @Test + @TestTemplate public void testRangeCopyOnWriteUpdateUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -943,7 +943,7 @@ public void testRangeCopyOnWriteUpdateUnpartitionedSortedTable() { checkCopyOnWriteDistributionAndOrdering(table, UPDATE, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testDefaultCopyOnWriteUpdatePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -972,7 +972,7 @@ public void testDefaultCopyOnWriteUpdatePartitionedUnsortedTable() { checkCopyOnWriteDistributionAndOrdering(table, UPDATE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testNoneCopyOnWriteUpdatePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1001,7 +1001,7 @@ public void testNoneCopyOnWriteUpdatePartitionedUnsortedTable() { table, UPDATE, UNSPECIFIED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testHashCopyOnWriteUpdatePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1032,7 +1032,7 @@ public void testHashCopyOnWriteUpdatePartitionedUnsortedTable() { checkCopyOnWriteDistributionAndOrdering(table, UPDATE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testRangeCopyOnWriteUpdatePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1061,7 +1061,7 @@ public void testRangeCopyOnWriteUpdatePartitionedUnsortedTable() { checkCopyOnWriteDistributionAndOrdering(table, UPDATE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testDefaultCopyOnWriteUpdatePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1085,7 +1085,7 @@ public void testDefaultCopyOnWriteUpdatePartitionedSortedTable() { checkCopyOnWriteDistributionAndOrdering(table, UPDATE, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testNoneCopyOnWriteUpdatePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1109,7 +1109,7 @@ public void testNoneCopyOnWriteUpdatePartitionedSortedTable() { table, UPDATE, UNSPECIFIED_DISTRIBUTION, expectedOrdering); } - @Test + @TestTemplate public void testHashCopyOnWriteUpdatePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1137,7 +1137,7 @@ public void testHashCopyOnWriteUpdatePartitionedSortedTable() { checkCopyOnWriteDistributionAndOrdering(table, UPDATE, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testRangeCopyOnWriteUpdatePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1198,7 +1198,7 @@ public void testRangeCopyOnWriteUpdatePartitionedSortedTable() { // merge mode is HASH -> CLUSTER BY date + LOCALLY ORDER BY date, id // merge mode is RANGE -> ORDERED BY date, id - @Test + @TestTemplate public void testDefaultCopyOnWriteMergeUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1207,7 +1207,7 @@ public void testDefaultCopyOnWriteMergeUnpartitionedUnsortedTable() { checkCopyOnWriteDistributionAndOrdering(table, MERGE, UNSPECIFIED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testNoneCopyOnWriteMergeUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1218,7 +1218,7 @@ public void testNoneCopyOnWriteMergeUnpartitionedUnsortedTable() { checkCopyOnWriteDistributionAndOrdering(table, MERGE, UNSPECIFIED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testHashCopyOnWriteMergeUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1229,7 +1229,7 @@ public void testHashCopyOnWriteMergeUnpartitionedUnsortedTable() { checkCopyOnWriteDistributionAndOrdering(table, MERGE, UNSPECIFIED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testRangeCopyOnWriteMergeUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1240,7 +1240,7 @@ public void testRangeCopyOnWriteMergeUnpartitionedUnsortedTable() { checkCopyOnWriteDistributionAndOrdering(table, MERGE, UNSPECIFIED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testDefaultCopyOnWriteMergeUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1259,7 +1259,7 @@ public void testDefaultCopyOnWriteMergeUnpartitionedSortedTable() { checkCopyOnWriteDistributionAndOrdering(table, MERGE, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testNoneCopyOnWriteMergeUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1279,7 +1279,7 @@ public void testNoneCopyOnWriteMergeUnpartitionedSortedTable() { table, MERGE, UNSPECIFIED_DISTRIBUTION, expectedOrdering); } - @Test + @TestTemplate public void testHashCopyOnWriteMergeUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1299,7 +1299,7 @@ public void testHashCopyOnWriteMergeUnpartitionedSortedTable() { table, MERGE, UNSPECIFIED_DISTRIBUTION, expectedOrdering); } - @Test + @TestTemplate public void testRangeCopyOnWriteMergeUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1320,7 +1320,7 @@ public void testRangeCopyOnWriteMergeUnpartitionedSortedTable() { checkCopyOnWriteDistributionAndOrdering(table, MERGE, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testDefaultCopyOnWriteMergePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1349,7 +1349,7 @@ public void testDefaultCopyOnWriteMergePartitionedUnsortedTable() { checkCopyOnWriteDistributionAndOrdering(table, MERGE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testNoneCopyOnWriteMergePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1377,7 +1377,7 @@ public void testNoneCopyOnWriteMergePartitionedUnsortedTable() { checkCopyOnWriteDistributionAndOrdering(table, MERGE, UNSPECIFIED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testHashCopyOnWriteMergePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1408,7 +1408,7 @@ public void testHashCopyOnWriteMergePartitionedUnsortedTable() { checkCopyOnWriteDistributionAndOrdering(table, MERGE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testRangeCopyOnWriteMergePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1437,7 +1437,7 @@ public void testRangeCopyOnWriteMergePartitionedUnsortedTable() { checkCopyOnWriteDistributionAndOrdering(table, MERGE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testDefaultCopyOnWriteMergePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1461,7 +1461,7 @@ public void testDefaultCopyOnWriteMergePartitionedSortedTable() { checkCopyOnWriteDistributionAndOrdering(table, MERGE, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testNoneCopyOnWriteMergePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1485,7 +1485,7 @@ public void testNoneCopyOnWriteMergePartitionedSortedTable() { table, MERGE, UNSPECIFIED_DISTRIBUTION, expectedOrdering); } - @Test + @TestTemplate public void testHashCopyOnWriteMergePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1513,7 +1513,7 @@ public void testHashCopyOnWriteMergePartitionedSortedTable() { checkCopyOnWriteDistributionAndOrdering(table, MERGE, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testRangeCopyOnWriteMergePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1573,7 +1573,7 @@ public void testRangeCopyOnWriteMergePartitionedSortedTable() { // LOCALLY ORDERED BY _spec_id, _partition, _file, _pos // delete mode is RANGE (fanout) -> RANGE DISTRIBUTE BY _spec_id, _partition + empty ordering - @Test + @TestTemplate public void testDefaultPositionDeltaDeleteUnpartitionedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1593,7 +1593,7 @@ public void testDefaultPositionDeltaDeleteUnpartitionedTable() { table, DELETE, SPEC_ID_PARTITION_FILE_CLUSTERED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testNonePositionDeltaDeleteUnpartitionedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1612,7 +1612,7 @@ public void testNonePositionDeltaDeleteUnpartitionedTable() { table, DELETE, UNSPECIFIED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testHashPositionDeltaDeleteUnpartitionedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1634,7 +1634,7 @@ public void testHashPositionDeltaDeleteUnpartitionedTable() { table, DELETE, SPEC_ID_PARTITION_FILE_CLUSTERED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testRangePositionDeltaDeleteUnpartitionedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1654,7 +1654,7 @@ public void testRangePositionDeltaDeleteUnpartitionedTable() { checkPositionDeltaDistributionAndOrdering(table, DELETE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testDefaultPositionDeltaDeletePartitionedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1678,7 +1678,7 @@ public void testDefaultPositionDeltaDeletePartitionedTable() { table, DELETE, SPEC_ID_PARTITION_CLUSTERED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testNonePositionDeltaDeletePartitionedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1701,7 +1701,7 @@ public void testNonePositionDeltaDeletePartitionedTable() { table, DELETE, UNSPECIFIED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testHashPositionDeltaDeletePartitionedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1727,7 +1727,7 @@ public void testHashPositionDeltaDeletePartitionedTable() { table, DELETE, SPEC_ID_PARTITION_CLUSTERED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testRangePositionDeltaDeletePartitionedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -1814,7 +1814,7 @@ public void testRangePositionDeltaDeletePartitionedTable() { // update mode is RANGE -> RANGE DISTRIBUTE BY _spec_id, _partition, date, id + // LOCALLY ORDERED BY _spec_id, _partition, _file, _pos, date, id - @Test + @TestTemplate public void testDefaultPositionDeltaUpdateUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1834,7 +1834,7 @@ public void testDefaultPositionDeltaUpdateUnpartitionedUnsortedTable() { table, UPDATE, SPEC_ID_PARTITION_FILE_CLUSTERED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testNonePositionDeltaUpdateUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1853,7 +1853,7 @@ public void testNonePositionDeltaUpdateUnpartitionedUnsortedTable() { table, UPDATE, UNSPECIFIED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testHashPositionDeltaUpdateUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1875,7 +1875,7 @@ public void testHashPositionDeltaUpdateUnpartitionedUnsortedTable() { table, UPDATE, SPEC_ID_PARTITION_FILE_CLUSTERED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testRangePositionDeltaUpdateUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1895,7 +1895,7 @@ public void testRangePositionDeltaUpdateUnpartitionedUnsortedTable() { checkPositionDeltaDistributionAndOrdering(table, UPDATE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testDefaultPositionDeltaUpdateUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1921,7 +1921,7 @@ public void testDefaultPositionDeltaUpdateUnpartitionedSortedTable() { table, UPDATE, SPEC_ID_PARTITION_FILE_CLUSTERED_DISTRIBUTION, expectedOrdering); } - @Test + @TestTemplate public void testNonePositionDeltaUpdateUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1949,7 +1949,7 @@ public void testNonePositionDeltaUpdateUnpartitionedSortedTable() { table, UPDATE, UNSPECIFIED_DISTRIBUTION, expectedOrdering); } - @Test + @TestTemplate public void testHashPositionDeltaUpdateUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -1977,7 +1977,7 @@ public void testHashPositionDeltaUpdateUnpartitionedSortedTable() { table, UPDATE, SPEC_ID_PARTITION_FILE_CLUSTERED_DISTRIBUTION, expectedOrdering); } - @Test + @TestTemplate public void testRangePositionDeltaUpdateUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -2018,7 +2018,7 @@ public void testRangePositionDeltaUpdateUnpartitionedSortedTable() { table, UPDATE, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testDefaultPositionDeltaUpdatePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -2061,7 +2061,7 @@ public void testDefaultPositionDeltaUpdatePartitionedUnsortedTable() { checkPositionDeltaDistributionAndOrdering(table, UPDATE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testNonePositionDeltaUpdatePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -2098,7 +2098,7 @@ public void testNonePositionDeltaUpdatePartitionedUnsortedTable() { table, UPDATE, UNSPECIFIED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testHashPositionDeltaUpdatePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -2143,7 +2143,7 @@ public void testHashPositionDeltaUpdatePartitionedUnsortedTable() { checkPositionDeltaDistributionAndOrdering(table, UPDATE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testRangePositionDeltaUpdatePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -2190,7 +2190,7 @@ public void testRangePositionDeltaUpdatePartitionedUnsortedTable() { checkPositionDeltaDistributionAndOrdering(table, UPDATE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testDefaultPositionDeltaUpdatePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -2230,7 +2230,7 @@ public void testDefaultPositionDeltaUpdatePartitionedSortedTable() { table, UPDATE, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testNonePositionDeltaUpdatePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -2263,7 +2263,7 @@ public void testNonePositionDeltaUpdatePartitionedSortedTable() { table, UPDATE, UNSPECIFIED_DISTRIBUTION, expectedOrdering); } - @Test + @TestTemplate public void testHashPositionDeltaUpdatePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -2305,7 +2305,7 @@ public void testHashPositionDeltaUpdatePartitionedSortedTable() { table, UPDATE, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testRangePositionDeltaUpdatePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -2414,7 +2414,7 @@ public void testRangePositionDeltaUpdatePartitionedSortedTable() { // merge mode is RANGE -> RANGE DISTRIBUTE BY _spec_id, _partition, date, id // LOCALLY ORDERED BY _spec_id, _partition, _file, _pos, date, id - @Test + @TestTemplate public void testDefaultPositionDeltaMergeUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -2438,7 +2438,7 @@ public void testDefaultPositionDeltaMergeUnpartitionedUnsortedTable() { checkPositionDeltaDistributionAndOrdering(table, MERGE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testNonePositionDeltaMergeUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -2457,7 +2457,7 @@ public void testNonePositionDeltaMergeUnpartitionedUnsortedTable() { table, MERGE, UNSPECIFIED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testHashPositionDeltaMergeUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -2483,7 +2483,7 @@ public void testHashPositionDeltaMergeUnpartitionedUnsortedTable() { checkPositionDeltaDistributionAndOrdering(table, MERGE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testRangePositionDeltaMergeUnpartitionedUnsortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -2512,7 +2512,7 @@ public void testRangePositionDeltaMergeUnpartitionedUnsortedTable() { checkPositionDeltaDistributionAndOrdering(table, MERGE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testDefaultPositionDeltaMergeUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -2545,7 +2545,7 @@ public void testDefaultPositionDeltaMergeUnpartitionedSortedTable() { checkPositionDeltaDistributionAndOrdering(table, MERGE, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testNonePositionDeltaMergeUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -2573,7 +2573,7 @@ public void testNonePositionDeltaMergeUnpartitionedSortedTable() { table, MERGE, UNSPECIFIED_DISTRIBUTION, expectedOrdering); } - @Test + @TestTemplate public void testHashPositionDeltaMergeUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -2608,7 +2608,7 @@ public void testHashPositionDeltaMergeUnpartitionedSortedTable() { checkPositionDeltaDistributionAndOrdering(table, MERGE, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testRangePositionDeltaMergeUnpartitionedSortedTable() { sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); @@ -2648,7 +2648,7 @@ public void testRangePositionDeltaMergeUnpartitionedSortedTable() { checkPositionDeltaDistributionAndOrdering(table, MERGE, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testDefaultPositionDeltaMergePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -2690,7 +2690,7 @@ public void testDefaultPositionDeltaMergePartitionedUnsortedTable() { checkPositionDeltaDistributionAndOrdering(table, MERGE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testNonePositionDeltaMergePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -2727,7 +2727,7 @@ public void testNonePositionDeltaMergePartitionedUnsortedTable() { table, MERGE, UNSPECIFIED_DISTRIBUTION, EMPTY_ORDERING); } - @Test + @TestTemplate public void testHashPositionDeltaMergePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -2771,7 +2771,7 @@ public void testHashPositionDeltaMergePartitionedUnsortedTable() { checkPositionDeltaDistributionAndOrdering(table, MERGE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testRangePositionDeltaMergePartitionedUnsortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -2817,7 +2817,7 @@ public void testRangePositionDeltaMergePartitionedUnsortedTable() { checkPositionDeltaDistributionAndOrdering(table, MERGE, expectedDistribution, EMPTY_ORDERING); } - @Test + @TestTemplate public void testNonePositionDeltaMergePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -2849,7 +2849,7 @@ public void testNonePositionDeltaMergePartitionedSortedTable() { table, MERGE, UNSPECIFIED_DISTRIBUTION, expectedOrdering); } - @Test + @TestTemplate public void testDefaultPositionDeltaMergePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -2888,7 +2888,7 @@ public void testDefaultPositionDeltaMergePartitionedSortedTable() { checkPositionDeltaDistributionAndOrdering(table, MERGE, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testHashPositionDeltaMergePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -2929,7 +2929,7 @@ public void testHashPositionDeltaMergePartitionedSortedTable() { checkPositionDeltaDistributionAndOrdering(table, MERGE, expectedDistribution, expectedOrdering); } - @Test + @TestTemplate public void testRangePositionDeltaMergePartitionedSortedTable() { sql( "CREATE TABLE %s (id BIGINT, data STRING, date DATE, ts TIMESTAMP) " @@ -2978,10 +2978,10 @@ private void checkWriteDistributionAndOrdering( SparkWriteRequirements requirements = writeConf.writeRequirements(); Distribution distribution = requirements.distribution(); - Assert.assertEquals("Distribution must match", expectedDistribution, distribution); + assertThat(distribution).as("Distribution must match").isEqualTo(expectedDistribution); SortOrder[] ordering = requirements.ordering(); - Assert.assertArrayEquals("Ordering must match", expectedOrdering, ordering); + assertThat(ordering).as("Ordering must match").isEqualTo(expectedOrdering); } private void checkCopyOnWriteDistributionAndOrdering( @@ -2994,10 +2994,10 @@ private void checkCopyOnWriteDistributionAndOrdering( SparkWriteRequirements requirements = writeConf.copyOnWriteRequirements(command); Distribution distribution = requirements.distribution(); - Assert.assertEquals("Distribution must match", expectedDistribution, distribution); + assertThat(distribution).as("Distribution must match").isEqualTo(expectedDistribution); SortOrder[] ordering = requirements.ordering(); - Assert.assertArrayEquals("Ordering must match", expectedOrdering, ordering); + assertThat(ordering).as("Ordering must match").isEqualTo(expectedOrdering); } private void checkPositionDeltaDistributionAndOrdering( @@ -3010,10 +3010,10 @@ private void checkPositionDeltaDistributionAndOrdering( SparkWriteRequirements requirements = writeConf.positionDeltaRequirements(command); Distribution distribution = requirements.distribution(); - Assert.assertEquals("Distribution must match", expectedDistribution, distribution); + assertThat(distribution).as("Distribution must match").isEqualTo(expectedDistribution); SortOrder[] ordering = requirements.ordering(); - Assert.assertArrayEquals("Ordering must match", expectedOrdering, ordering); + assertThat(ordering).as("Ordering must match").isEqualTo(expectedOrdering); } private void disableFanoutWriters(Table table) { diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkFilters.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkFilters.java index b1f8cf61a755..a6205ae9ea3f 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkFilters.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkFilters.java @@ -18,6 +18,8 @@ */ package org.apache.iceberg.spark; +import static org.assertj.core.api.Assertions.assertThat; + import java.sql.Date; import java.sql.Timestamp; import java.time.Instant; @@ -40,8 +42,7 @@ import org.apache.spark.sql.sources.LessThan; import org.apache.spark.sql.sources.LessThanOrEqual; import org.apache.spark.sql.sources.Not; -import org.junit.Assert; -import org.junit.Test; +import org.junit.jupiter.api.Test; public class TestSparkFilters { @@ -59,54 +60,61 @@ public void testQuotedAttributes() { IsNull isNull = IsNull.apply(quoted); Expression expectedIsNull = Expressions.isNull(unquoted); Expression actualIsNull = SparkFilters.convert(isNull); - Assert.assertEquals( - "IsNull must match", expectedIsNull.toString(), actualIsNull.toString()); + assertThat(actualIsNull.toString()) + .as("IsNull must match") + .isEqualTo(expectedIsNull.toString()); IsNotNull isNotNull = IsNotNull.apply(quoted); Expression expectedIsNotNull = Expressions.notNull(unquoted); Expression actualIsNotNull = SparkFilters.convert(isNotNull); - Assert.assertEquals( - "IsNotNull must match", expectedIsNotNull.toString(), actualIsNotNull.toString()); + assertThat(actualIsNotNull.toString()) + .as("IsNotNull must match") + .isEqualTo(expectedIsNotNull.toString()); LessThan lt = LessThan.apply(quoted, 1); Expression expectedLt = Expressions.lessThan(unquoted, 1); Expression actualLt = SparkFilters.convert(lt); - Assert.assertEquals("LessThan must match", expectedLt.toString(), actualLt.toString()); + assertThat(actualLt.toString()) + .as("LessThan must match") + .isEqualTo(expectedLt.toString()); LessThanOrEqual ltEq = LessThanOrEqual.apply(quoted, 1); Expression expectedLtEq = Expressions.lessThanOrEqual(unquoted, 1); Expression actualLtEq = SparkFilters.convert(ltEq); - Assert.assertEquals( - "LessThanOrEqual must match", expectedLtEq.toString(), actualLtEq.toString()); + assertThat(actualLtEq.toString()) + .as("LessThanOrEqual must match") + .isEqualTo(expectedLtEq.toString()); GreaterThan gt = GreaterThan.apply(quoted, 1); Expression expectedGt = Expressions.greaterThan(unquoted, 1); Expression actualGt = SparkFilters.convert(gt); - Assert.assertEquals("GreaterThan must match", expectedGt.toString(), actualGt.toString()); + assertThat(actualGt.toString()) + .as("GreaterThan must match") + .isEqualTo(expectedGt.toString()); GreaterThanOrEqual gtEq = GreaterThanOrEqual.apply(quoted, 1); Expression expectedGtEq = Expressions.greaterThanOrEqual(unquoted, 1); Expression actualGtEq = SparkFilters.convert(gtEq); - Assert.assertEquals( - "GreaterThanOrEqual must match", expectedGtEq.toString(), actualGtEq.toString()); + assertThat(actualGtEq.toString()) + .as("GreaterThanOrEqual must match") + .isEqualTo(expectedGtEq.toString()); EqualTo eq = EqualTo.apply(quoted, 1); Expression expectedEq = Expressions.equal(unquoted, 1); Expression actualEq = SparkFilters.convert(eq); - Assert.assertEquals("EqualTo must match", expectedEq.toString(), actualEq.toString()); + assertThat(actualEq.toString()).as("EqualTo must match").isEqualTo(expectedEq.toString()); EqualNullSafe eqNullSafe = EqualNullSafe.apply(quoted, 1); Expression expectedEqNullSafe = Expressions.equal(unquoted, 1); Expression actualEqNullSafe = SparkFilters.convert(eqNullSafe); - Assert.assertEquals( - "EqualNullSafe must match", - expectedEqNullSafe.toString(), - actualEqNullSafe.toString()); + assertThat(actualEqNullSafe.toString()) + .as("EqualNullSafe must match") + .isEqualTo(expectedEqNullSafe.toString()); In in = In.apply(quoted, new Integer[] {1}); Expression expectedIn = Expressions.in(unquoted, 1); Expression actualIn = SparkFilters.convert(in); - Assert.assertEquals("In must match", expectedIn.toString(), actualIn.toString()); + assertThat(actualIn.toString()).as("In must match").isEqualTo(expectedIn.toString()); }); } @@ -120,14 +128,13 @@ public void testTimestampFilterConversion() { Expression timestampExpression = SparkFilters.convert(GreaterThan.apply("x", timestamp)); Expression rawExpression = Expressions.greaterThan("x", epochMicros); - Assert.assertEquals( - "Generated Timestamp expression should be correct", - rawExpression.toString(), - timestampExpression.toString()); - Assert.assertEquals( - "Generated Instant expression should be correct", - rawExpression.toString(), - instantExpression.toString()); + assertThat(timestampExpression.toString()) + .as("Generated Timestamp expression should be correct") + .isEqualTo(rawExpression.toString()); + + assertThat(instantExpression.toString()) + .as("Generated Instant expression should be correct") + .isEqualTo(rawExpression.toString()); } @Test @@ -139,10 +146,9 @@ public void testLocalDateTimeFilterConversion() { Expression instantExpression = SparkFilters.convert(GreaterThan.apply("x", ldt)); Expression rawExpression = Expressions.greaterThan("x", epochMicros); - Assert.assertEquals( - "Generated Instant expression should be correct", - rawExpression.toString(), - instantExpression.toString()); + assertThat(instantExpression.toString()) + .as("Generated Instant expression should be correct") + .isEqualTo(rawExpression.toString()); } @Test @@ -155,15 +161,13 @@ public void testDateFilterConversion() { Expression dateExpression = SparkFilters.convert(GreaterThan.apply("x", date)); Expression rawExpression = Expressions.greaterThan("x", epochDay); - Assert.assertEquals( - "Generated localdate expression should be correct", - rawExpression.toString(), - localDateExpression.toString()); + assertThat(localDateExpression.toString()) + .as("Generated localdate expression should be correct") + .isEqualTo(rawExpression.toString()); - Assert.assertEquals( - "Generated date expression should be correct", - rawExpression.toString(), - dateExpression.toString()); + assertThat(dateExpression.toString()) + .as("Generated date expression should be correct") + .isEqualTo(rawExpression.toString()); } @Test @@ -171,7 +175,7 @@ public void testNestedInInsideNot() { Not filter = Not.apply(And.apply(EqualTo.apply("col1", 1), In.apply("col2", new Integer[] {1, 2}))); Expression converted = SparkFilters.convert(filter); - Assert.assertNull("Expression should not be converted", converted); + assertThat(converted).as("Expression should not be converted").isNull(); } @Test @@ -180,6 +184,6 @@ public void testNotIn() { Expression actual = SparkFilters.convert(filter); Expression expected = Expressions.and(Expressions.notNull("col"), Expressions.notIn("col", 1, 2)); - Assert.assertEquals("Expressions should match", expected.toString(), actual.toString()); + assertThat(actual.toString()).as("Expressions should match").isEqualTo(expected.toString()); } } diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java index 2264efe79e05..4d4091bf9a9a 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java @@ -19,6 +19,7 @@ package org.apache.iceberg.spark; import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.assertj.core.api.Assertions.assertThat; import java.io.IOException; import java.util.List; @@ -29,8 +30,7 @@ import org.apache.spark.sql.catalyst.expressions.MetadataAttribute; import org.apache.spark.sql.catalyst.types.DataTypeUtils; import org.apache.spark.sql.types.StructType; -import org.junit.Assert; -import org.junit.Test; +import org.junit.jupiter.api.Test; public class TestSparkSchemaUtil { private static final Schema TEST_SCHEMA = @@ -46,23 +46,22 @@ public class TestSparkSchemaUtil { @Test public void testEstimateSizeMaxValue() throws IOException { - Assert.assertEquals( - "estimateSize returns Long max value", - Long.MAX_VALUE, - SparkSchemaUtil.estimateSize(null, Long.MAX_VALUE)); + assertThat(SparkSchemaUtil.estimateSize(null, Long.MAX_VALUE)) + .as("estimateSize returns Long max value") + .isEqualTo(Long.MAX_VALUE); } @Test public void testEstimateSizeWithOverflow() throws IOException { long tableSize = SparkSchemaUtil.estimateSize(SparkSchemaUtil.convert(TEST_SCHEMA), Long.MAX_VALUE - 1); - Assert.assertEquals("estimateSize handles overflow", Long.MAX_VALUE, tableSize); + assertThat(tableSize).as("estimateSize handles overflow").isEqualTo(Long.MAX_VALUE); } @Test public void testEstimateSize() throws IOException { long tableSize = SparkSchemaUtil.estimateSize(SparkSchemaUtil.convert(TEST_SCHEMA), 1); - Assert.assertEquals("estimateSize matches with expected approximation", 24, tableSize); + assertThat(tableSize).as("estimateSize matches with expected approximation").isEqualTo(24); } @Test @@ -72,13 +71,13 @@ public void testSchemaConversionWithMetaDataColumnSchema() { scala.collection.JavaConverters.seqAsJavaList(DataTypeUtils.toAttributes(structType)); for (AttributeReference attrRef : attrRefs) { if (MetadataColumns.isMetadataColumn(attrRef.name())) { - Assert.assertTrue( - "metadata columns should have __metadata_col in attribute metadata", - MetadataAttribute.unapply(attrRef).isDefined()); + assertThat(MetadataAttribute.unapply(attrRef).isDefined()) + .as("metadata columns should have __metadata_col in attribute metadata") + .isTrue(); } else { - Assert.assertFalse( - "non metadata columns should not have __metadata_col in attribute metadata", - MetadataAttribute.unapply(attrRef).isDefined()); + assertThat(MetadataAttribute.unapply(attrRef).isDefined()) + .as("non metadata columns should not have __metadata_col in attribute metadata") + .isFalse(); } } } diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkSessionCatalog.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkSessionCatalog.java index 82a2fb473360..b8062a4a49fe 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkSessionCatalog.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkSessionCatalog.java @@ -19,18 +19,19 @@ package org.apache.iceberg.spark; import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTOREURIS; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; -import org.junit.Assert; -import org.junit.Before; -import org.junit.BeforeClass; -import org.junit.Test; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; -public class TestSparkSessionCatalog extends SparkTestBase { +public class TestSparkSessionCatalog extends TestBase { private final String envHmsUriKey = "spark.hadoop." + METASTOREURIS.varname; private final String catalogHmsUriKey = "spark.sql.catalog.spark_catalog.uri"; private final String hmsUri = hiveConf.get(METASTOREURIS.varname); - @BeforeClass + @BeforeAll public static void setUpCatalog() { spark .conf() @@ -38,7 +39,7 @@ public static void setUpCatalog() { spark.conf().set("spark.sql.catalog.spark_catalog.type", "hive"); } - @Before + @BeforeEach public void setupHmsUri() { spark.sessionState().catalogManager().reset(); spark.conf().set(envHmsUriKey, hmsUri); @@ -48,52 +49,35 @@ public void setupHmsUri() { @Test public void testValidateHmsUri() { // HMS uris match - Assert.assertTrue( - spark - .sessionState() - .catalogManager() - .v2SessionCatalog() - .defaultNamespace()[0] - .equals("default")); + assertThat(spark.sessionState().catalogManager().v2SessionCatalog().defaultNamespace()[0]) + .isEqualTo("default"); // HMS uris doesn't match spark.sessionState().catalogManager().reset(); String catalogHmsUri = "RandomString"; spark.conf().set(envHmsUriKey, hmsUri); spark.conf().set(catalogHmsUriKey, catalogHmsUri); - IllegalArgumentException exception = - Assert.assertThrows( - IllegalArgumentException.class, - () -> spark.sessionState().catalogManager().v2SessionCatalog()); - String errorMessage = - String.format( - "Inconsistent Hive metastore URIs: %s (Spark session) != %s (spark_catalog)", - hmsUri, catalogHmsUri); - Assert.assertEquals(errorMessage, exception.getMessage()); + + assertThatThrownBy(() -> spark.sessionState().catalogManager().v2SessionCatalog()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage( + String.format( + "Inconsistent Hive metastore URIs: %s (Spark session) != %s (spark_catalog)", + hmsUri, catalogHmsUri)); // no env HMS uri, only catalog HMS uri spark.sessionState().catalogManager().reset(); spark.conf().set(catalogHmsUriKey, hmsUri); spark.conf().unset(envHmsUriKey); - Assert.assertTrue( - spark - .sessionState() - .catalogManager() - .v2SessionCatalog() - .defaultNamespace()[0] - .equals("default")); + assertThat(spark.sessionState().catalogManager().v2SessionCatalog().defaultNamespace()[0]) + .isEqualTo("default"); // no catalog HMS uri, only env HMS uri spark.sessionState().catalogManager().reset(); spark.conf().set(envHmsUriKey, hmsUri); spark.conf().unset(catalogHmsUriKey); - Assert.assertTrue( - spark - .sessionState() - .catalogManager() - .v2SessionCatalog() - .defaultNamespace()[0] - .equals("default")); + assertThat(spark.sessionState().catalogManager().v2SessionCatalog().defaultNamespace()[0]) + .isEqualTo("default"); } @Test @@ -102,11 +86,15 @@ public void testLoadFunction() { // load permanent UDF in Hive via FunctionCatalog spark.sql(String.format("CREATE FUNCTION perm_upper AS '%s'", functionClass)); - Assert.assertEquals("Load permanent UDF in Hive", "XYZ", scalarSql("SELECT perm_upper('xyz')")); + assertThat(scalarSql("SELECT perm_upper('xyz')")) + .as("Load permanent UDF in Hive") + .isEqualTo("XYZ"); // load temporary UDF in Hive via FunctionCatalog spark.sql(String.format("CREATE TEMPORARY FUNCTION temp_upper AS '%s'", functionClass)); - Assert.assertEquals("Load temporary UDF in Hive", "XYZ", scalarSql("SELECT temp_upper('xyz')")); + assertThat(scalarSql("SELECT temp_upper('xyz')")) + .as("Load temporary UDF in Hive") + .isEqualTo("XYZ"); // TODO: fix loading Iceberg built-in functions in SessionCatalog } diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkTableUtil.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkTableUtil.java index 1e51caadd0e9..772ae3a224ac 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkTableUtil.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkTableUtil.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.Map; import org.apache.iceberg.KryoHelpers; @@ -27,9 +29,7 @@ import org.apache.iceberg.TestHelpers; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.spark.SparkTableUtil.SparkPartition; -import org.assertj.core.api.Assertions; -import org.junit.Assert; -import org.junit.Test; +import org.junit.jupiter.api.Test; public class TestSparkTableUtil { @Test @@ -40,7 +40,7 @@ public void testSparkPartitionOKryoSerialization() throws IOException { SparkPartition sparkPartition = new SparkPartition(values, uri, format); SparkPartition deserialized = KryoHelpers.roundTripSerialize(sparkPartition); - Assertions.assertThat(sparkPartition).isEqualTo(deserialized); + assertThat(sparkPartition).isEqualTo(deserialized); } @Test @@ -51,7 +51,7 @@ public void testSparkPartitionJavaSerialization() throws IOException, ClassNotFo SparkPartition sparkPartition = new SparkPartition(values, uri, format); SparkPartition deserialized = TestHelpers.roundTripSerialize(sparkPartition); - Assertions.assertThat(sparkPartition).isEqualTo(deserialized); + assertThat(sparkPartition).isEqualTo(deserialized); } @Test @@ -68,13 +68,12 @@ public void testMetricsConfigKryoSerialization() throws Exception { MetricsConfig config = MetricsConfig.fromProperties(metricsConfig); MetricsConfig deserialized = KryoHelpers.roundTripSerialize(config); - Assert.assertEquals( - MetricsModes.Full.get().toString(), deserialized.columnMode("col1").toString()); - Assert.assertEquals( - MetricsModes.Truncate.withLength(16).toString(), - deserialized.columnMode("col2").toString()); - Assert.assertEquals( - MetricsModes.Counts.get().toString(), deserialized.columnMode("col3").toString()); + assertThat(deserialized.columnMode("col1").toString()) + .isEqualTo(MetricsModes.Full.get().toString()); + assertThat(deserialized.columnMode("col2").toString()) + .isEqualTo(MetricsModes.Truncate.withLength(16).toString()); + assertThat(deserialized.columnMode("col3").toString()) + .isEqualTo(MetricsModes.Counts.get().toString()); } @Test @@ -91,12 +90,11 @@ public void testMetricsConfigJavaSerialization() throws Exception { MetricsConfig config = MetricsConfig.fromProperties(metricsConfig); MetricsConfig deserialized = TestHelpers.roundTripSerialize(config); - Assert.assertEquals( - MetricsModes.Full.get().toString(), deserialized.columnMode("col1").toString()); - Assert.assertEquals( - MetricsModes.Truncate.withLength(16).toString(), - deserialized.columnMode("col2").toString()); - Assert.assertEquals( - MetricsModes.Counts.get().toString(), deserialized.columnMode("col3").toString()); + assertThat(deserialized.columnMode("col1").toString()) + .isEqualTo(MetricsModes.Full.get().toString()); + assertThat(deserialized.columnMode("col2").toString()) + .isEqualTo(MetricsModes.Truncate.withLength(16).toString()); + assertThat(deserialized.columnMode("col3").toString()) + .isEqualTo(MetricsModes.Counts.get().toString()); } } diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkV2Filters.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkV2Filters.java index 2a40e6e50fc5..44fb64120ca0 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkV2Filters.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkV2Filters.java @@ -18,6 +18,9 @@ */ package org.apache.iceberg.spark; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + import java.time.Instant; import java.time.LocalDate; import java.time.temporal.ChronoUnit; @@ -49,9 +52,7 @@ import org.apache.spark.sql.types.DataTypes; import org.apache.spark.sql.types.StructType; import org.apache.spark.unsafe.types.UTF8String; -import org.assertj.core.api.Assertions; -import org.junit.Assert; -import org.junit.Test; +import org.junit.jupiter.api.Test; public class TestSparkV2Filters { @@ -88,98 +89,114 @@ public void testV2Filters() { Predicate isNull = new Predicate("IS_NULL", attrOnly); Expression expectedIsNull = Expressions.isNull(unquoted); Expression actualIsNull = SparkV2Filters.convert(isNull); - Assert.assertEquals( - "IsNull must match", expectedIsNull.toString(), actualIsNull.toString()); + assertThat(actualIsNull.toString()) + .as("IsNull must match") + .isEqualTo(expectedIsNull.toString()); Predicate isNotNull = new Predicate("IS_NOT_NULL", attrOnly); Expression expectedIsNotNull = Expressions.notNull(unquoted); Expression actualIsNotNull = SparkV2Filters.convert(isNotNull); - Assert.assertEquals( - "IsNotNull must match", expectedIsNotNull.toString(), actualIsNotNull.toString()); + assertThat(actualIsNotNull.toString()) + .as("IsNotNull must match") + .isEqualTo(expectedIsNotNull.toString()); Predicate lt1 = new Predicate("<", attrAndValue); Expression expectedLt1 = Expressions.lessThan(unquoted, 1); Expression actualLt1 = SparkV2Filters.convert(lt1); - Assert.assertEquals("LessThan must match", expectedLt1.toString(), actualLt1.toString()); + assertThat(actualLt1.toString()) + .as("LessThan must match") + .isEqualTo(expectedLt1.toString()); Predicate lt2 = new Predicate("<", valueAndAttr); Expression expectedLt2 = Expressions.greaterThan(unquoted, 1); Expression actualLt2 = SparkV2Filters.convert(lt2); - Assert.assertEquals("LessThan must match", expectedLt2.toString(), actualLt2.toString()); + assertThat(actualLt2.toString()) + .as("LessThan must match") + .isEqualTo(expectedLt2.toString()); Predicate ltEq1 = new Predicate("<=", attrAndValue); Expression expectedLtEq1 = Expressions.lessThanOrEqual(unquoted, 1); Expression actualLtEq1 = SparkV2Filters.convert(ltEq1); - Assert.assertEquals( - "LessThanOrEqual must match", expectedLtEq1.toString(), actualLtEq1.toString()); + assertThat(actualLtEq1.toString()) + .as("LessThanOrEqual must match") + .isEqualTo(expectedLtEq1.toString()); Predicate ltEq2 = new Predicate("<=", valueAndAttr); Expression expectedLtEq2 = Expressions.greaterThanOrEqual(unquoted, 1); Expression actualLtEq2 = SparkV2Filters.convert(ltEq2); - Assert.assertEquals( - "LessThanOrEqual must match", expectedLtEq2.toString(), actualLtEq2.toString()); + assertThat(actualLtEq2.toString()) + .as("LessThanOrEqual must match") + .isEqualTo(expectedLtEq2.toString()); Predicate gt1 = new Predicate(">", attrAndValue); Expression expectedGt1 = Expressions.greaterThan(unquoted, 1); Expression actualGt1 = SparkV2Filters.convert(gt1); - Assert.assertEquals( - "GreaterThan must match", expectedGt1.toString(), actualGt1.toString()); + assertThat(actualGt1.toString()) + .as("GreaterThan must match") + .isEqualTo(expectedGt1.toString()); Predicate gt2 = new Predicate(">", valueAndAttr); Expression expectedGt2 = Expressions.lessThan(unquoted, 1); Expression actualGt2 = SparkV2Filters.convert(gt2); - Assert.assertEquals( - "GreaterThan must match", expectedGt2.toString(), actualGt2.toString()); + assertThat(actualGt2.toString()) + .as("GreaterThan must match") + .isEqualTo(expectedGt2.toString()); Predicate gtEq1 = new Predicate(">=", attrAndValue); Expression expectedGtEq1 = Expressions.greaterThanOrEqual(unquoted, 1); Expression actualGtEq1 = SparkV2Filters.convert(gtEq1); - Assert.assertEquals( - "GreaterThanOrEqual must match", expectedGtEq1.toString(), actualGtEq1.toString()); + assertThat(actualGtEq1.toString()) + .as("GreaterThanOrEqual must match") + .isEqualTo(expectedGtEq1.toString()); Predicate gtEq2 = new Predicate(">=", valueAndAttr); Expression expectedGtEq2 = Expressions.lessThanOrEqual(unquoted, 1); Expression actualGtEq2 = SparkV2Filters.convert(gtEq2); - Assert.assertEquals( - "GreaterThanOrEqual must match", expectedGtEq2.toString(), actualGtEq2.toString()); + assertThat(actualGtEq2.toString()) + .as("GreaterThanOrEqual must match") + .isEqualTo(expectedGtEq2.toString()); Predicate eq1 = new Predicate("=", attrAndValue); Expression expectedEq1 = Expressions.equal(unquoted, 1); Expression actualEq1 = SparkV2Filters.convert(eq1); - Assert.assertEquals("EqualTo must match", expectedEq1.toString(), actualEq1.toString()); + assertThat(actualEq1.toString()) + .as("EqualTo must match") + .isEqualTo(expectedEq1.toString()); Predicate eq2 = new Predicate("=", valueAndAttr); Expression expectedEq2 = Expressions.equal(unquoted, 1); Expression actualEq2 = SparkV2Filters.convert(eq2); - Assert.assertEquals("EqualTo must match", expectedEq2.toString(), actualEq2.toString()); + assertThat(actualEq2.toString()) + .as("EqualTo must match") + .isEqualTo(expectedEq2.toString()); Predicate notEq1 = new Predicate("<>", attrAndValue); Expression expectedNotEq1 = Expressions.notEqual(unquoted, 1); Expression actualNotEq1 = SparkV2Filters.convert(notEq1); - Assert.assertEquals( - "NotEqualTo must match", expectedNotEq1.toString(), actualNotEq1.toString()); + assertThat(actualNotEq1.toString()) + .as("NotEqualTo must match") + .isEqualTo(expectedNotEq1.toString()); Predicate notEq2 = new Predicate("<>", valueAndAttr); Expression expectedNotEq2 = Expressions.notEqual(unquoted, 1); Expression actualNotEq2 = SparkV2Filters.convert(notEq2); - Assert.assertEquals( - "NotEqualTo must match", expectedNotEq2.toString(), actualNotEq2.toString()); + assertThat(actualNotEq2.toString()) + .as("NotEqualTo must match") + .isEqualTo(expectedNotEq2.toString()); Predicate eqNullSafe1 = new Predicate("<=>", attrAndValue); Expression expectedEqNullSafe1 = Expressions.equal(unquoted, 1); Expression actualEqNullSafe1 = SparkV2Filters.convert(eqNullSafe1); - Assert.assertEquals( - "EqualNullSafe must match", - expectedEqNullSafe1.toString(), - actualEqNullSafe1.toString()); + assertThat(actualEqNullSafe1.toString()) + .as("EqualNullSafe must match") + .isEqualTo(expectedEqNullSafe1.toString()); Predicate eqNullSafe2 = new Predicate("<=>", valueAndAttr); Expression expectedEqNullSafe2 = Expressions.equal(unquoted, 1); Expression actualEqNullSafe2 = SparkV2Filters.convert(eqNullSafe2); - Assert.assertEquals( - "EqualNullSafe must match", - expectedEqNullSafe2.toString(), - actualEqNullSafe2.toString()); + assertThat(actualEqNullSafe2.toString()) + .as("EqualNullSafe must match") + .isEqualTo(expectedEqNullSafe2.toString()); LiteralValue str = new LiteralValue(UTF8String.fromString("iceberg"), DataTypes.StringType); @@ -188,18 +205,19 @@ public void testV2Filters() { Predicate startsWith = new Predicate("STARTS_WITH", attrAndStr); Expression expectedStartsWith = Expressions.startsWith(unquoted, "iceberg"); Expression actualStartsWith = SparkV2Filters.convert(startsWith); - Assert.assertEquals( - "StartsWith must match", expectedStartsWith.toString(), actualStartsWith.toString()); + assertThat(actualStartsWith.toString()) + .as("StartsWith must match") + .isEqualTo(expectedStartsWith.toString()); Predicate in = new Predicate("IN", attrAndValue); Expression expectedIn = Expressions.in(unquoted, 1); Expression actualIn = SparkV2Filters.convert(in); - Assert.assertEquals("In must match", expectedIn.toString(), actualIn.toString()); + assertThat(actualIn.toString()).as("In must match").isEqualTo(expectedIn.toString()); Predicate and = new And(lt1, eq1); Expression expectedAnd = Expressions.and(expectedLt1, expectedEq1); Expression actualAnd = SparkV2Filters.convert(and); - Assert.assertEquals("And must match", expectedAnd.toString(), actualAnd.toString()); + assertThat(actualAnd.toString()).as("And must match").isEqualTo(expectedAnd.toString()); org.apache.spark.sql.connector.expressions.Expression[] attrAndAttr = new org.apache.spark.sql.connector.expressions.Expression[] { @@ -208,21 +226,21 @@ public void testV2Filters() { Predicate invalid = new Predicate("<", attrAndAttr); Predicate andWithInvalidLeft = new And(invalid, eq1); Expression convertedAnd = SparkV2Filters.convert(andWithInvalidLeft); - Assert.assertEquals("And must match", convertedAnd, null); + assertThat(convertedAnd).as("And must match").isNull(); Predicate or = new Or(lt1, eq1); Expression expectedOr = Expressions.or(expectedLt1, expectedEq1); Expression actualOr = SparkV2Filters.convert(or); - Assert.assertEquals("Or must match", expectedOr.toString(), actualOr.toString()); + assertThat(actualOr.toString()).as("Or must match").isEqualTo(expectedOr.toString()); Predicate orWithInvalidLeft = new Or(invalid, eq1); Expression convertedOr = SparkV2Filters.convert(orWithInvalidLeft); - Assert.assertEquals("Or must match", convertedOr, null); + assertThat(convertedOr).as("Or must match").isNull(); Predicate not = new Not(lt1); Expression expectedNot = Expressions.not(expectedLt1); Expression actualNot = SparkV2Filters.convert(not); - Assert.assertEquals("Not must match", expectedNot.toString(), actualNot.toString()); + assertThat(actualNot.toString()).as("Not must match").isEqualTo(expectedNot.toString()); }); } @@ -238,23 +256,23 @@ public void testEqualToNull() { new org.apache.spark.sql.connector.expressions.Expression[] {value, namedReference}; Predicate eq1 = new Predicate("=", attrAndValue); - Assertions.assertThatThrownBy(() -> SparkV2Filters.convert(eq1)) + assertThatThrownBy(() -> SparkV2Filters.convert(eq1)) .isInstanceOf(NullPointerException.class) .hasMessageContaining("Expression is always false"); Predicate eq2 = new Predicate("=", valueAndAttr); - Assertions.assertThatThrownBy(() -> SparkV2Filters.convert(eq2)) + assertThatThrownBy(() -> SparkV2Filters.convert(eq2)) .isInstanceOf(NullPointerException.class) .hasMessageContaining("Expression is always false"); Predicate eqNullSafe1 = new Predicate("<=>", attrAndValue); Expression expectedEqNullSafe = Expressions.isNull(col); Expression actualEqNullSafe1 = SparkV2Filters.convert(eqNullSafe1); - Assertions.assertThat(actualEqNullSafe1.toString()).isEqualTo(expectedEqNullSafe.toString()); + assertThat(actualEqNullSafe1.toString()).isEqualTo(expectedEqNullSafe.toString()); Predicate eqNullSafe2 = new Predicate("<=>", valueAndAttr); Expression actualEqNullSafe2 = SparkV2Filters.convert(eqNullSafe2); - Assertions.assertThat(actualEqNullSafe2.toString()).isEqualTo(expectedEqNullSafe.toString()); + assertThat(actualEqNullSafe2.toString()).isEqualTo(expectedEqNullSafe.toString()); } @Test @@ -271,11 +289,11 @@ public void testEqualToNaN() { Predicate eqNaN1 = new Predicate("=", attrAndValue); Expression expectedEqNaN = Expressions.isNaN(col); Expression actualEqNaN1 = SparkV2Filters.convert(eqNaN1); - Assertions.assertThat(actualEqNaN1.toString()).isEqualTo(expectedEqNaN.toString()); + assertThat(actualEqNaN1.toString()).isEqualTo(expectedEqNaN.toString()); Predicate eqNaN2 = new Predicate("=", valueAndAttr); Expression actualEqNaN2 = SparkV2Filters.convert(eqNaN2); - Assertions.assertThat(actualEqNaN2.toString()).isEqualTo(expectedEqNaN.toString()); + assertThat(actualEqNaN2.toString()).isEqualTo(expectedEqNaN.toString()); } @Test @@ -290,12 +308,12 @@ public void testNotEqualToNull() { new org.apache.spark.sql.connector.expressions.Expression[] {value, namedReference}; Predicate notEq1 = new Predicate("<>", attrAndValue); - Assertions.assertThatThrownBy(() -> SparkV2Filters.convert(notEq1)) + assertThatThrownBy(() -> SparkV2Filters.convert(notEq1)) .isInstanceOf(NullPointerException.class) .hasMessageContaining("Expression is always false"); Predicate notEq2 = new Predicate("<>", valueAndAttr); - Assertions.assertThatThrownBy(() -> SparkV2Filters.convert(notEq2)) + assertThatThrownBy(() -> SparkV2Filters.convert(notEq2)) .isInstanceOf(NullPointerException.class) .hasMessageContaining("Expression is always false"); } @@ -314,11 +332,11 @@ public void testNotEqualToNaN() { Predicate notEqNaN1 = new Predicate("<>", attrAndValue); Expression expectedNotEqNaN = Expressions.notNaN(col); Expression actualNotEqNaN1 = SparkV2Filters.convert(notEqNaN1); - Assertions.assertThat(actualNotEqNaN1.toString()).isEqualTo(expectedNotEqNaN.toString()); + assertThat(actualNotEqNaN1.toString()).isEqualTo(expectedNotEqNaN.toString()); Predicate notEqNaN2 = new Predicate("<>", valueAndAttr); Expression actualNotEqNaN2 = SparkV2Filters.convert(notEqNaN2); - Assertions.assertThat(actualNotEqNaN2.toString()).isEqualTo(expectedNotEqNaN.toString()); + assertThat(actualNotEqNaN2.toString()).isEqualTo(expectedNotEqNaN.toString()); } @Test @@ -378,10 +396,9 @@ public void testTimestampFilterConversion() { Expression tsExpression = SparkV2Filters.convert(predicate); Expression rawExpression = Expressions.greaterThan("x", epochMicros); - Assert.assertEquals( - "Generated Timestamp expression should be correct", - rawExpression.toString(), - tsExpression.toString()); + assertThat(tsExpression.toString()) + .as("Generated Timestamp expression should be correct") + .isEqualTo(rawExpression.toString()); } @Test @@ -398,10 +415,9 @@ public void testDateFilterConversion() { Expression dateExpression = SparkV2Filters.convert(predicate); Expression rawExpression = Expressions.greaterThan("x", epochDay); - Assert.assertEquals( - "Generated date expression should be correct", - rawExpression.toString(), - dateExpression.toString()); + assertThat(dateExpression.toString()) + .as("Generated date expression should be correct") + .isEqualTo(rawExpression.toString()); } @Test @@ -420,7 +436,7 @@ public void testNestedInInsideNot() { Not filter = new Not(new And(equal, in)); Expression converted = SparkV2Filters.convert(filter); - Assert.assertNull("Expression should not be converted", converted); + assertThat(converted).as("Expression should not be converted").isNull(); } @Test @@ -437,7 +453,7 @@ public void testNotIn() { Expression actual = SparkV2Filters.convert(not); Expression expected = Expressions.and(Expressions.notNull("col"), Expressions.notIn("col", 1, 2)); - Assert.assertEquals("Expressions should match", expected.toString(), actual.toString()); + assertThat(actual.toString()).as("Expressions should match").isEqualTo(expected.toString()); } @Test @@ -630,7 +646,7 @@ public void testUnsupportedUDFConvert() { Predicate predicate = new Predicate("=", expressions(udf, literalValue)); Expression icebergExpr = SparkV2Filters.convert(predicate); - Assertions.assertThat(icebergExpr).isNull(); + assertThat(icebergExpr).isNull(); } private void testUDF( @@ -746,7 +762,7 @@ private void testUDF( Predicate invalid = new Predicate("<", attrAndAttr); Predicate andWithInvalidLeft = new And(invalid, eq1); Expression convertedAnd = SparkV2Filters.convert(andWithInvalidLeft); - Assertions.assertThat(convertedAnd).isNull(); + assertThat(convertedAnd).isNull(); Predicate or = new Or(lt1, eq1); Expression expectedOr = Expressions.or(expectedLt1, expectedEq1); @@ -755,7 +771,7 @@ private void testUDF( Predicate orWithInvalidLeft = new Or(invalid, eq1); Expression convertedOr = SparkV2Filters.convert(orWithInvalidLeft); - Assertions.assertThat(convertedOr).isNull(); + assertThat(convertedOr).isNull(); Predicate not = new Not(lt1); Expression expectedNot = Expressions.not(expectedLt1); @@ -764,7 +780,7 @@ private void testUDF( } private static void assertEquals(Expression expected, Expression actual) { - Assertions.assertThat(ExpressionUtil.equivalent(expected, actual, STRUCT, true)).isTrue(); + assertThat(ExpressionUtil.equivalent(expected, actual, STRUCT, true)).isTrue(); } private org.apache.spark.sql.connector.expressions.Expression[] expressions( diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkValueConverter.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkValueConverter.java index 7f00c7edd8a9..c7a2e6c18fca 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkValueConverter.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkValueConverter.java @@ -18,14 +18,15 @@ */ package org.apache.iceberg.spark; +import static org.assertj.core.api.Assertions.assertThat; + import org.apache.iceberg.Schema; import org.apache.iceberg.data.GenericRecord; import org.apache.iceberg.data.Record; import org.apache.iceberg.types.Types; import org.apache.spark.sql.Row; import org.apache.spark.sql.RowFactory; -import org.junit.Assert; -import org.junit.Test; +import org.junit.jupiter.api.Test; public class TestSparkValueConverter { @Test @@ -86,9 +87,8 @@ private void assertCorrectNullConversion(Schema schema) { Row sparkRow = RowFactory.create(1, null); Record record = GenericRecord.create(schema); record.set(0, 1); - Assert.assertEquals( - "Round-trip conversion should produce original value", - record, - SparkValueConverter.convert(schema, sparkRow)); + assertThat(SparkValueConverter.convert(schema, sparkRow)) + .as("Round-trip conversion should produce original value") + .isEqualTo(record); } } 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 7c3ecac676ef..30b3b8a88576 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 @@ -57,15 +57,15 @@ import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.spark.sql.internal.SQLConf; -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; -public class TestSparkWriteConf extends SparkTestBaseWithCatalog { +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 +73,12 @@ public void before() { tableName); } - @After + @AfterEach public void after() { sql("DROP TABLE IF EXISTS %s", tableName); } - @Test + @TestTemplate public void testDeleteGranularityDefault() { Table table = validationCatalog.loadTable(tableIdent); SparkWriteConf writeConf = new SparkWriteConf(spark, table, ImmutableMap.of()); @@ -87,7 +87,7 @@ public void testDeleteGranularityDefault() { assertThat(value).isEqualTo(DeleteGranularity.PARTITION); } - @Test + @TestTemplate public void testDeleteGranularityTableProperty() { Table table = validationCatalog.loadTable(tableIdent); @@ -102,7 +102,7 @@ public void testDeleteGranularityTableProperty() { assertThat(value).isEqualTo(DeleteGranularity.FILE); } - @Test + @TestTemplate public void testDeleteGranularityWriteOption() { Table table = validationCatalog.loadTable(tableIdent); @@ -120,7 +120,7 @@ public void testDeleteGranularityWriteOption() { assertThat(value).isEqualTo(DeleteGranularity.FILE); } - @Test + @TestTemplate public void testDeleteGranularityInvalidValue() { Table table = validationCatalog.loadTable(tableIdent); @@ -133,7 +133,7 @@ public void testDeleteGranularityInvalidValue() { .hasMessageContaining("Unknown delete granularity"); } - @Test + @TestTemplate public void testAdvisoryPartitionSize() { Table table = validationCatalog.loadTable(tableIdent); @@ -151,7 +151,7 @@ public void testAdvisoryPartitionSize() { assertThat(value3).isGreaterThan(10L * 1024 * 1024); } - @Test + @TestTemplate public void testSparkWriteConfDistributionDefault() { Table table = validationCatalog.loadTable(tableIdent); @@ -160,7 +160,7 @@ public void testSparkWriteConfDistributionDefault() { checkMode(DistributionMode.HASH, writeConf); } - @Test + @TestTemplate public void testSparkWriteConfDistributionModeWithWriteOption() { Table table = validationCatalog.loadTable(tableIdent); @@ -171,7 +171,7 @@ public void testSparkWriteConfDistributionModeWithWriteOption() { checkMode(DistributionMode.NONE, writeConf); } - @Test + @TestTemplate public void testSparkWriteConfDistributionModeWithSessionConfig() { withSQLConf( ImmutableMap.of(SparkSQLProperties.DISTRIBUTION_MODE, DistributionMode.NONE.modeName()), @@ -182,7 +182,7 @@ public void testSparkWriteConfDistributionModeWithSessionConfig() { }); } - @Test + @TestTemplate public void testSparkWriteConfDistributionModeWithTableProperties() { Table table = validationCatalog.loadTable(tableIdent); @@ -198,7 +198,7 @@ public void testSparkWriteConfDistributionModeWithTableProperties() { checkMode(DistributionMode.NONE, writeConf); } - @Test + @TestTemplate public void testSparkWriteConfDistributionModeWithTblPropAndSessionConfig() { withSQLConf( ImmutableMap.of(SparkSQLProperties.DISTRIBUTION_MODE, DistributionMode.NONE.modeName()), @@ -219,7 +219,7 @@ public void testSparkWriteConfDistributionModeWithTblPropAndSessionConfig() { }); } - @Test + @TestTemplate public void testSparkWriteConfDistributionModeWithWriteOptionAndSessionConfig() { withSQLConf( ImmutableMap.of(SparkSQLProperties.DISTRIBUTION_MODE, DistributionMode.RANGE.modeName()), @@ -236,7 +236,7 @@ public void testSparkWriteConfDistributionModeWithWriteOptionAndSessionConfig() }); } - @Test + @TestTemplate public void testSparkWriteConfDistributionModeWithEverything() { withSQLConf( ImmutableMap.of(SparkSQLProperties.DISTRIBUTION_MODE, DistributionMode.RANGE.modeName()), @@ -261,7 +261,7 @@ public void testSparkWriteConfDistributionModeWithEverything() { }); } - @Test + @TestTemplate public void testSparkConfOverride() { List>> propertiesSuites = Lists.newArrayList( @@ -334,7 +334,7 @@ public void testSparkConfOverride() { } } - @Test + @TestTemplate public void testDataPropsDefaultsAsDeleteProps() { List>> propertiesSuites = Lists.newArrayList( @@ -403,7 +403,7 @@ public void testDataPropsDefaultsAsDeleteProps() { } } - @Test + @TestTemplate public void testDeleteFileWriteConf() { List>> propertiesSuites = Lists.newArrayList( @@ -498,9 +498,9 @@ private void testWriteProperties(List> propertiesSuite) { SparkWriteConf writeConf = new SparkWriteConf(spark, table, ImmutableMap.of()); Map writeProperties = writeConf.writeProperties(); Map expectedProperties = propertiesSuite.get(2); - Assert.assertEquals(expectedProperties.size(), writeConf.writeProperties().size()); + assertThat(writeConf.writeProperties()).hasSameSizeAs(expectedProperties); for (Map.Entry entry : writeProperties.entrySet()) { - Assert.assertEquals(entry.getValue(), expectedProperties.get(entry.getKey())); + assertThat(expectedProperties).containsEntry(entry.getKey(), entry.getValue()); } table.refresh(); @@ -514,12 +514,12 @@ private void testWriteProperties(List> propertiesSuite) { } 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); } }