Skip to content

Core: Replace Set with Bitmap to make delete filtering simpler and faster #3535

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
rdblue merged 2 commits into from
Jan 9, 2022
Merged

Core: Replace Set with Bitmap to make delete filtering simpler and faster #3535

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9edb689
Core: Replace Set<Long> with Bitmap to reduce memory footprint.
flyrain Dec 17, 2021
a52e6d4
Resolve the comments.
flyrain Dec 17, 2021
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
7 changes: 6 additions & 1 deletion core/src/main/java/org/apache/iceberg/deletes/BitmapPositionDeleteIndex.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,12 @@ public void delete(long posStart, long posEnd) {
}

@Override
public boolean deleted(long position) {
public boolean isDeleted(long position) {
Copy link
Contributor

@jackye1995 jackye1995 Dec 17, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why changing the name?

Copy link
Contributor Author

@flyrain flyrain Dec 17, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is suggested by @rdblue. Make sense to me as well. It is less ambiguous.

return roaring64Bitmap.contains(position);
}

@Override
public boolean isEmpty() {
return roaring64Bitmap.isEmpty();
}
}
38 changes: 8 additions & 30 deletions core/src/main/java/org/apache/iceberg/deletes/Deletes.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@
import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.List;
import java.util.Set;
import java.util.function.Function;
import org.apache.iceberg.Accessor;
import org.apache.iceberg.MetadataColumns;
Expand All @@ -35,7 +34,6 @@
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.Filter;
import org.apache.iceberg.util.SortedMerge;
Expand Down Expand Up @@ -66,7 +64,7 @@ public static <T> CloseableIterable<T> filter(CloseableIterable<T> rows, Functio
}

public static <T> CloseableIterable<T> filter(CloseableIterable<T> rows, Function<T, Long> rowToPosition,
Set<Long> deleteSet) {
PositionDeleteIndex deleteSet) {
if (deleteSet.isEmpty()) {
return rows;
}
Expand All @@ -85,35 +83,15 @@ public static StructLikeSet toEqualitySet(CloseableIterable<StructLike> eqDelete
}
}

public static Set<Long> toPositionSet(CharSequence dataLocation, CloseableIterable<? extends StructLike> deleteFile) {
return toPositionSet(dataLocation, ImmutableList.of(deleteFile));
}

public static <T extends StructLike> Set<Long> toPositionSet(CharSequence dataLocation,
List<CloseableIterable<T>> deleteFiles) {
DataFileFilter<T> locationFilter = new DataFileFilter<>(dataLocation);
List<CloseableIterable<Long>> positions = Lists.transform(deleteFiles, deletes ->
CloseableIterable.transform(locationFilter.filter(deletes), row -> (Long) POSITION_ACCESSOR.get(row)));
return toPositionSet(CloseableIterable.concat(positions));
}

public static Set<Long> toPositionSet(CloseableIterable<Long> posDeletes) {
try (CloseableIterable<Long> deletes = posDeletes) {
return Sets.newHashSet(deletes);
} catch (IOException e) {
throw new UncheckedIOException("Failed to close position delete source", e);
}
}

