You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by bl...@apache.org on 2021/11/22 22:44:42 UTC
[iceberg] branch master updated: Core: Fix javadoc for delete-file-threshold, improve test (#3567)
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new 53fc8a7 Core: Fix javadoc for delete-file-threshold, improve test (#3567)
53fc8a7 is described below
commit 53fc8a7bde7ec4069fc0a1a305c29607631670be
Author: Jack Ye <yz...@amazon.com>
AuthorDate: Mon Nov 22 14:44:33 2021 -0800
Core: Fix javadoc for delete-file-threshold, improve test (#3567)
---
.../apache/iceberg/actions/BinPackStrategy.java | 12 ++--
.../spark/actions/TestRewriteDataFilesAction.java | 77 ++++++++++++++--------
2 files changed, 57 insertions(+), 32 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/actions/BinPackStrategy.java b/core/src/main/java/org/apache/iceberg/actions/BinPackStrategy.java
index 0fe809e..ae60ca7 100644
--- a/core/src/main/java/org/apache/iceberg/actions/BinPackStrategy.java
+++ b/core/src/main/java/org/apache/iceberg/actions/BinPackStrategy.java
@@ -77,7 +77,7 @@ public abstract class BinPackStrategy implements RewriteStrategy {
/**
* The minimum number of deletes that needs to be associated with a data file for it to be considered for rewriting.
- * If a data file has more than this number of deletes, it will be rewritten regardless of its file size determined
+ * If a data file has this number of deletes or more, it will be rewritten regardless of its file size determined
* by {@link #MIN_FILE_SIZE_BYTES} and {@link #MAX_FILE_SIZE_BYTES}.
* If a file group contains a file that satisfies this condition, the file group will be rewritten regardless of
* the number of files in the file group determined by {@link #MIN_INPUT_FILES}
@@ -88,7 +88,7 @@ public abstract class BinPackStrategy implements RewriteStrategy {
public static final int DELETE_FILE_THRESHOLD_DEFAULT = Integer.MAX_VALUE;
private int minInputFiles;
- private int minDeletesPerFile;
+ private int deleteFileThreshold;
private long minFileSize;
private long maxFileSize;
private long targetFileSize;
@@ -134,7 +134,7 @@ public abstract class BinPackStrategy implements RewriteStrategy {
MIN_INPUT_FILES,
MIN_INPUT_FILES_DEFAULT);
- minDeletesPerFile = PropertyUtil.propertyAsInt(options,
+ deleteFileThreshold = PropertyUtil.propertyAsInt(options,
DELETE_FILE_THRESHOLD,
DELETE_FILE_THRESHOLD_DEFAULT);
@@ -239,7 +239,7 @@ public abstract class BinPackStrategy implements RewriteStrategy {
}
private boolean taskHasTooManyDeletes(FileScanTask task) {
- return task.deletes() != null && task.deletes().size() >= minDeletesPerFile;
+ return task.deletes() != null && task.deletes().size() >= deleteFileThreshold;
}
private void validateOptions() {
@@ -263,8 +263,8 @@ public abstract class BinPackStrategy implements RewriteStrategy {
"Cannot set %s is less than 1. All values less than 1 have the same effect as 1. %d < 1",
MIN_INPUT_FILES, minInputFiles);
- Preconditions.checkArgument(minDeletesPerFile > 0,
+ Preconditions.checkArgument(deleteFileThreshold > 0,
"Cannot set %s is less than 1. All values less than 1 have the same effect as 1. %d < 1",
- DELETE_FILE_THRESHOLD, minDeletesPerFile);
+ DELETE_FILE_THRESHOLD, deleteFileThreshold);
}
}
diff --git a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
index 97151a6..61ed16a 100644
--- a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
+++ b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
@@ -20,6 +20,8 @@
package org.apache.iceberg.spark.actions;
import java.io.File;
+import java.io.IOException;
+import java.io.UncheckedIOException;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
@@ -34,6 +36,7 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.AssertHelpers;
import org.apache.iceberg.ContentFile;
import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.MetadataTableType;
@@ -61,6 +64,7 @@ import org.apache.iceberg.exceptions.CommitStateUnknownException;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.hadoop.HadoopTables;
import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
@@ -198,48 +202,36 @@ public class TestRewriteDataFilesAction extends SparkTestBase {
CloseableIterable<FileScanTask> tasks = table.newScan().planFiles();
List<DataFile> dataFiles = Lists.newArrayList(CloseableIterable.transform(tasks, FileScanTask::file));
- GenericAppenderFactory appenderFactory = new GenericAppenderFactory(table.schema(), table.spec(),
- null, null, null);
int total = (int) dataFiles.stream().mapToLong(ContentFile::recordCount).sum();
RowDelta rowDelta = table.newRowDelta();
- // remove 2 rows for odd files, 1 row for even files
- for (int i = 0; i < dataFiles.size(); i++) {
- DataFile dataFile = dataFiles.get(i);
- EncryptedOutputFile outputFile = EncryptedFiles.encryptedOutput(
- table.io().newOutputFile(table.locationProvider().newDataLocation(UUID.randomUUID().toString())),
- EncryptionKeyMetadata.EMPTY);
- PositionDeleteWriter<Record> posDeleteWriter = appenderFactory.newPosDeleteWriter(
- outputFile, FileFormat.PARQUET, dataFile.partition());
- posDeleteWriter.delete(dataFile.path(), 0);
- posDeleteWriter.close();
- rowDelta.addDeletes(posDeleteWriter.toDeleteFile());
-
- if (i % 2 != 0) {
- outputFile = EncryptedFiles.encryptedOutput(
- table.io().newOutputFile(table.locationProvider().newDataLocation(UUID.randomUUID().toString())),
- EncryptionKeyMetadata.EMPTY);
- posDeleteWriter = appenderFactory.newPosDeleteWriter(outputFile, FileFormat.PARQUET, dataFile.partition());
- posDeleteWriter.delete(dataFile.path(), 1);
- posDeleteWriter.close();
- rowDelta.addDeletes(posDeleteWriter.toDeleteFile());
- }
+ // 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);
}
rowDelta.commit();
table.refresh();
List<Object[]> expectedRecords = currentData();
Result result = actions().rewriteDataFiles(table)
+ // do not include any file based on bin pack file size configs
.option(BinPackStrategy.MIN_FILE_SIZE_BYTES, "0")
.option(RewriteDataFiles.TARGET_FILE_SIZE_BYTES, Long.toString(Long.MAX_VALUE - 1))
.option(BinPackStrategy.MAX_FILE_SIZE_BYTES, Long.toString(Long.MAX_VALUE))
- .option(BinPackStrategy.DELETE_FILE_THRESHOLD, "4")
+ .option(BinPackStrategy.DELETE_FILE_THRESHOLD, "2")
.execute();
- Assert.assertEquals("Action should rewrite 4 data files", 4, result.rewrittenDataFilesCount());
+ Assert.assertEquals("Action should rewrite 2 data files", 2, result.rewrittenDataFilesCount());
List<Object[]> actualRecords = currentData();
assertEquals("Rows must match", expectedRecords, actualRecords);
- Assert.assertEquals("12 rows are removed", total - 12, actualRecords.size());
+ Assert.assertEquals("7 rows are removed", total - 7, actualRecords.size());
}
@Test
@@ -1117,6 +1109,39 @@ public class TestRewriteDataFilesAction extends SparkTestBase {
.save(tableLocation);
}
+ private List<DeleteFile> writePosDeletesToFile(Table table, DataFile dataFile, int outputDeleteFiles) {
+ return writePosDeletes(table, dataFile.partition(), dataFile.path().toString(), outputDeleteFiles);
+ }
+
+ private List<DeleteFile> writePosDeletes(Table table, StructLike partition, String path, int outputDeleteFiles) {
+ List<DeleteFile> results = Lists.newArrayList();
+ int rowPosition = 0;
+ for (int file = 0; file < outputDeleteFiles; file++) {
+ OutputFile outputFile = table.io().newOutputFile(
+ table.locationProvider().newDataLocation(UUID.randomUUID().toString()));
+ EncryptedOutputFile encryptedOutputFile = EncryptedFiles.encryptedOutput(
+ outputFile, EncryptionKeyMetadata.EMPTY);
+
+ GenericAppenderFactory appenderFactory = new GenericAppenderFactory(
+ table.schema(), table.spec(), null, null, null);
+ PositionDeleteWriter<Record> posDeleteWriter = appenderFactory
+ .set(TableProperties.DEFAULT_WRITE_METRICS_MODE, "full")
+ .newPosDeleteWriter(encryptedOutputFile, FileFormat.PARQUET, partition);
+
+ posDeleteWriter.delete(path, rowPosition);
+ try {
+ posDeleteWriter.close();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+
+ results.add(posDeleteWriter.toDeleteFile());
+ rowPosition++;
+ }
+
+ return results;
+ }
+
private ActionsProvider actions() {
return SparkActions.get();
}