Skip to content

Core: Partition prune delete manifests in ManifestFilterManager #18154

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

Open
grantatspothero wants to merge 3 commits into
base: main
Choose a base branch
from
+513 −21
Open

Core: Partition prune delete manifests in ManifestFilterManager #18154

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
3d1b472
Core: Check dropDeleteFilesOlderThan regardless of removed data files
grantatspothero Sep 17, 2026
4e7642e
Core: Measure partitioned tables and object store latency in RewriteD…
grantatspothero Sep 16, 2026
6dca9e2
Core: Prune delete manifests that cannot hold a dangling DV
grantatspothero Sep 16, 2026
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
297 changes: 281 additions & 16 deletions core/src/jmh/java/org/apache/iceberg/RewriteDataFilesBenchmark.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,14 @@
import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import org.apache.iceberg.encryption.EncryptionManager;
import org.apache.iceberg.hadoop.HadoopTables;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.io.LocationProvider;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.io.PositionOutputStream;
import org.apache.iceberg.io.SeekableInputStream;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
Expand Down Expand Up @@ -73,23 +79,46 @@ public class RewriteDataFilesBenchmark {
required(5, "timestamp_col", Types.TimestampType.withoutZone()),
required(6, "timestamp_tz_col", Types.TimestampType.withZone()),
required(7, "str_col", Types.StringType.get()));
private static final PartitionSpec SPEC = PartitionSpec.unpartitioned();
private static final HadoopTables TABLES = new HadoopTables();

private Table table;
private DataFileSet dataFilesToRemove;
private DataFileSet dataFilesToAdd;

@Param({"50000", "100000", "500000", "1000000", "2000000"})
@Param({"50000", "100000", "500000"})
private int numFiles;

@Param({"5", "25", "50", "100"})
@Param({"1", "5", "50", "100"})
private int percentDataFilesRewritten;

/** Number of partitions to spread the data files over, or 0 for an unpartitioned table. */
@Param({"0", "1000"})
private int numPartitions;

/**
* Latency charged to every metadata file read and write, or 0 for no latency. Helps simulate
* object storage like latency even though using local disk.
*/
@Param({"0", "50"})
private int metadataFileLatencyMs;

/**
* Target manifest size, which controls how many manifests the files are spread over. Helps
* simulate larger tables with many manifests without generating extremely large table metadata.
*/
@Param({"65536", "8388608"})
private long manifestTargetSizeBytes;

@Setup
public void setupBenchmark() throws IOException {
initTable();
initFiles();

this.table =
new BaseTable(
new LatencyInjectingTableOperations(
((HasTableOperations) table).operations(), metadataFileLatencyMs),
TABLE_IDENT);
}

@TearDown
Expand All @@ -116,7 +145,27 @@ private void initTable() {

this.table =
TABLES.create(
SCHEMA, SPEC, ImmutableMap.of(TableProperties.FORMAT_VERSION, "3"), TABLE_IDENT);
SCHEMA,
spec(),
ImmutableMap.of(
TableProperties.FORMAT_VERSION,
"3",
TableProperties.MANIFEST_TARGET_SIZE_BYTES,
String.valueOf(manifestTargetSizeBytes)),
TABLE_IDENT);
}

private PartitionSpec spec() {
return numPartitions > 0
? PartitionSpec.builderFor(SCHEMA).identity("int_col").build()
: PartitionSpec.unpartitioned();
}

private int partitionValue(int ordinal) {
// Assign files to partitions in contiguous blocks to pack them by partition value.
// This allows partition pruning optimization, as manifests are clustered by partition value.
int filesPerPartition = (numFiles + numPartitions - 1) / numPartitions;
return ordinal / filesPerPartition;
}

