[Kernel] During Active-AddFile-Log-Replay do not pass the RemoveFile to checkpoint reader

#### Which Delta project/connector is this regarding?

- [ ] Spark
- [ ] Standalone
- [ ] Flink
- [X] Kernel
- [ ] Other (fill in here)

## Description

Implemented a minor performance improvement to not read any RemoveFiles when we read checkpoint Parquet files during active-add-file-log-replay.

## How was this patch tested?

Existing unit test, manual test using
./run-tests.py --group kernel
and
delta/kernel/examples/run-kernel-examples.py --use-local

## Does this PR introduce _any_ user-facing changes?

No

Author: anoopj

kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/ActionsIterator.java

@@ -62,6 +62,8 @@ public class ActionsIterator implements CloseableIterator<ActionWrapper> {
 
   private final StructType readSchema;
 
+  private final StructType checkpointReadSchema;
+
   private final boolean schemaContainsAddOrRemoveFiles;
 
   /**
@@ -79,12 +81,22 @@ public ActionsIterator(
       List<FileStatus> files,
       StructType readSchema,
       Optional<Predicate> checkpointPredicate) {
+    this(engine, files, readSchema, readSchema, checkpointPredicate);
+  }
+
+  public ActionsIterator(
+      Engine engine,
+      List<FileStatus> files,
+      StructType readSchema,
+      StructType checkpointReadSchema,
+      Optional<Predicate> checkpointPredicate) {
     this.engine = engine;
     this.checkpointPredicate = checkpointPredicate;
     this.filesList = new LinkedList<>();
     this.filesList.addAll(
         files.stream().map(DeltaLogFile::forCommitOrCheckpoint).collect(Collectors.toList()));
     this.readSchema = readSchema;
+    this.checkpointReadSchema = checkpointReadSchema;
     this.actionsIter = Optional.empty();
     this.schemaContainsAddOrRemoveFiles = LogReplay.containsAddOrRemoveFileActions(readSchema);
   }
@@ -177,8 +189,10 @@ private CloseableIterator<ColumnarBatch> getActionsIterFromSinglePartOrV2Checkpo
     // If the sidecars may contain the current action, read sidecars from the top-level v2
     // checkpoint file(to be read later).
     StructType modifiedReadSchema = readSchema;
+    StructType modifiedCheckpointReadSchema = checkpointReadSchema;
     if (schemaContainsAddOrRemoveFiles) {
       modifiedReadSchema = LogReplay.withSidecarFileSchema(readSchema);
+      modifiedCheckpointReadSchema = LogReplay.withSidecarFileSchema(checkpointReadSchema);
     }
 
     long checkpointVersion = checkpointVersion(file.getPath());
@@ -195,7 +209,12 @@ private CloseableIterator<ColumnarBatch> getActionsIterFromSinglePartOrV2Checkpo
       checkpointPredicateIncludingSidecars = checkpointPredicate;
     }
     final CloseableIterator<ColumnarBatch> topLevelIter;
-    StructType finalModifiedReadSchema = modifiedReadSchema;
+    StructType finalCommitReadSchema = modifiedReadSchema;
+    // We do not need to look at any `remove` files from the checkpoints. Skip the column to save
+    // I/O. Note that we are still going to process the row groups. Adds and removes are randomly
+    // scattered through checkpoint part files, so row group push down is unlikely to be useful.
+    StructType finalCheckpointReadSchema = modifiedCheckpointReadSchema;
+
     if (fileName.endsWith(".parquet")) {
       topLevelIter =
           wrapEngineExceptionThrowsIO(
@@ -204,7 +223,7 @@ private CloseableIterator<ColumnarBatch> getActionsIterFromSinglePartOrV2Checkpo
                       .getParquetHandler()
                       .readParquetFiles(
                           singletonCloseableIterator(file),
-                          finalModifiedReadSchema,
+                          finalCheckpointReadSchema,
                           checkpointPredicateIncludingSidecars),
               "Reading parquet log file `%s` with readSchema=%s and predicate=%s",
               file,
@@ -218,7 +237,7 @@ private CloseableIterator<ColumnarBatch> getActionsIterFromSinglePartOrV2Checkpo
                       .getJsonHandler()
                       .readJsonFiles(
                           singletonCloseableIterator(file),
-                          finalModifiedReadSchema,
+                          finalCommitReadSchema,
                           checkpointPredicateIncludingSidecars),
               "Reading JSON log file `%s` with readSchema=%s and predicate=%s",
               file,

kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/ActiveAddFilesIterator.java

@@ -233,9 +233,12 @@ private void prepareNext() {
       }
     }
 
+    ColumnarBatch scanAddFiles = addRemoveColumnarBatch;
     // Step 3: Drop the RemoveFile column and use the selection vector to build a new
     //         FilteredColumnarBatch
-    ColumnarBatch scanAddFiles = addRemoveColumnarBatch.withDeletedColumnAt(1);
+    if (scanAddFiles.getSchema().length() > 1) {
+      scanAddFiles = scanAddFiles.withDeletedColumnAt(1);
+    }
 
     // Step 4: TODO: remove this step. This is a temporary requirement until the path
     //         in `add` is converted to absolute path.

kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/LogReplay.java

@@ -106,6 +106,11 @@ public static StructType getAddRemoveReadSchema(boolean shouldReadStats) {
         .add(REMOVEFILE_FIELD_NAME, REMOVE_FILE_SCHEMA);
   }
 
+  /** Read schema when searching only for AddFiles */
+  public static StructType getAddReadSchema(boolean shouldReadStats) {
+    return new StructType().add(ADDFILE_FIELD_NAME, getAddSchema(shouldReadStats));
+  }
+
   public static int ADD_FILE_ORDINAL = 0;
   public static int ADD_FILE_PATH_ORDINAL = AddFile.SCHEMA_WITHOUT_STATS.indexOf("path");
   public static int ADD_FILE_DV_ORDINAL = AddFile.SCHEMA_WITHOUT_STATS.indexOf("deletionVector");
@@ -201,6 +206,7 @@ public CloseableIterator<FilteredColumnarBatch> getAddFilesAsColumnarBatches(
             engine,
             logSegment.allLogFilesReversed(),
             getAddRemoveReadSchema(shouldReadStats),
+            getAddReadSchema(shouldReadStats),
             checkpointPredicate);
     return new ActiveAddFilesIterator(engine, addRemoveIter, dataPath, scanMetrics);
   }