This is an automated email from the ASF dual-hosted git repository.
huaxingao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new f8f4b7476c Spark 4.2: Test legacy position deletes when upgrading to
v3 tables (#18214)
f8f4b7476c is described below
commit f8f4b7476c42a592b6a50ef59dd9dfb1c21757ae
Author: Manu Zhang <[email protected]>
AuthorDate: Thu Oct 29 09:31:32 2026 +0800
Spark 4.2: Test legacy position deletes when upgrading to v3 tables (#18214)
Position delete files that contain deletes for more than one data file
must be kept in table metadata until all of their deletes are replaced
by DVs. Cover that in the reader and in the rewrite position deletes
action.
Generated-by: Claude Code
---
.../TestRewritePositionDeleteFilesAction.java | 37 ++++++
.../spark/source/TestSparkReaderDeletes.java | 143 +++++++++++++++++----
2 files changed, 153 insertions(+), 27 deletions(-)
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java
index ba7a2af2fe..411d08a1f0 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java
@@ -890,6 +890,43 @@ public class TestRewritePositionDeleteFilesAction extends
CatalogTestBase {
.isEmpty();
}
+ @TestTemplate
+ public void rewriteSharedV2PositionDeletesToV3DVsRemovesLegacyDeleteFiles()
throws IOException {
+ Table table = createTableUnpartitioned(2, SCALE);
+ List<DataFile> dataFiles = TestHelpers.dataFiles(table);
+ writePosDeletesForFiles(table, 1, DELETES_SCALE, dataFiles);
+ assertThat(dataFiles).hasSize(2);
+
+ List<DeleteFile> deleteFiles = deleteFiles(table);
+ assertThat(deleteFiles).hasSize(1).allMatch(file -> file.format() ==
FileFormat.PARQUET);
+
+ List<Object[]> expectedRecords = records(table);
+ List<Object[]> expectedDeletes = deleteRecords(table);
+ assertThat(expectedRecords).hasSize(2000);
+ assertThat(expectedDeletes).hasSize(2000);
+ assertThat(dvRecords(table)).isEmpty();
+
+ table.updateProperties().set(TableProperties.FORMAT_VERSION, "3").commit();
+
+ Result result =
+ SparkActions.get(spark)
+ .rewritePositionDeletes(table)
+ .option(SizeBasedFileRewritePlanner.REWRITE_ALL, "true")
+ .execute();
+
+ List<DeleteFile> newDeleteFiles = deleteFiles(table);
+ assertThat(result.rewrittenDeleteFilesCount()).isEqualTo(1);
+ assertThat(result.addedDeleteFilesCount()).isEqualTo(2);
+ assertThat(newDeleteFiles).hasSize(2).allMatch(file -> file.format() ==
FileFormat.PUFFIN);
+ assertThat(newDeleteFiles).noneMatch(file -> file.format() ==
FileFormat.PARQUET);
+ assertThat(dvRecords(table)).hasSize(2);
+
+ List<Object[]> actualRecords = records(table);
+ List<Object[]> actualDeletes = deleteRecords(table);
+ assertEquals("Rows must match", expectedRecords, actualRecords);
+ assertEquals("Position deletes must match", expectedDeletes,
actualDeletes);
+ }
+
private List<Row> dvRecords(Table table) {
return spark
.read()
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderDeletes.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderDeletes.java
index 0d61930571..a40ffb5eab 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderDeletes.java
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderDeletes.java
@@ -41,6 +41,7 @@ import org.apache.iceberg.DataFile;
import org.apache.iceberg.DataFiles;
import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.Files;
import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.Parameter;
@@ -93,6 +94,7 @@ import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.internal.SQLConf;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.sql.vectorized.ColumnarBatch;
+import org.assertj.core.groups.Tuple;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
@@ -175,33 +177,7 @@ public class TestSparkReaderDeletes extends
DeleteReadTests {
@Override
protected Table createTable(String name, Schema schema, PartitionSpec spec) {
- Table table = catalog.createTable(TableIdentifier.of("default", name),
schema);
- TableOperations ops = ((BaseTable) table).operations();
- TableMetadata meta = ops.current();
- ops.commit(meta, meta.upgradeToFormatVersion(formatVersion));
- table
- .updateProperties()
- .set(TableProperties.DEFAULT_FILE_FORMAT, format.name())
- .set(TableProperties.DATA_PLANNING_MODE, planningMode.modeName())
- .set(TableProperties.DELETE_PLANNING_MODE, planningMode.modeName())
- .set(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion))
- .commit();
- if (format.equals(FileFormat.PARQUET) || format.equals(FileFormat.ORC)) {
- String vectorizationEnabled =
- format.equals(FileFormat.PARQUET)
- ? TableProperties.PARQUET_VECTORIZATION_ENABLED
- : TableProperties.ORC_VECTORIZATION_ENABLED;
- String batchSize =
- format.equals(FileFormat.PARQUET)
- ? TableProperties.PARQUET_BATCH_SIZE
- : TableProperties.ORC_BATCH_SIZE;
- table.updateProperties().set(vectorizationEnabled,
String.valueOf(vectorized)).commit();
- if (vectorized) {
- // split 7 records to two batches to cover more code paths
- table.updateProperties().set(batchSize, "4").commit();
- }
- }
- return table;
+ return createTable(name, schema, formatVersion);
}
@Override
@@ -415,6 +391,89 @@ public class TestSparkReaderDeletes extends
DeleteReadTests {
checkDeleteCount(4L);
}
+ @TestTemplate
+ public void sharedPositionDeletesAreRetainedDuringPartialDVReplacement()
throws IOException {
+ assumeThat(format).isEqualTo(FileFormat.PARQUET);
+ assumeThat(formatVersion).isEqualTo(3);
+
+ String tableName = "test3";
+ Table table = createTable(tableName, SCHEMA, 2);
+
+ GenericRecord record = GenericRecord.create(table.schema());
+ List<Record> firstFileRecords =
+ Lists.newArrayList(
+ record.copy("id", 1, "data", "a"),
+ record.copy("id", 2, "data", "b"),
+ record.copy("id", 3, "data", "c"));
+ List<Record> secondFileRecords =
+ Lists.newArrayList(
+ record.copy("id", 4, "data", "d"),
+ record.copy("id", 5, "data", "e"),
+ record.copy("id", 6, "data", "f"));
+
+ DataFile firstDataFile =
+ FileHelpers.writeDataFile(
+ table,
+ Files.localOutput(temp.resolve("junit" +
System.nanoTime()).toFile()),
+ TestHelpers.Row.of(),
+ firstFileRecords);
+ DataFile secondDataFile =
+ FileHelpers.writeDataFile(
+ table,
+ Files.localOutput(temp.resolve("junit" +
System.nanoTime()).toFile()),
+ TestHelpers.Row.of(),
+ secondFileRecords);
+
table.newAppend().appendFile(firstDataFile).appendFile(secondDataFile).commit();
+
+ Pair<DeleteFile, CharSequenceSet> sharedPositionDeletes =
+ FileHelpers.writeDeleteFile(
+ table,
+ Files.localOutput(temp.resolve("junit" +
System.nanoTime()).toFile()),
+ TestHelpers.Row.of(),
+ Lists.newArrayList(
+ Pair.of(firstDataFile.location(), 0L),
+ Pair.of(firstDataFile.location(), 1L),
+ Pair.of(secondDataFile.location(), 0L)),
+ 2);
+ table
+ .newRowDelta()
+ .addDeletes(sharedPositionDeletes.first())
+ .validateDataFilesExist(sharedPositionDeletes.second())
+ .commit();
+
+ table.updateProperties().set(TableProperties.FORMAT_VERSION, "3").commit();
+
+ Pair<DeleteFile, CharSequenceSet> secondFileDV =
+ FileHelpers.writeDeleteFile(
+ table,
+ Files.localOutput(temp.resolve("junit" +
System.nanoTime()).toFile()),
+ TestHelpers.Row.of(),
+ Lists.newArrayList(
+ Pair.of(secondDataFile.location(), 0L),
Pair.of(secondDataFile.location(), 1L)),
+ 3);
+ table
+ .newRowDelta()
+ .addDeletes(secondFileDV.first())
+ .validateDataFilesExist(secondFileDV.second())
+ .commit();
+
+ StructLikeSet actual = rowSet(tableName, table, "*");
+ StructLikeSet expected = rowSetWithoutIds(table, firstFileRecords, 1, 2);
+ expected.addAll(rowSetWithoutIds(table, secondFileRecords, 4, 5));
+ assertThat(actual).as("Table should contain expected
rows").isEqualTo(expected);
+
+ try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
+ List<FileScanTask> fileTasks = Lists.newArrayList(tasks);
+ assertThat(fileTasks).hasSize(2);
+ assertThat(fileTasks).allSatisfy(task ->
assertThat(task.deletes()).hasSize(1));
+ assertThat(fileTasks)
+ .extracting(task -> task.file().location(), task ->
task.deletes().get(0).format())
+ .containsExactlyInAnyOrder(
+ Tuple.tuple(firstDataFile.location(), FileFormat.PARQUET),
+ Tuple.tuple(secondDataFile.location(), FileFormat.PUFFIN));
+ }
+ }
+
@TestTemplate
public void testEqualityDeleteWithDeletedColumn() throws IOException {
String tableName = table.name().substring(table.name().lastIndexOf(".") +
1);
@@ -869,4 +928,34 @@ public class TestSparkReaderDeletes extends
DeleteReadTests {
records.add(record.copy("id", 122, "data", "g", "_deleted", false));
return records;
}
+
+ private Table createTable(String name, Schema schema, int
tableFormatVersion) {
+ Table table = catalog.createTable(TableIdentifier.of("default", name),
schema);
+ TableOperations ops = ((BaseTable) table).operations();
+ TableMetadata meta = ops.current();
+ ops.commit(meta, meta.upgradeToFormatVersion(tableFormatVersion));
+ table
+ .updateProperties()
+ .set(TableProperties.DEFAULT_FILE_FORMAT, format.name())
+ .set(TableProperties.DATA_PLANNING_MODE, planningMode.modeName())
+ .set(TableProperties.DELETE_PLANNING_MODE, planningMode.modeName())
+ .set(TableProperties.FORMAT_VERSION,
String.valueOf(tableFormatVersion))
+ .commit();
+ if (format.equals(FileFormat.PARQUET) || format.equals(FileFormat.ORC)) {
+ String vectorizationEnabled =
+ format.equals(FileFormat.PARQUET)
+ ? TableProperties.PARQUET_VECTORIZATION_ENABLED
+ : TableProperties.ORC_VECTORIZATION_ENABLED;
+ String batchSize =
+ format.equals(FileFormat.PARQUET)
+ ? TableProperties.PARQUET_BATCH_SIZE
+ : TableProperties.ORC_BATCH_SIZE;
+ table.updateProperties().set(vectorizationEnabled,
String.valueOf(vectorized)).commit();
+ if (vectorized) {
+ // split 7 records to two batches to cover more code paths
+ table.updateProperties().set(batchSize, "4").commit();
+ }
+ }
+ return table;
+ }
}