private void dropTable() {
Expand All @@ -129,13 +178,14 @@ private void initFiles() throws IOException {
Map<String, DataFile> filesToReplace = Maps.newHashMapWithExpectedSize(numDataFilesToRewrite);
RowDelta rowDelta = table.newRowDelta();
for (int ordinal = 0; ordinal < numFiles; ordinal++) {
DataFile dataFile = generateDataFile();
DataFile dataFile = generateDataFile(ordinal);
rowDelta.addRows(dataFile);
DeleteFile deleteFile = FileGenerationUtil.generateDV(table, dataFile);
rowDelta.addDeletes(deleteFile);
if (numDataFilesToRewrite > 0) {
filesToReplace.put(dataFile.location(), dataFile);
DataFile pendingDataFile = generateDataFile(dataFile.recordCount());
// the replacement lands in the same partition as the file it replaces
DataFile pendingDataFile = generateDataFile(ordinal, dataFile.recordCount());
rowDelta.addRows(pendingDataFile);
pendingDataFiles.add(pendingDataFile);
numDataFilesToRewrite--;
Expand Down Expand Up @@ -163,15 +213,224 @@ private void initFiles() throws IOException {
this.dataFilesToAdd = DataFileSet.of(pendingDataFiles);
}

private DataFile generateDataFile() {
return generateDataFile(-1L);
/** Delegating operations whose FileIO charges a fixed latency for each open. */
private static class LatencyInjectingTableOperations implements TableOperations {
private final TableOperations delegate;
private final int latencyMs;

LatencyInjectingTableOperations(TableOperations delegate, int latencyMs) {
this.delegate = delegate;
this.latencyMs = latencyMs;
}

@Override
public TableMetadata current() {
return delegate.current();
}

@Override
public TableMetadata refresh() {
return delegate.refresh();
}

@Override
public void commit(TableMetadata base, TableMetadata metadata) {
delegate.commit(base, metadata);
}

@Override
public FileIO io() {
return new LatencyInjectingFileIO(delegate.io(), latencyMs);
}

@Override
public EncryptionManager encryption() {
return delegate.encryption();
}

@Override
public String metadataFileLocation(String fileName) {
return delegate.metadataFileLocation(fileName);
}

@Override
public LocationProvider locationProvider() {
return delegate.locationProvider();
}

@Override
public TableOperations temp(TableMetadata uncommittedMetadata) {
return new LatencyInjectingTableOperations(delegate.temp(uncommittedMetadata), latencyMs);
}

@Override
public long newSnapshotId() {
return delegate.newSnapshotId();
}

@Override
public boolean requireStrictCleanup() {
return delegate.requireStrictCleanup();
}
}

private static class LatencyInjectingFileIO implements FileIO {
private final FileIO delegate;
private final int latencyMs;

LatencyInjectingFileIO(FileIO delegate, int latencyMs) {
this.delegate = delegate;
this.latencyMs = latencyMs;
}

@Override
public InputFile newInputFile(String path) {
return new LatencyInjectingInputFile(delegate.newInputFile(path), latencyMs);
}

@Override
public InputFile newInputFile(String path, long length) {
return new LatencyInjectingInputFile(delegate.newInputFile(path, length), latencyMs);
}

@Override
public OutputFile newOutputFile(String path) {
return new LatencyInjectingOutputFile(delegate.newOutputFile(path), latencyMs);
}

@Override
public void deleteFile(String path) {
delegate.deleteFile(path);
}
}

private DataFile generateDataFile(long recordCount) {
/**
* An object store pays for a write when the object is completed rather than when the handle is
* created, so the latency is charged on close.
*/
private static class LatencyInjectingOutputFile implements OutputFile {
private final OutputFile delegate;
private final int latencyMs;

LatencyInjectingOutputFile(OutputFile delegate, int latencyMs) {
this.delegate = delegate;
this.latencyMs = latencyMs;
}

@Override
public PositionOutputStream create() {
return new LatencyInjectingPositionOutputStream(delegate.create(), latencyMs);
}

@Override
public PositionOutputStream createOrOverwrite() {
return new LatencyInjectingPositionOutputStream(delegate.createOrOverwrite(), latencyMs);
}

@Override
public String location() {
return delegate.location();
}

@Override
public InputFile toInputFile() {
return new LatencyInjectingInputFile(delegate.toInputFile(), latencyMs);
}
}

private static class LatencyInjectingPositionOutputStream extends PositionOutputStream {
private final PositionOutputStream delegate;
private final int latencyMs;

LatencyInjectingPositionOutputStream(PositionOutputStream delegate, int latencyMs) {
this.delegate = delegate;
this.latencyMs = latencyMs;
}

@Override
public long getPos() throws IOException {
return delegate.getPos();
}

@Override
public void write(int b) throws IOException {
delegate.write(b);
}

@Override
public void write(byte[] buffer, int offset, int length) throws IOException {
delegate.write(buffer, offset, length);
}

@Override
public void flush() throws IOException {
delegate.flush();
}

@Override
public void close() throws IOException {
sleep(latencyMs);
delegate.close();
}
}

private static class LatencyInjectingInputFile implements InputFile {
private final InputFile delegate;
private final int latencyMs;

LatencyInjectingInputFile(InputFile delegate, int latencyMs) {
this.delegate = delegate;
this.latencyMs = latencyMs;
}

@Override
public long getLength() {
return delegate.getLength();
}

@Override
public SeekableInputStream newStream() {
sleep(latencyMs);
return delegate.newStream();
}

@Override
public String location() {
return delegate.location();
}

@Override
public boolean exists() {
return delegate.exists();
}
}

private static void sleep(int latencyMs) {
if (latencyMs <= 0) {
return;
}

try {
Thread.sleep(latencyMs);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("Interrupted while simulating latency", e);
}
}

private DataFile generateDataFile(int ordinal) {
return generateDataFile(ordinal, -1L);
}

private DataFile generateDataFile(int ordinal, long recordCount) {
Schema schema = table.schema();
PartitionSpec spec = table.spec();
LocationProvider locations = table.locationProvider();
String path = locations.newDataLocation(spec, null, FileGenerationUtil.generateFileName());
String fileName = FileGenerationUtil.generateFileName();
String path =
numPartitions > 0
? locations.newDataLocation(fileName)
: locations.newDataLocation(spec, null, fileName);
long fileSize = ThreadLocalRandom.current().nextLong(50_000L);
MetricsConfig metricsConfig = MetricsConfig.forTable(table);
Metrics metrics =
Expand All @@ -187,11 +446,17 @@ private DataFile generateDataFile(long recordCount) {
metrics.nanValueCounts());
}

return DataFiles.builder(spec)
.withPath(path)
.withFileSizeInBytes(fileSize)
.withFormat(FileFormat.PARQUET)
.withMetrics(metrics)
.build();
DataFiles.Builder builder =
DataFiles.builder(spec)
.withPath(path)
.withFileSizeInBytes(fileSize)
.withFormat(FileFormat.PARQUET)
.withMetrics(metrics);

if (numPartitions > 0) {
builder.withPartitionPath("int_col=" + partitionValue(ordinal));
}

return builder.build();
}
}
Loading
Loading