public static <T extends StructLike> PositionDeleteIndex toPositionBitmap(CharSequence dataLocation,
List<CloseableIterable<T>> deleteFiles) {
public static <T extends StructLike> PositionDeleteIndex toPositionIndex(CharSequence dataLocation,
List<CloseableIterable<T>> deleteFiles) {
DataFileFilter<T> locationFilter = new DataFileFilter<>(dataLocation);
List<CloseableIterable<Long>> positions = Lists.transform(deleteFiles, deletes ->
CloseableIterable.transform(locationFilter.filter(deletes), row -> (Long) POSITION_ACCESSOR.get(row)));
return toPositionBitmap(CloseableIterable.concat(positions));
return toPositionIndex(CloseableIterable.concat(positions));
}

public static PositionDeleteIndex toPositionBitmap(CloseableIterable<Long> posDeletes) {
public static PositionDeleteIndex toPositionIndex(CloseableIterable<Long> posDeletes) {
try (CloseableIterable<Long> deletes = posDeletes) {
PositionDeleteIndex positionDeleteIndex = new BitmapPositionDeleteIndex();
deletes.forEach(positionDeleteIndex::delete);
Expand Down Expand Up @@ -161,16 +139,16 @@ protected boolean shouldKeep(T row) {

private static class PositionSetDeleteFilter<T> extends Filter<T> {
private final Function<T, Long> rowToPosition;
private final Set<Long> deleteSet;
private final PositionDeleteIndex deleteSet;

private PositionSetDeleteFilter(Function<T, Long> rowToPosition, Set<Long> deleteSet) {
private PositionSetDeleteFilter(Function<T, Long> rowToPosition, PositionDeleteIndex deleteSet) {
this.rowToPosition = rowToPosition;
this.deleteSet = deleteSet;
}

@Override
protected boolean shouldKeep(T row) {
return !deleteSet.contains(rowToPosition.apply(row));
return !deleteSet.isDeleted(rowToPosition.apply(row));
}
}

Expand Down
7 changes: 6 additions & 1 deletion core/src/main/java/org/apache/iceberg/deletes/PositionDeleteIndex.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,5 +38,10 @@ public interface PositionDeleteIndex {
* @param position deleted row position
* @return whether the position is deleted
*/
boolean deleted(long position);
boolean isDeleted(long position);

/**
* Returns true if this collection contains no element.
*/
boolean isEmpty();
}
4 changes: 2 additions & 2 deletions core/src/test/java/org/apache/iceberg/deletes/TestPositionFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -218,7 +218,7 @@ public void testPositionSetRowFilter() {

CloseableIterable<StructLike> actual = Deletes.filter(
rows, row -> row.get(0, Long.class),
Deletes.toPositionSet(deletes));
Deletes.toPositionIndex(deletes));
Assert.assertEquals("Filter should produce expected rows",
Lists.newArrayList(1L, 2L, 5L, 6L, 8L),
Lists.newArrayList(Iterables.transform(actual, row -> row.get(0, Long.class))));
Expand Down Expand Up @@ -256,7 +256,7 @@ public void testCombinedPositionSetRowFilter() {

CloseableIterable<StructLike> actual = Deletes.filter(
rows, row -> row.get(0, Long.class),
Deletes.toPositionSet("file_a.avro", ImmutableList.of(positionDeletes1, positionDeletes2)));
Deletes.toPositionIndex("file_a.avro", ImmutableList.of(positionDeletes1, positionDeletes2)));

Assert.assertEquals("Filter should produce expected rows",
Lists.newArrayList(1L, 2L, 5L, 6L, 8L),
Expand Down
6 changes: 2 additions & 4 deletions data/src/main/java/org/apache/iceberg/data/DeleteFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -214,7 +214,7 @@ public PositionDeleteIndex deletedRowPositions() {

if (deleteRowPositions == null) {
List<CloseableIterable<Record>> deletes = Lists.transform(posDeletes, this::openPosDeletes);
deleteRowPositions = Deletes.toPositionBitmap(dataFile.path(), deletes);
deleteRowPositions = Deletes.toPositionIndex(dataFile.path(), deletes);
}
return deleteRowPositions;
}
Expand All @@ -228,9 +228,7 @@ private CloseableIterable<T> applyPosDeletes(CloseableIterable<T> records) {

// if there are fewer deletes than a reasonable number to keep in memory, use a set
if (posDeletes.stream().mapToLong(DeleteFile::recordCount).sum() < setFilterThreshold) {
return Deletes.filter(
records, this::pos,
Deletes.toPositionSet(dataFile.path(), CloseableIterable.concat(deletes)));
return Deletes.filter(records, this::pos, Deletes.toPositionIndex(dataFile.path(), deletes));
}

return Deletes.streamingFilter(records, this::pos, Deletes.deletePositions(dataFile.path(), deletes));
Expand Down
2 changes: 1 addition & 1 deletion ...3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,7 +159,7 @@ Pair<int[], Integer> buildPosDelRowIdMapping(PositionDeleteIndex deletedRowPosit
int originalRowId = 0;
int currentRowId = 0;
while (originalRowId < numRowsToRead) {
if (!deletedRowPositions.deleted(originalRowId + rowStartPosInBatch)) {
if (!deletedRowPositions.isDeleted(originalRowId + rowStartPosInBatch)) {
posDelRowIdMapping[currentRowId] = originalRowId;
currentRowId++;
}
Expand Down
7 changes: 6 additions & 1 deletion ...park/src/test/java/org/apache/iceberg/spark/data/TestSparkParquetReadMetadataColumns.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -217,9 +217,14 @@ public void delete(long posStart, long posEnd) {
}

@Override
public boolean deleted(long position) {
public boolean isDeleted(long position) {
return deleteIndex.contains(position);
}

@Override
public boolean isEmpty() {
return deleteIndex.isEmpty();
}
}

@Test
Expand Down