rdblue commented on code in PR #13555:
URL: https://github.com/apache/iceberg/pull/13555#discussion_r2223158213
##########
spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -477,71 +481,106 @@ public void testBinPackWithDeletes() throws IOException {
int total = (int)
dataFiles.stream().mapToLong(ContentFile::recordCount).sum();
RowDelta rowDelta = table.newRowDelta();
- if (formatVersion >= 3) {
- // delete 1 position for data files 0, 1, 2
- for (int i = 0; i < 3; i++) {
- writeDV(table, dataFiles.get(i).partition(),
dataFiles.get(i).location(), 1)
- .forEach(rowDelta::addDeletes);
- }
-
- // delete 2 positions for data files 3, 4
- for (int i = 3; i < 5; i++) {
- writeDV(table, dataFiles.get(i).partition(),
dataFiles.get(i).location(), 2)
- .forEach(rowDelta::addDeletes);
- }
- } else {
- // add 1 delete file for data files 0, 1, 2
- for (int i = 0; i < 3; i++) {
- writePosDeletesToFile(table, dataFiles.get(i),
1).forEach(rowDelta::addDeletes);
- }
+ // add 1 delete file for data files 0, 1, 2
+ for (int i = 0; i < 3; i++) {
+ writePosDeletesToFile(table, dataFiles.get(i),
1).forEach(rowDelta::addDeletes);
+ }
- // add 2 delete files for data files 3, 4
- for (int i = 3; i < 5; i++) {
- writePosDeletesToFile(table, dataFiles.get(i),
2).forEach(rowDelta::addDeletes);
- }
+ // add 2 delete files for data files 3, 4
+ for (int i = 3; i < 5; i++) {
+ writePosDeletesToFile(table, dataFiles.get(i),
2).forEach(rowDelta::addDeletes);
}
rowDelta.commit();
table.refresh();
List<Object[]> expectedRecords = currentData();
long dataSizeBefore = testDataSize(table);
-
- if (formatVersion >= 3) {
- Result result =
- actions()
- .rewriteDataFiles(table)
- // do not include any file based on bin pack file size configs
- .option(BinPackRewriteFilePlanner.MIN_FILE_SIZE_BYTES, "0")
- .option(RewriteDataFiles.TARGET_FILE_SIZE_BYTES,
Long.toString(Long.MAX_VALUE - 1))
- .option(BinPackRewriteFilePlanner.MAX_FILE_SIZE_BYTES,
Long.toString(Long.MAX_VALUE))
- // set DELETE_FILE_THRESHOLD to 1 since DVs only produce one
delete file per data file
- .option(BinPackRewriteFilePlanner.DELETE_FILE_THRESHOLD, "1")
- .execute();
- assertThat(result.rewrittenDataFilesCount())
- .as("Action should rewrite 5 data files")
- .isEqualTo(5);
-
assertThat(result.rewrittenBytesCount()).isGreaterThan(0L).isLessThan(dataSizeBefore);
- } else {
- Result result =
- actions()
- .rewriteDataFiles(table)
- // do not include any file based on bin pack file size configs
- .option(BinPackRewriteFilePlanner.MIN_FILE_SIZE_BYTES, "0")
- .option(RewriteDataFiles.TARGET_FILE_SIZE_BYTES,
Long.toString(Long.MAX_VALUE - 1))
- .option(BinPackRewriteFilePlanner.MAX_FILE_SIZE_BYTES,
Long.toString(Long.MAX_VALUE))
- .option(BinPackRewriteFilePlanner.DELETE_FILE_THRESHOLD, "2")
- .execute();
- assertThat(result.rewrittenDataFilesCount())
- .as("Action should rewrite 2 data files")
- .isEqualTo(2);
-
assertThat(result.rewrittenBytesCount()).isGreaterThan(0L).isLessThan(dataSizeBefore);
- }
+ Result result =
+ actions()
+ .rewriteDataFiles(table)
+ // do not include any file based on bin pack file size configs
+ .option(BinPackRewriteFilePlanner.MIN_FILE_SIZE_BYTES, "0")
+ .option(RewriteDataFiles.TARGET_FILE_SIZE_BYTES,
Long.toString(Long.MAX_VALUE - 1))
+ .option(BinPackRewriteFilePlanner.MAX_FILE_SIZE_BYTES,
Long.toString(Long.MAX_VALUE))
+ .option(BinPackRewriteFilePlanner.DELETE_FILE_THRESHOLD, "2")
+ .execute();
+ assertThat(result.rewrittenDataFilesCount())
+ .as("Action should rewrite 2 data files")
+ .isEqualTo(2);
+
assertThat(result.rewrittenBytesCount()).isGreaterThan(0L).isLessThan(dataSizeBefore);
List<Object[]> actualRecords = currentData();
assertEquals("Rows must match", expectedRecords, actualRecords);
assertThat(actualRecords).as("7 rows are removed").hasSize(total - 7);
}
+ @TestTemplate
+ public void testBinPackWithDVs() throws IOException {
+ assumeThat(formatVersion).isGreaterThanOrEqualTo(3);
+ Table table = createTablePartitioned(4, 2);
+ shouldHaveFiles(table, 8);
+ table.refresh();
+ List<Object[]> initialRecords = currentDataWithLineage();
+ Set<Long> rowIds =
+ initialRecords.stream().map(record -> (Long)
record[0]).collect(Collectors.toSet());
+ Set<Long> lastUpdatedSequenceNumbers =
+ initialRecords.stream().map(record -> (Long)
record[1]).collect(Collectors.toSet());
+ assertThat(rowIds)
+ .isEqualTo(LongStream.range(0,
initialRecords.size()).boxed().collect(Collectors.toSet()));
+ assertThat(lastUpdatedSequenceNumbers).allMatch(sequenceNumber ->
sequenceNumber.equals(1L));
+ List<DataFile> dataFiles = TestHelpers.dataFiles(table);
+ int total = (int)
dataFiles.stream().mapToLong(ContentFile::recordCount).sum();
+
+ RowDelta rowDelta = table.newRowDelta();
+ Set<Long> rowIdsBeingRemoved = Sets.newHashSet();
+
+ // add 1 DV for data files 0, 1, 2
+ for (int i = 0; i < 3; i++) {
+ writeDV(table, dataFiles.get(i).partition(),
dataFiles.get(i).location(), 1)
+ .forEach(rowDelta::addDeletes);
+ rowIdsBeingRemoved.add(dataFiles.get(i).firstRowId());
+ }
+
+ // delete 2 positions for data files 3, 4
+ for (int i = 3; i < 5; i++) {
+ writeDV(table, dataFiles.get(i).partition(),
dataFiles.get(i).location(), 2)
+ .forEach(rowDelta::addDeletes);
+ long dataFileFirstRowId = dataFiles.get(i).firstRowId();
+ rowIdsBeingRemoved.add(dataFileFirstRowId);
+ rowIdsBeingRemoved.add(dataFileFirstRowId + 1);
+ }
+
+ rowDelta.commit();
+ table.refresh();
+ List<Object[]> recordsWithLineageAfterDelete = currentDataWithLineage();
+ rowIds.removeAll(rowIdsBeingRemoved);
Review Comment:
Nit: I'd prefer using a different name for the original set of row IDs and
the set of row IDs after removal because it makes the test easier to read.
Right now, I have to remember that `rowIds` is modified between assertions at
the top and bottom of the test case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
