Read deleted rows with metadata column IS_DELETED #4683

flyrain · 2022-05-02T23:03:42Z

Per the discussion in #4539, created this new PR. With the change in #2538 and this PR, we can read deleted rows from row-level deletes.
~~Please note that there is no IS_DELETED column filter push down at this moment. This will be done as a followup PR.~~ Think a bit more, we may not need pushdown, especially for non-vectorized read. It shouldn't have any difference between Spark filters out rows or Iceberg filters out rows. We iterate through all rows anyway.

cc @aokolnychyi @RussellSpitzer @chenjunjiedada @stevenzwu @Reo-LEI @hameizi @singhpk234 @rajarshisarkar @kbendick @rdblue

arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorHolder.java

core/src/main/java/org/apache/iceberg/deletes/Deletes.java

data/src/main/java/org/apache/iceberg/data/DeleteFilter.java

spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/RowDataReader.java

spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderDeletes.java

core/src/main/java/org/apache/iceberg/deletes/Deletes.java

data/src/main/java/org/apache/iceberg/data/DeleteFilter.java

flyrain · 2022-05-04T21:47:44Z

Do we still need class EqualityDeleteRowReader? Its functionality can be archived by a filter on the IS_DELETED column. cc @chenjunjiedada

stevenzwu · 2022-05-04T22:27:47Z

api/src/main/java/org/apache/iceberg/Schema.java

