Change Data Capture(CDC)[Draft]

api/src/main/java/org/apache/iceberg/actions/ActionsProvider.java

@@ -74,4 +74,11 @@ default ExpireSnapshots expireSnapshots(Table table) {
   default DeleteReachableFiles deleteReachableFiles(String metadataLocation) {
     throw new UnsupportedOperationException(this.getClass().getName() + " does not implement deleteReachableFiles");
   }
+
+  /**
+   * Instantiates an action to generate CDC records.
+   */
+  default Cdc generateCdcRecords(Table table) {
+    throw new UnsupportedOperationException(this.getClass().getName() + " does not implement generateCdcRecords");
+  }
 }

api/src/main/java/org/apache/iceberg/actions/Cdc.java

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.actions;
+
+public interface Cdc extends Action<Cdc, Cdc.Result> {
+  /**
+   * Emit changed data set by a snapshot id.
+   *
+   * @param snapshotId id of the snapshot to generate changed data
+   * @return this for method chaining
+   */
+  Cdc useSnapshot(long snapshotId);
+
+  /**
+   * Emit changed data set by a range of snapshots
+   *
+   * @param fromSnapshotId id of the first snapshot
+   * @param toSnapshotId id of the last snapshot
+   * @return this for method chaining
+   */
+  Cdc between(long fromSnapshotId, long toSnapshotId);
+
+  /**
+   * The action result that contains a dataset of changed rows.
+   */
+  interface Result {
+    /**
+     * Returns CDC records.
+     */
+    Object cdcRecords();
+  }
+}

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

@@ -104,6 +104,10 @@ public static <T> VectorHolder constantHolder(int numRows, T constantValue) {
     return new ConstantVectorHolder(numRows, constantValue);
   }
 
+  public static <T> VectorHolder deleteMetaColumnHolder(int numRows) {
+    return new DeletedVectorHolder(numRows);
+  }
+
   public static VectorHolder dummyHolder(int numRows) {
     return new ConstantVectorHolder(numRows);
   }
@@ -146,4 +150,17 @@ public PositionVectorHolder(FieldVector vector, Type type, NullabilityHolder nul
     }
   }
 
+  public static class DeletedVectorHolder extends VectorHolder {
+    private final int numRows;
+
+    public DeletedVectorHolder(int numRows) {
+      this.numRows = numRows;
+    }
+
+    @Override
+    public int numValues() {
+      return numRows;
+    }
+  }
+
 }

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

@@ -517,5 +517,32 @@ public void setBatchSize(int batchSize) {
     }
   }
 
+  /**
+   * A Dummy Vector Reader which doesn't actually read files, instead it returns a dummy
+   * VectorHolder which indicates whether the row is deleted.
+   */
+  public static class DeletedVectorReader extends VectorizedArrowReader {
+    public DeletedVectorReader() {
+    }
+
+    @Override
+    public VectorHolder read(VectorHolder reuse, int numValsToRead) {
+      return VectorHolder.deleteMetaColumnHolder(numValsToRead);
+    }
+
+    @Override
+    public void setRowGroupInfo(PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata, long rowPosition) {
+    }
+
+    @Override
+    public String toString() {
+      return "DeletedVectorReader";
+    }
+
+    @Override
+    public void setBatchSize(int batchSize) {
+    }
+  }
+
 }
 

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

