Skip to content

Spark 3.4: Migrate source and spark tests #12998

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nastra merged 3 commits into from
May 8, 2025
Merged

Spark 3.4: Migrate source and spark tests #12998

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fbe4aef
Migrate Spark 3.4 source and spark tests
tomtongue May 7, 2025
87f46e9
Run SpotlessApply
tomtongue May 7, 2025
f802fe3
Remove curly braces
tomtongue May 7, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 27 additions & 26 deletions spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestFileRewriteCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -26,34 +28,30 @@
import java.util.stream.Collectors;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Table;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.spark.source.SimpleRecord;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
import org.junit.After;
import org.junit.Assert;
import org.junit.Test;

public class TestFileRewriteCoordinator extends SparkCatalogTestBase {
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;

public TestFileRewriteCoordinator(
String catalogName, String implementation, Map<String, String> config) {
super(catalogName, implementation, config);
}
@ExtendWith(ParameterizedTestExtension.class)
public class TestFileRewriteCoordinator extends CatalogTestBase {

@After
@AfterEach
public void removeTables() {
sql("DROP TABLE IF EXISTS %s", tableName);
}

@Test
@TestTemplate
public void testBinPackRewrite() throws NoSuchTableException, IOException {
sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName);

Expand All @@ -64,7 +62,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<Row> fileDF =
spark.read().format("iceberg").load(tableName(tableIdent.name() + ".files"));
Expand Down Expand Up @@ -106,14 +104,15 @@ public void testBinPackRewrite() throws NoSuchTableException, IOException {
table.refresh();

Map<String, String> summary = table.currentSnapshot().summary();
Assert.assertEquals("Deleted files count must match", "4", summary.get("deleted-data-files"));
Assert.assertEquals("Added files count must match", "2", summary.get("added-data-files"));
assertThat(summary)
.containsEntry("deleted-data-files", "4")
.containsEntry("added-data-files", "2");

Object rowCount = scalarSql("SELECT count(*) FROM %s", tableName);
Assert.assertEquals("Row count must match", 4000L, rowCount);
assertThat(rowCount).as("Row count must match").isEqualTo(4000L);
}

@Test
@TestTemplate
public void testSortRewrite() throws NoSuchTableException, IOException {
sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName);

Expand All @@ -124,7 +123,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<FileScanTask> fileScanTasks = table.newScan().planFiles()) {
String fileSetID = UUID.randomUUID().toString();
Expand Down Expand Up @@ -176,14 +175,15 @@ public void testSortRewrite() throws NoSuchTableException, IOException {
table.refresh();

Map<String, String> summary = table.currentSnapshot().summary();
Assert.assertEquals("Deleted files count must match", "4", summary.get("deleted-data-files"));
Assert.assertEquals("Added files count must match", "2", summary.get("added-data-files"));
assertThat(summary)
.containsEntry("deleted-data-files", "4")
.containsEntry("added-data-files", "2");

Object rowCount = scalarSql("SELECT count(*) FROM %s", tableName);
Assert.assertEquals("Row count must match", 4000L, rowCount);
assertThat(rowCount).as("Row count must match").isEqualTo(4000L);
}

@Test
@TestTemplate
public void testCommitMultipleRewrites() throws NoSuchTableException, IOException {
sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName);

Expand Down Expand Up @@ -253,14 +253,15 @@ 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<String, String> summary = table.currentSnapshot().summary();
Assert.assertEquals("Deleted files count must match", "4", summary.get("deleted-data-files"));
Assert.assertEquals("Added files count must match", "2", summary.get("added-data-files"));
assertThat(summary)
.containsEntry("deleted-data-files", "4")
.containsEntry("added-data-files", "2");

Object rowCount = scalarSql("SELECT count(*) FROM %s", tableName);
Assert.assertEquals("Row count must match", 4000L, rowCount);
assertThat(rowCount).as("Row count must match").isEqualTo(4000L);
}

private Dataset<Row> newDF(int numRecords) {
Expand Down
29 changes: 20 additions & 9 deletions spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestSparkCachedTableCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,32 +18,43 @@
*/
package org.apache.iceberg.spark;

import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Parameters;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;

public class TestSparkCachedTableCatalog extends SparkTestBaseWithCatalog {
@ExtendWith(ParameterizedTestExtension.class)
public class TestSparkCachedTableCatalog extends TestBaseWithCatalog {

private static final SparkTableCache TABLE_CACHE = SparkTableCache.get();

@BeforeClass
@BeforeAll
public static void setupCachedTableCatalog() {
spark.conf().set("spark.sql.catalog.testcache", SparkCachedTableCatalog.class.getName());
}

@AfterClass
@AfterAll
public static void unsetCachedTableCatalog() {
spark.conf().unset("spark.sql.catalog.testcache");
}

public TestSparkCachedTableCatalog() {
super(SparkCatalogConfig.HIVE);
@Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
protected static Object[][] parameters() {
return new Object[][] {
{
SparkCatalogConfig.HIVE.catalogName(),
SparkCatalogConfig.HIVE.implementation(),
SparkCatalogConfig.HIVE.properties()
},
};
}

@Test
@TestTemplate
public void testTimeTravel() {
sql("CREATE TABLE %s (id INT, dep STRING) USING iceberg", tableName);

Expand Down
52 changes: 24 additions & 28 deletions spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/TestSparkCatalogOperations.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,37 +18,37 @@
*/
package org.apache.iceberg.spark;

import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.atIndex;

import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Schema;
import org.apache.iceberg.catalog.Catalog;
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
import org.apache.spark.sql.connector.catalog.Column;
import org.apache.spark.sql.connector.catalog.Identifier;
import org.apache.spark.sql.connector.catalog.Table;
import org.apache.spark.sql.connector.catalog.TableChange;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;

public class TestSparkCatalogOperations extends SparkCatalogTestBase {
public TestSparkCatalogOperations(
String catalogName, String implementation, Map<String, String> config) {
super(catalogName, implementation, config);
}
@ExtendWith(ParameterizedTestExtension.class)
public class TestSparkCatalogOperations extends CatalogTestBase {

@Before
@BeforeEach
public void createTable() {
sql("CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg", tableName);
}

@After
@AfterEach
public void removeTable() {
sql("DROP TABLE IF EXISTS %s", tableName);
}

@Test
@TestTemplate
public void testAlterTable() throws NoSuchTableException {
BaseCatalog catalog = (BaseCatalog) spark.sessionState().catalogManager().catalog(catalogName);
Identifier identifier = Identifier.of(tableIdent.namespace().levels(), tableIdent.name());
Expand All @@ -62,24 +62,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);
Column expectedField = Column.create(fieldName, DataTypes.StringType, true);
assertThat(table.columns())
.as("Adding a column to a table should return the updated table with the new column")
.contains(expectedField, atIndex(2));

Assert.assertTrue(
"Adding a property to a table should return the updated table with the new property",
table.properties().containsKey(propsKey));
Assert.assertEquals(
"Altering a table to add a new property should add the correct value",
propsValue,
table.properties().get(propsKey));
assertThat(table.properties())
.as(
"Adding a property to a table should return the updated table with the new property with the new correct value")
.containsEntry(propsKey, propsValue);
}

@Test
@TestTemplate
public void testInvalidateTable() {
// load table to CachingCatalog
sql("SELECT count(1) FROM %s", tableName);
Expand Down
Loading