+   * @param fieldId a column id in this schema
+   * @return the index of the field in the schema, or -1 if one wasn't found
+   */
+  public int idToIndex(Integer fieldId) {


idToPosition? position might be more clear.

Also we are using Integer arg type, which means it can be null. Probably following the pattern from the idToAlias above. But I am wondering if the fieldId arg should be primitive int type. If we stay with Integer type, we probably need null check here.

Also, should the return type be Integer and nullable just to be consistent with other methods in this class?

Thanks for the name suggestion. There are always performance concerns about non-primitive type. I'm kind of OK with both primitive type and non-primitive type here. In terms of consistency, these two public methods also return int. I will leave the return type to int unless we have a strong reason to use Integer.

public int schemaId() public int highestFieldId()

Even this place uses int, I'd prefer to use int for the parameter as well. It is unnecessary to do this boxing and unboxing.

iceberg/api/src/main/java/org/apache/iceberg/types/Types.java

Line 479 in 44c1d00

public int fieldId() {

Even this place uses int, I'd prefer to use int for the parameter as well. It is unnecessary to do this boxing and unboxing.

iceberg/api/src/main/java/org/apache/iceberg/types/Types.java

Line 479 in 44c1d00

public int fieldId() {

That is different. For NestedField, fieldId will never be null. hence it returns primitive int, which makes sense. But here, we may not find the field with the provided id, we are returning a special value -1. If we follow the style of other find methods in the Schema class, we can see aliasToId returns null if no matching field is found.

Thanks for the comment. With @szehon-ho's suggestion #4683 (comment), I removed the the new method.

stevenzwu · 2022-05-05T00:13:15Z

data/src/main/java/org/apache/iceberg/data/DeleteFilter.java

@@ -94,6 +95,12 @@ protected DeleteFilter(String filePath, List<DeleteFile> deletes, Schema tableSc
    this.eqDeletes = eqDeleteBuilder.build();
    this.requiredSchema = fileProjection(tableSchema, requestedSchema, posDeletes, eqDeletes);
    this.posAccessor = requiredSchema.accessorForField(MetadataColumns.ROW_POSITION.fieldId());
+    this.hasColumnIsDeleted = requestedSchema.findField(MetadataColumns.IS_DELETED.fieldId()) != null;


nit: I saw a lot of places in Iceberg just use null to indicate the condition. should we use null for columnIsDeletedIndex too?

A boolean variable should be expressive.

data/src/main/java/org/apache/iceberg/data/DeleteFilter.java

stevenzwu · 2022-05-05T00:16:33Z

core/src/main/java/org/apache/iceberg/deletes/Deletes.java

@@ -227,6 +237,39 @@ public void close() {
    }
  }

+  private static class PositionStreamDeleteMarker<T> extends PositionStreamDeleteFilter<T> {


as mentioned in the draft PR, this is not actually a filter as it always returns true. It is only to use the iteration part of the filter. Maybe use CloseableIterable#transform instead for iteration?

Yeah, I've replaced filter with transform for all places except this one. This is trickier to change. Let me think a bit more.

Nit: Do you think it will be cleaner to have most of the logic at:

abstract class PositionStreamDeleteIterable { CloseableIterator createPosDeleteIterator(CloseableIterator<T> items); }

and have the two concrete subclass (PositionStreamDeleteMarker and PositionStreamDeleteFilter) extend it?

I think having the Marker extend the Filter still seems a bit strange, though the logic is correctly refactored now.

chenjunjiedada · 2022-05-05T02:08:44Z

@flyrain , I agree the implementation of EqualityDeleteRowReader can be replaced, but we are using the class for rewrite equality deletes. If we have the utilities to produce only equality deleted rows or position deleted rows, I'm OK to delete them.

flyrain · 2022-05-05T19:08:13Z

@flyrain , I agree the implementation of EqualityDeleteRowReader can be replaced, but we are using the class for rewrite equality deletes. If we have the utilities to produce only equality deleted rows or position deleted rows, I'm OK to delete them.

We don't plan for that utilities in this PR. Let's not change anything about EqualityDeleteRowReader. Thanks for the input.

core/src/main/java/org/apache/iceberg/deletes/Deletes.java

data/src/main/java/org/apache/iceberg/data/DeleteFilter.java

stevenzwu · 2022-05-06T23:26:22Z

core/src/main/java/org/apache/iceberg/deletes/Deletes.java


-    PositionSetDeleteFilter<T> filter = new PositionSetDeleteFilter<>(rowToPosition, deleteSet);
+  public static <T> CloseableIterable<T> filter(CloseableIterable<T> rows, Predicate<T> shouldKeep) {


I like the new filter method API. only concern is for compatibility. this is a public static method from a public class. Probably very few users call this API directly. if we want to be safe, we can keep the old API and mark it as deprecated.

That's a good point. I had the same concern. However, we've discussed what should be considered as APIs in one of the community sync. we agreed that only consider the public things in the API module as APIs. cc @rdblue. With that, it should be OK to change this public method?

...k/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java

...2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/IcebergArrowColumnVector.java

spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderDeletes.java

szehon-ho

Looks good to me

aokolnychyi · 2022-05-26T15:59:27Z

Let me take another look too. Sorry for the delay.

spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/RowDataReader.java

aokolnychyi · 2022-05-26T20:20:08Z

core/src/main/java/org/apache/iceberg/deletes/Deletes.java

-            keep = false;
-          }
+    @Override
+    protected CloseableIterator createPosDeleteIterator(CloseableIterator<T> items) {


nit: raw type

I'd say it's a bit more than a nit! Iceberg PRs should always use the type system.

Fixed it in a new commit.

aokolnychyi · 2022-05-26T20:20:38Z

core/src/main/java/org/apache/iceberg/deletes/Deletes.java

+      return isDeleted;
+    }
+
+    protected abstract CloseableIterator createPosDeleteIterator(CloseableIterator<T> items);


nit: raw type

aokolnychyi · 2022-05-26T20:21:11Z

core/src/main/java/org/apache/iceberg/deletes/Deletes.java

-        } catch (IOException e) {
-          throw new UncheckedIOException("Failed to close delete positions iterator", e);
+    @Override
+    protected CloseableIterator createPosDeleteIterator(CloseableIterator<T> items) {


nit: raw type

aokolnychyi · 2022-05-26T21:33:27Z

core/src/main/java/org/apache/iceberg/deletes/Deletes.java

+      return isDeleted;
+    }
+
+    protected abstract CloseableIterator createPosDeleteIterator(CloseableIterator<T> items);


Do we really create a position delete iterator? Don't we iterate over records here?

We do iterate the pos delete records, for example, line 191, this.nextDeletePos = deletePosIterator.next();.

We do iterate over position deletes but I think this iterator is for remaining data records, no?

Discussed offline with @aokolnychyi, I've changed the name to applyDelete

core/src/main/java/org/apache/iceberg/deletes/Deletes.java

aokolnychyi · 2022-05-26T22:00:58Z

data/src/main/java/org/apache/iceberg/data/DeleteFilter.java

@@ -67,6 +66,8 @@
  private final List<DeleteFile> eqDeletes;
  private final Schema requiredSchema;
  private final Accessor<StructLike> posAccessor;
+  private final boolean hasColumnIsDeleted;


nit: hasIsDeletedColumn?

aokolnychyi · 2022-05-26T22:01:14Z

data/src/main/java/org/apache/iceberg/data/DeleteFilter.java

@@ -67,6 +66,8 @@
  private final List<DeleteFile> eqDeletes;
  private final Schema requiredSchema;
  private final Accessor<StructLike> posAccessor;
+  private final boolean hasColumnIsDeleted;
+  private final int columnIsDeletedPosition;


nit: isDeletedColumnPosition?

aokolnychyi · 2022-05-26T22:01:33Z

data/src/main/java/org/apache/iceberg/data/DeleteFilter.java

+    this.columnIsDeletedPosition = requiredSchema.columns().indexOf(MetadataColumns.IS_DELETED);
+  }
+
+  protected int columnIsDeletedPosition() {


nit: isDeletedColumnPosition()?

aokolnychyi · 2022-05-26T22:11:26Z

data/src/main/java/org/apache/iceberg/data/DeleteFilter.java

+        Deletes.streamingFilter(records, this::pos, Deletes.deletePositions(filePath, deletes));
+  }
+
+  private CloseableIterable<T> createDeleteIterable(CloseableIterable<T> records, Predicate<T> isDeleted) {


Is there a better name? Is it an iterable of remaining rows?

aokolnychyi · 2022-05-26T22:18:28Z

This looks close to me. I'd switch to an explicit call to set the _deleted value to avoid any assumptions about default values.

RussellSpitzer · 2022-05-26T23:05:25Z

core/src/main/java/org/apache/iceberg/deletes/Deletes.java

-        this.deletePosIterator = deletePositions;
+      // consume delete positions until the next is past the current position
+      boolean isDeleted = currentPos == nextDeletePos;
+      while (deletePosIterator.hasNext() && nextDeletePos <= currentPos) {
        this.nextDeletePos = deletePosIterator.next();


This is now the only place in the code which refers to it as "this.nextDeletePos"

RussellSpitzer · 2022-05-27T00:06:29Z

core/src/main/java/org/apache/iceberg/deletes/Deletes.java

-      protected PositionFilterIterator(CloseableIterator<T> items, CloseableIterator<Long> deletePositions) {
-        super(items);
-        this.deletePosIterator = deletePositions;
+      // consume delete positions until the next is past the current position


Not sure if this is any simpler but we can remove one if statement, not sure if this is more clear.

// Consume nextDeletePos till past currentPos, if currentPos equals any consumed nextDeletePos the current row has been deleted boolean isDeleted = currentPos == nextDeletePos; while (deletePosIterator.hasNext() && nextDeletePos <= currentPos) { this.nextDeletePos = deletePosIterator.next(); isDeleted |= currentPos == nextDeletePos } return isDeleted; }

I'm OK with either one. I kind of think if is easier to read. If we will make it simpler, here is another way, which we don't have to check on isDeleted. What do you think?

while (deletePosIterator.hasNext() && nextDeletePos <= currentPos) { this.nextDeletePos = deletePosIterator.next(); if (currentPos == nextDeletePos) { isDeleted = true; } }

Let's keep it as-is to avoid last-minute changes in this tricky place.

flyrain · 2022-05-27T07:52:23Z

spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderDeletes.java

+  }
+
+  @Test
+  public void testIsDeletedColumnWithoutDeleteFile() {


Added a test to project is_deleted column when there is no delete file.

aokolnychyi · 2022-05-27T21:03:27Z

Thanks, @flyrain! Great to have this done. Thanks everyone who reviewed!

flyrain · 2022-05-27T21:12:05Z

Thank @aokolnychyi! Thanks everyone for the review.

…e#4683)

github-actions bot added arrow core data flink spark labels May 2, 2022

singhpk234 reviewed May 3, 2022

View reviewed changes

chenjunjiedada reviewed May 4, 2022

View reviewed changes

core/src/main/java/org/apache/iceberg/deletes/Deletes.java Outdated Show resolved Hide resolved

chenjunjiedada reviewed May 4, 2022

View reviewed changes

data/src/main/java/org/apache/iceberg/data/DeleteFilter.java Outdated Show resolved Hide resolved

github-actions bot added the API label May 4, 2022

flyrain added 5 commits May 4, 2022 14:37

The init commit of is_delete column

3e45982

Resolve the compile error.

df17bed

Resolve the compile error.

653c893

Resolve the comments

0f812b5

Fixed the compile error for flink1.15

5b3ecc7

flyrain force-pushed the deleteColumn branch from 20cb74f to 5b3ecc7 Compare May 4, 2022 21:41

stevenzwu reviewed May 4, 2022

View reviewed changes

stevenzwu reviewed May 5, 2022

View reviewed changes

data/src/main/java/org/apache/iceberg/data/DeleteFilter.java Outdated Show resolved Hide resolved

stevenzwu reviewed May 5, 2022

View reviewed changes

Resolve comments

f578762

stevenzwu reviewed May 6, 2022

View reviewed changes

core/src/main/java/org/apache/iceberg/deletes/Deletes.java Outdated Show resolved Hide resolved

stevenzwu reviewed May 6, 2022

View reviewed changes

data/src/main/java/org/apache/iceberg/data/DeleteFilter.java Outdated Show resolved Hide resolved

stevenzwu reviewed May 6, 2022

View reviewed changes

stevenzwu reviewed May 7, 2022

View reviewed changes

...k/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java Outdated Show resolved Hide resolved

stevenzwu reviewed May 7, 2022

View reviewed changes

...2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/IcebergArrowColumnVector.java Outdated Show resolved Hide resolved

stevenzwu reviewed May 7, 2022

View reviewed changes

spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderDeletes.java Outdated Show resolved Hide resolved

stevenzwu reviewed May 7, 2022

View reviewed changes

spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderDeletes.java Outdated Show resolved Hide resolved

szehon-ho approved these changes May 25, 2022

View reviewed changes

aokolnychyi reviewed May 26, 2022

View reviewed changes

spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/RowDataReader.java Show resolved Hide resolved

aokolnychyi reviewed May 26, 2022

View reviewed changes

core/src/main/java/org/apache/iceberg/deletes/Deletes.java Show resolved Hide resolved

aokolnychyi reviewed May 26, 2022

View reviewed changes

RussellSpitzer reviewed May 26, 2022

View reviewed changes

RussellSpitzer reviewed May 27, 2022

View reviewed changes

Resolve comments.

ebb5b7b

flyrain commented May 27, 2022

View reviewed changes

aokolnychyi approved these changes May 27, 2022

View reviewed changes

aokolnychyi merged commit 534f3c9 into apache:master May 27, 2022

This was referenced May 28, 2022

Core: Support _deleted metadata column in vectorized read #4888

Merged

Spark: Add custom metric for number of deletes applied by a SparkScan #4588

Merged

Initial-neko pushed a commit to Initial-neko/iceberg that referenced this pull request Jul 18, 2022

Core: Support _deleted metadata column in non-vectorized reads (apach…

250c9b8

…e#4683)

Initial-neko pushed a commit to Initial-neko/iceberg that referenced this pull request Jul 18, 2022

Core: Support _deleted metadata column in non-vectorized reads (apach…

a8d72bf

…e#4683)

Initial-neko pushed a commit to Initial-neko/iceberg that referenced this pull request Jul 20, 2022

Core: Support _deleted metadata column in non-vectorized reads (apach…

f8044d7

…e#4683)

Initial-neko pushed a commit to Initial-neko/iceberg that referenced this pull request Jul 25, 2022

Core: Support _deleted metadata column in non-vectorized reads (apach…

e762506

…e#4683)

hililiwei added a commit to hililiwei/iceberg that referenced this pull request Aug 15, 2022

Spark 3.1:Port apache#4683 to Spark 3.1

27ceaf7

hililiwei mentioned this pull request Aug 15, 2022

Spark 3.1:Port #4683 to Spark 3.1 #5524

Closed

hililiwei added a commit to hililiwei/iceberg that referenced this pull request Aug 15, 2022

Spark 3.1:Port apache#4683 to Spark 3.1

c91db32


		PositionSetDeleteFilter<T> filter = new PositionSetDeleteFilter<>(rowToPosition, deleteSet);
		public static <T> CloseableIterable<T> filter(CloseableIterable<T> rows, Predicate<T> shouldKeep) {

Read deleted rows with metadata column IS_DELETED #4683

Read deleted rows with metadata column IS_DELETED #4683

Conversation

flyrain commented May 2, 2022 • edited Loading

flyrain commented May 4, 2022 • edited Loading

stevenzwu May 4, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

flyrain May 5, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chenjunjiedada commented May 5, 2022

flyrain commented May 5, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

szehon-ho left a comment

Choose a reason for hiding this comment

aokolnychyi commented May 26, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi May 26, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi May 26, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi commented May 26, 2022

Choose a reason for hiding this comment

RussellSpitzer May 27, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

flyrain May 27, 2022 • edited Loading

Choose a reason for hiding this comment

aokolnychyi commented May 27, 2022

flyrain commented May 27, 2022

flyrain commented May 2, 2022 •

edited

Loading

flyrain commented May 4, 2022 •

edited

Loading

stevenzwu May 4, 2022 •

edited

Loading

flyrain May 5, 2022 •

edited

Loading

aokolnychyi May 26, 2022 •

edited

Loading

aokolnychyi May 26, 2022 •

edited

Loading

RussellSpitzer May 27, 2022 •

edited

Loading

flyrain May 27, 2022 •

edited

Loading