@@ -91,7 +91,7 @@ public VectorizedReader<?> message(
           reorderedFields.add(VectorizedArrowReader.positions());
         }
       } else if (id == MetadataColumns.IS_DELETED.fieldId()) {
-        reorderedFields.add(new VectorizedArrowReader.ConstantVectorReader<>(false));
+        reorderedFields.add(new VectorizedArrowReader.DeletedVectorReader());
       } else if (reader != null) {
         reorderedFields.add(reader);
       } else {

core/src/main/java/org/apache/iceberg/BaseFileScanTask.java

@@ -47,6 +47,10 @@ public BaseFileScanTask(DataFile file, DeleteFile[] deletes, String schemaString
     this.residuals = residuals;
   }
 
+  public BaseFileScanTask cloneWithoutDeletes() {
+    return new BaseFileScanTask(file, new DeleteFile[0], schemaString, specString, residuals);
+  }
+
   @Override
   public DataFile file() {
     return file;

core/src/main/java/org/apache/iceberg/ManifestGroup.java

@@ -41,7 +41,7 @@
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.ParallelIterable;
 
-class ManifestGroup {
+public class ManifestGroup {
   private static final Types.StructType EMPTY_STRUCT = Types.StructType.of();
 
   private final FileIO io;
@@ -55,6 +55,8 @@ class ManifestGroup {
   private Expression partitionFilter;
   private boolean ignoreDeleted;
   private boolean ignoreExisting;
+  private boolean ignoreAdded;
+  private boolean onlyWithDeletes;
   private boolean ignoreResiduals;
   private List<String> columns;
   private boolean caseSensitive;
@@ -66,7 +68,7 @@ class ManifestGroup {
         Iterables.filter(manifests, manifest -> manifest.content() == ManifestContent.DELETES));
   }
 
-  ManifestGroup(FileIO io, Iterable<ManifestFile> dataManifests, Iterable<ManifestFile> deleteManifests) {
+  public ManifestGroup(FileIO io, Iterable<ManifestFile> dataManifests, Iterable<ManifestFile> deleteManifests) {
     this.io = io;
     this.dataManifests = Sets.newHashSet(dataManifests);
     this.deleteIndexBuilder = DeleteFileIndex.builderFor(io, deleteManifests);
@@ -75,20 +77,22 @@ class ManifestGroup {
     this.partitionFilter = Expressions.alwaysTrue();
     this.ignoreDeleted = false;
     this.ignoreExisting = false;
+    this.ignoreAdded = false;
+    this.onlyWithDeletes = false;
     this.ignoreResiduals = false;
     this.columns = ManifestReader.ALL_COLUMNS;
     this.caseSensitive = true;
     this.manifestPredicate = m -> true;
     this.manifestEntryPredicate = e -> true;
   }
 
-  ManifestGroup specsById(Map<Integer, PartitionSpec> newSpecsById) {
+  public ManifestGroup specsById(Map<Integer, PartitionSpec> newSpecsById) {
     this.specsById = newSpecsById;
     deleteIndexBuilder.specsById(newSpecsById);
     return this;
   }
 
-  ManifestGroup filterData(Expression newDataFilter) {
+  public ManifestGroup filterData(Expression newDataFilter) {
     this.dataFilter = Expressions.and(dataFilter, newDataFilter);
     deleteIndexBuilder.filterData(newDataFilter);
     return this;
@@ -125,6 +129,16 @@ ManifestGroup ignoreExisting() {
     return this;
   }
 
+  public ManifestGroup ignoreAdded() {
+    this.ignoreAdded = true;
+    return this;
+  }
+
+  public ManifestGroup onlyWithDeletes() {
+    this.onlyWithDeletes = true;
+    return this;
+  }
+
   ManifestGroup ignoreResiduals() {
     this.ignoreResiduals = true;
     return this;
@@ -180,7 +194,7 @@ public CloseableIterable<FileScanTask> planFiles() {
         return CloseableIterable.transform(entries, e -> new BaseFileScanTask(
             e.file().copy(), deleteFiles.forEntry(e), schemaString, specString, residuals));
       }
-    });
+    }, deleteFiles);
 
     if (executorService != null) {
       return new ParallelIterable<>(tasks, executorService);
@@ -198,11 +212,13 @@ public CloseableIterable<FileScanTask> planFiles() {
    * @return a CloseableIterable of manifest entries.
    */
   public CloseableIterable<ManifestEntry<DataFile>> entries() {
-    return CloseableIterable.concat(entries((manifest, entries) -> entries));
+    return CloseableIterable.concat(entries((manifest, entries) -> entries, null));
   }
 
+  @SuppressWarnings({"unchecked", "checkstyle:CyclomaticComplexity"})
   private <T> Iterable<CloseableIterable<T>> entries(
-      BiFunction<ManifestFile, CloseableIterable<ManifestEntry<DataFile>>, CloseableIterable<T>> entryFn) {
+      BiFunction<ManifestFile, CloseableIterable<ManifestEntry<DataFile>>, CloseableIterable<T>> entryFn,
+      DeleteFileIndex deleteFiles) {
     LoadingCache<Integer, ManifestEvaluator> evalCache = specsById == null ?
         null : Caffeine.newBuilder().build(specId -> {
           PartitionSpec spec = specsById.get(specId);
@@ -237,6 +253,12 @@ private <T> Iterable<CloseableIterable<T>> entries(
           manifest -> manifest.hasAddedFiles() || manifest.hasDeletedFiles());
     }
 
+    if (ignoreAdded) {
+      // only scan manifests that have entries other than added
+      matchingManifests = Iterables.filter(matchingManifests,
+          manifest -> manifest.hasExistingFiles() || manifest.hasDeletedFiles());
+    }
+
     matchingManifests = Iterables.filter(matchingManifests, manifestPredicate::test);
 
     return Iterables.transform(
@@ -258,6 +280,14 @@ private <T> Iterable<CloseableIterable<T>> entries(
                 entry -> entry.status() != ManifestEntry.Status.EXISTING);
           }
 
+          if (ignoreAdded) {
+            entries = CloseableIterable.filter(entries, entry -> entry.status() != ManifestEntry.Status.ADDED);
+          }
+
+          if (onlyWithDeletes && deleteFiles != null) {
+            entries = CloseableIterable.filter(entries, entry -> deleteFiles.forEntry(entry).length > 0);
+          }
+
           if (evaluator != null) {
             entries = CloseableIterable.filter(entries,
                 entry -> evaluator.eval((GenericDataFile) entry.file()));

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

@@ -22,6 +22,7 @@
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.util.List;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import org.apache.iceberg.Accessor;
 import org.apache.iceberg.MetadataColumns;
@@ -73,6 +74,16 @@ public static <T> CloseableIterable<T> filter(CloseableIterable<T> rows, Functio
     return filter.filter(rows);
   }
 
+  public static <T> CloseableIterable<T> marker(CloseableIterable<T> rows, Function<T, Long> rowToPosition,
+                                                PositionDeleteIndex deleteSet, Consumer<T> markDeleted) {
+    if (deleteSet.isEmpty()) {
+      return rows;
+    }
+
+    PositionSetDeleteMarker<T> deleteMarker = new PositionSetDeleteMarker<>(rowToPosition, deleteSet, markDeleted);
+    return deleteMarker.filter(rows);
+  }
+
   public static StructLikeSet toEqualitySet(CloseableIterable<StructLike> eqDeletes, Types.StructType eqType) {
     try (CloseableIterable<StructLike> deletes = eqDeletes) {
       StructLikeSet deleteSet = StructLikeSet.create(eqType);
@@ -107,6 +118,14 @@ public static <T> CloseableIterable<T> streamingFilter(CloseableIterable<T> rows
     return new PositionStreamDeleteFilter<>(rows, rowToPosition, posDeletes);
   }
 
+
+  public static <T> CloseableIterable<T> streamingMarker(CloseableIterable<T> rows,
+                                                         Function<T, Long> rowToPosition,
+                                                         CloseableIterable<Long> posDeletes,
+                                                         Consumer<T> markDeleted) {
+    return new PositionStreamDeleteMarker<>(rows, rowToPosition, posDeletes, markDeleted);
+  }
+
   public static CloseableIterable<Long> deletePositions(CharSequence dataLocation,
                                                         CloseableIterable<StructLike> deleteFile) {
     return deletePositions(dataLocation, ImmutableList.of(deleteFile));
@@ -152,6 +171,29 @@ protected boolean shouldKeep(T row) {
     }
   }
 
+  private static class PositionSetDeleteMarker<T> extends Filter<T> {
+    private final Function<T, Long> rowToPosition;
+    private final PositionDeleteIndex deleteSet;
+    private final Consumer<T> markDeleted;
+
+    private PositionSetDeleteMarker(Function<T, Long> rowToPosition, PositionDeleteIndex deleteSet,
+                                    Consumer<T> markDeleted) {
+      this.rowToPosition = rowToPosition;
+      this.deleteSet = deleteSet;
+      this.markDeleted = markDeleted;
+    }
+
+    @Override
+    protected boolean shouldKeep(T row) {
+      if (deleteSet.isDeleted(rowToPosition.apply(row))) {
+        markDeleted.accept(row);
+      }
+
+      // always return true, since we don't want to remove the row
+      return true;
+    }
+  }
+
   private static class PositionStreamDeleteFilter<T> extends CloseableGroup implements CloseableIterable<T> {
     private final CloseableIterable<T> rows;
     private final Function<T, Long> extractPos;
@@ -170,7 +212,7 @@ public CloseableIterator<T> iterator() {
 
       CloseableIterator<T> iter;
       if (deletePosIterator.hasNext()) {
-        iter = new PositionFilterIterator(rows.iterator(), deletePosIterator);
+        iter = createPosDeleteIterator(rows.iterator(), deletePosIterator);
       } else {
         iter = rows.iterator();
         try {
@@ -185,7 +227,12 @@ public CloseableIterator<T> iterator() {
       return iter;
     }
 
-    private class PositionFilterIterator extends FilterIterator<T> {
+    protected PositionFilterIterator createPosDeleteIterator(CloseableIterator<T> items,
+                                                             CloseableIterator<Long> deletePosIterator) {
+      return new PositionFilterIterator(items, deletePosIterator);
+    }
+
+    protected class PositionFilterIterator extends FilterIterator<T> {
       private final CloseableIterator<Long> deletePosIterator;
       private long nextDeletePos;
 
@@ -227,6 +274,37 @@ public void close() {
     }
   }
 
+  private static class PositionStreamDeleteMarker<T> extends PositionStreamDeleteFilter<T> {
+    private final Consumer<T> markDeleted;
+
+    private PositionStreamDeleteMarker(CloseableIterable<T> rows, Function<T, Long> extractPos,
+                                       CloseableIterable<Long> deletePositions, Consumer<T> markDeleted) {
+      super(rows, extractPos, deletePositions);
+      this.markDeleted = markDeleted;
+    }
+
+    @Override
+    protected PositionFilterIterator createPosDeleteIterator(CloseableIterator<T> items,
+                                                             CloseableIterator<Long> deletePosIterator) {
+      return new PositionDeleteMarkerIterator(items, deletePosIterator);
+    }
+
+    private class PositionDeleteMarkerIterator extends PositionFilterIterator {
+      private PositionDeleteMarkerIterator(CloseableIterator<T> items, CloseableIterator<Long> deletePositions) {
+        super(items, deletePositions);
+      }
+
+      @Override
+      protected boolean shouldKeep(T row) {
+        boolean isDeleted = !super.shouldKeep(row);
+        if (isDeleted) {
+          markDeleted.accept(row);
+        }
+        return true;
+      }
+    }
+  }
+
   private static class DataFileFilter<T extends StructLike> extends Filter<T> {
     private final CharSequence dataLocation;