[Kernel]Remove two file IO in CRC loading by reusing log listing (#4112)

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md
2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  3. Be sure to keep the PR description updated to reflect all changes.
  4. Please write your PR title to summarize what this PR proposes.
5. If possible, provide a concise example to reproduce the issue for a
faster review.
6. If applicable, include the corresponding issue number in the PR title
and link it in the body.
-->

#### Which Delta project/connector is this regarding?
<!--
Please add the component selected below to the beginning of the pull
request title
For example: [Spark] Title of my pull request
-->

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

## Description

<!--
- Describe what this PR changes.
- Describe why we need the change.
 
If this PR resolves an issue be sure to include "Resolves #XXX" to
correctly link and close the issue upon merge.
-->
Incremental CRC loading for P&M was added in
#4077.

For snapshot N, we try to read N.crc, if N.crc missing, we list CRC
files and find the one up to N-100. These two file listing operation
could be actually avoided, because we have already listed files under
_delta_log folder.

This PR collects latest CRC in the first file listing and remove two
unnecessary IO.

## How was this patch tested?

<!--
If tests were added, say they were added here. Please make sure to test
the changes thoroughly including negative and positive cases if
possible.
If the changes were tested in any way other than unit tests, please
clarify how you tested step by step (ideally copy and paste-able, so
that other reviewers can test and check, and descendants can verify in
the future).
If the changes were not tested, please explain why.
-->
LogReplay test
## Does this PR introduce _any_ user-facing changes?

<!--
If yes, please clarify the previous behavior and the change this PR
proposes - provide the console output, description and/or an example to
show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change
compared to the released Delta Lake versions or within the unreleased
branches such as master.
If no, write 'No'.
-->
No

Author: huan233usc

kernel/kernel-api/src/main/java/io/delta/kernel/internal/DeltaLogActionUtils.java

@@ -219,8 +219,10 @@ public static CloseableIterator<FileStatus> listDeltaLogFilesAsIter(
         startVersion,
         endVersionOpt);
 
+    // This variable is used to help determine if we should throw an error if the table history is
+    // not reconstructable. Only commit and checkpoint files are applicable.
     // Must be final to be used in lambda
-    final AtomicBoolean hasReturnedAnElement = new AtomicBoolean(false);
+    final AtomicBoolean hasReturnedCommitOrCheckpoint = new AtomicBoolean(false);
 
     return listLogDir(engine, tablePath, startVersion)
         .breakableFilter(
@@ -237,6 +239,9 @@ public static CloseableIterator<FileStatus> listDeltaLogFilesAsIter(
                 // Checkpoint files of 0 size are invalid but may be ignored silently when read,
                 // hence we ignore them so that we never pick up such checkpoints.
                 // Here, we do nothing (we will consume this file).
+              } else if (fileTypes.contains(DeltaLogFileType.CHECKSUM)
+                  && FileNames.isChecksumFile(fileName)) {
+                // Here, we do nothing (we will consume this file).
               } else {
                 logger.debug("Ignoring file {} as it is not of the desired type", fs.getPath());
                 return BreakableFilterResult.EXCLUDE; // Here, we exclude and filter out this file.
@@ -277,7 +282,7 @@ public static CloseableIterator<FileStatus> listDeltaLogFilesAsIter(
                 final long endVersion = endVersionOpt.get();
 
                 if (fileVersion > endVersion) {
-                  if (mustBeRecreatable && !hasReturnedAnElement.get()) {
+                  if (mustBeRecreatable && !hasReturnedCommitOrCheckpoint.get()) {
                     final long earliestVersion =
                         DeltaHistoryManager.getEarliestRecreatableCommit(engine, logPath);
                     throw DeltaErrors.versionBeforeFirstAvailableCommit(
@@ -292,7 +297,11 @@ public static CloseableIterator<FileStatus> listDeltaLogFilesAsIter(
                 }
               }
 
-              hasReturnedAnElement.set(true);
+              if (FileNames.isCommitFile(fileName)
+                  || FileNames.isCheckpointFile(fileName)
+                  || FileNames.isLogCompactionFile(fileName)) {
+                hasReturnedCommitOrCheckpoint.set(true);
+              }
 
               return BreakableFilterResult.INCLUDE;
             });

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checksum/ChecksumReader.java

@@ -17,15 +17,13 @@
 
 import static io.delta.kernel.internal.util.FileNames.*;
 import static io.delta.kernel.internal.util.Utils.singletonCloseableIterator;
-import static java.lang.Math.min;
 
 import io.delta.kernel.data.ColumnarBatch;
 import io.delta.kernel.engine.Engine;
 import io.delta.kernel.internal.fs.Path;
 import io.delta.kernel.internal.util.FileNames;
 import io.delta.kernel.utils.CloseableIterator;
 import io.delta.kernel.utils.FileStatus;
-import java.io.IOException;
 import java.util.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -35,89 +33,41 @@ public class ChecksumReader {
   private static final Logger logger = LoggerFactory.getLogger(ChecksumReader.class);
 
   /**
-   * Load the CRCInfo from the checksum file at the given version. If the checksum file is not found
-   * at the given version, it will try to find the latest checksum file that is created at or after
-   * the lower bound version.
+   * Load the CRCInfo from the provided checksum file.
    *
    * @param engine the engine to use for reading the checksum file
-   * @param logPath the path to the Delta log
-   * @param targetedVersion the target version to read the checksum file from
-   * @param lowerBound the inclusive lower bound version to search for the checksum file
-   * @return Optional {@link CRCInfo} containing the protocol and metadata, and the version of the
-   *     checksum file. If the checksum file is not found, it will return an empty
+   * @param checkSumFile the file status of the checksum file to read
+   * @return Optional {@link CRCInfo} containing the information included in the checksum file, such
+   *     as protocol, metadata.
    */
-  public static Optional<CRCInfo> getCRCInfo(
-      Engine engine, Path logPath, long targetedVersion, long lowerBound) {
-    // lower bound should always smaller than the targetedVersion.
-    lowerBound = min(lowerBound, targetedVersion);
-    logger.info("Loading CRC file for version {} with lower bound {}", targetedVersion, lowerBound);
-    // First try to load the CRC at given version. If not found or failed to read then try to
-    // find the latest CRC file that is created at or after the lower bound version.
-    Path crcFilePath = checksumFile(logPath, targetedVersion);
-    Optional<CRCInfo> crcInfoOpt = readChecksumFile(engine, crcFilePath);
-    if (crcInfoOpt.isPresent()
-        ||
-        // we don't expect any more checksum files as it is the first version
-        targetedVersion == 0
-        || targetedVersion == lowerBound) {
-      return crcInfoOpt;
-    }
-    logger.info(
-        "CRC file for version {} not found, listing CRC files from version {}",
-        targetedVersion,
-        lowerBound);
-
-    Path lowerBoundFilePath = checksumFile(logPath, lowerBound);
-    try (CloseableIterator<FileStatus> crcFiles =
-        engine.getFileSystemClient().listFrom(lowerBoundFilePath.toString())) {
-      List<FileStatus> crcFilesList =
-          crcFiles
-              .filter(file -> isChecksumFile(file.getPath()))
-              .takeWhile(file -> checksumVersion(new Path(file.getPath())) <= targetedVersion)
-              .toInMemoryList();
-
-      // pick the last file which is the latest version that has the CRC file
-      if (crcFilesList.isEmpty()) {
-        logger.warn("No checksum files found in the range {} to {}", lowerBound, targetedVersion);
-        return Optional.empty();
-      }
-
-      FileStatus latestCRCFile = crcFilesList.get(crcFilesList.size() - 1);
-      return readChecksumFile(engine, new Path(latestCRCFile.getPath()));
-    } catch (IOException e) {
-      logger.warn("Failed to list checksum files from {}", lowerBoundFilePath, e);
-      return Optional.empty();
-    }
-  }
-
-  private static Optional<CRCInfo> readChecksumFile(Engine engine, Path filePath) {
+  public static Optional<CRCInfo> getCRCInfo(Engine engine, FileStatus checkSumFile) {
     try (CloseableIterator<ColumnarBatch> iter =
         engine
             .getJsonHandler()
             .readJsonFiles(
-                singletonCloseableIterator(FileStatus.of(filePath.toString())),
+                singletonCloseableIterator(checkSumFile),
                 CRCInfo.CRC_FILE_SCHEMA,
                 Optional.empty())) {
       // We do this instead of iterating through the rows or using `getSingularRow` so we
       // can use the existing fromColumnVector methods in Protocol, Metadata, Format etc
       if (!iter.hasNext()) {
-        logger.warn("Checksum file is empty: {}", filePath);
+        logger.warn("Checksum file is empty: {}", checkSumFile.getPath());
         return Optional.empty();
       }
 
       ColumnarBatch batch = iter.next();
       if (batch.getSize() != 1) {
         String msg = "Expected exactly one row in the checksum file {}, found {} rows";
-        logger.warn(msg, filePath, batch.getSize());
+        logger.warn(msg, checkSumFile.getPath(), batch.getSize());
         return Optional.empty();
       }
 
-      long crcVersion = FileNames.checksumVersion(filePath);
+      long crcVersion = FileNames.checksumVersion(new Path(checkSumFile.getPath()));
 
-      return CRCInfo.fromColumnarBatch(crcVersion, batch, 0 /* rowId */, filePath.toString());
+      return CRCInfo.fromColumnarBatch(crcVersion, batch, 0 /* rowId */, checkSumFile.getPath());
     } catch (Exception e) {
       // This can happen when the version does not have a checksum file
-      logger.warn("Failed to read checksum file {}", filePath, e);
+      logger.warn("Failed to read checksum file {}", checkSumFile.getPath(), e);
       return Optional.empty();
     }
   }

kernel/kernel-api/src/main/java/io/delta/kernel/internal/compaction/LogCompactionWriter.java

@@ -105,6 +105,7 @@ public void writeLogCompactionFile(Engine engine) throws IOException {
             deltas,
             Collections.emptyList(),
             Collections.emptyList(),
+            Optional.empty(),
             lastCommitTimestamp);
     CreateCheckpointIterator checkpointIterator =
         new CreateCheckpointIterator(engine, segment, minFileRetentionTimestampMillis);

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

@@ -19,6 +19,7 @@
 import static io.delta.kernel.internal.DeltaErrors.wrapEngineExceptionThrowsIO;
 import static io.delta.kernel.internal.TableConfig.IN_COMMIT_TIMESTAMPS_ENABLED;
 import static io.delta.kernel.internal.actions.SingleAction.*;
+import static io.delta.kernel.internal.util.FileNames.checksumFile;
 import static io.delta.kernel.internal.util.FileNames.deltaFile;
 import static io.delta.kernel.internal.util.Preconditions.checkArgument;
 import static io.delta.kernel.internal.util.Preconditions.checkState;
@@ -175,7 +176,8 @@ public TransactionRebaseState resolveConflicts(Engine engine) throws ConcurrentW
 
     Optional<CRCInfo> updatedCrcInfo =
         ChecksumReader.getCRCInfo(
-            engine, snapshot.getLogPath(), lastWinningVersion, lastWinningVersion);
+            engine,
+            FileStatus.of(checksumFile(snapshot.getLogPath(), lastWinningVersion).toString()));
 
     // if we get here, we have successfully rebased (i.e no logical conflicts)
     // against the winning transactions

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

@@ -18,8 +18,7 @@
 
 import static io.delta.kernel.internal.replay.LogReplayUtils.assertLogFilesBelongToTable;
 import static io.delta.kernel.internal.util.Preconditions.checkArgument;
-import static java.util.Arrays.asList;
-import static java.util.Collections.max;
+import static java.util.Objects.requireNonNull;
 
 import io.delta.kernel.data.ColumnVector;
 import io.delta.kernel.data.ColumnarBatch;
@@ -38,6 +37,7 @@
 import io.delta.kernel.internal.snapshot.SnapshotHint;
 import io.delta.kernel.internal.tablefeatures.TableFeatures;
 import io.delta.kernel.internal.util.DomainMetadataUtils;
+import io.delta.kernel.internal.util.FileNames;
 import io.delta.kernel.internal.util.Tuple2;
 import io.delta.kernel.types.StringType;
 import io.delta.kernel.types.StructType;
@@ -135,7 +135,7 @@ public static StructType getAddReadSchema(boolean shouldReadStats) {
   private final LogSegment logSegment;
   private final Tuple2<Protocol, Metadata> protocolAndMetadata;
   private final Lazy<Map<String, DomainMetadata>> domainMetadataMap;
-  private final Optional<CRCInfo> currentCrcInfo;
+  private final CrcInfoContext crcInfoContext;
 
   public LogReplay(
       Path logPath,
@@ -147,17 +147,23 @@ public LogReplay(
       SnapshotMetrics snapshotMetrics) {
 
     assertLogFilesBelongToTable(logPath, logSegment.allLogFilesUnsorted());
-    Tuple2<Optional<SnapshotHint>, Optional<CRCInfo>> newerSnapshotHintAndCurrentCrcInfo =
-        maybeGetNewerSnapshotHintAndCurrentCrcInfo(
-            engine, logSegment, snapshotHint, snapshotVersion);
-    this.currentCrcInfo = newerSnapshotHintAndCurrentCrcInfo._2;
+
+    // Ignore the snapshot hint whose version is larger than the snapshot version.
+    if (snapshotHint.isPresent() && snapshotHint.get().getVersion() > snapshotVersion) {
+      snapshotHint = Optional.empty();
+    }
+
+    this.crcInfoContext = new CrcInfoContext(engine);
     this.dataPath = dataPath;
     this.logSegment = logSegment;
+    Optional<SnapshotHint> newerSnapshotHint =
+        crcInfoContext.maybeGetNewerSnapshotHintAndUpdateCache(
+            engine, logSegment, snapshotHint, snapshotVersion);
     this.protocolAndMetadata =
         snapshotMetrics.loadInitialDeltaActionsTimer.time(
             () ->
                 loadTableProtocolAndMetadata(
-                    engine, logSegment, newerSnapshotHintAndCurrentCrcInfo._1, snapshotVersion));
+                    engine, logSegment, newerSnapshotHint, snapshotVersion));
     // Lazy loading of domain metadata only when needed
     this.domainMetadataMap = new Lazy<>(() -> loadDomainMetadataMap(engine));
   }
@@ -186,9 +192,11 @@ public long getVersion() {
     return logSegment.getVersion();
   }
 
-  /** Returns the crc info for the current snapshot if the checksum file is read */
+  /** Returns the crc info for the current snapshot if it is cached */
   public Optional<CRCInfo> getCurrentCrcInfo() {
-    return currentCrcInfo;
+    return crcInfoContext
+        .getLastSeenCrcInfo()
+        .filter(crcInfo -> crcInfo.getVersion() == getVersion());
   }
 
   /**
@@ -367,6 +375,7 @@ private Optional<Long> loadLatestTransactionVersion(Engine engine, String applic
    */
   private Map<String, DomainMetadata> loadDomainMetadataMap(Engine engine) {
     // First try to load from CRC info if available
+    Optional<CRCInfo> currentCrcInfo = getCurrentCrcInfo();
     if (currentCrcInfo.isPresent() && currentCrcInfo.get().getDomainMetadata().isPresent()) {
       return currentCrcInfo.get().getDomainMetadata().get().stream()
           .collect(Collectors.toMap(DomainMetadata::getDomain, Function.identity()));
@@ -415,50 +424,87 @@ private Map<String, DomainMetadata> loadDomainMetadataMapFromLog(Engine engine)
   }
 
   /**
-   * Calculates the latest snapshot hint before or at the current snapshot version, returns the
-   * CRCInfo if checksum file at the current version is read
+   * Encapsulates CRC-related functionality and state for the LogReplay. This includes caching CRC
+   * info and extracting snapshot hints from CRC files.
+   *
+   * <p>This class uses {@code maybeGetNewerSnapshotHintAndUpdateCache} to calculate a {@code
+   * SnapshotHint} and also exposes a {@code getLastSeenCrcInfo} method. Their relationship is:
+   *
+   * <ul>
+   *   <li>We want to find the latest {@code SnapshotHint} to use during log replay for Protocol and
+   *       Metadata loading
+   *   <li>If we are not provided a SnapshotHint for this version, or are provided a stale hint, we
+   *       will try to read the latest seen (by file listing) CRC file (if it exists). If so, we
+   *       read it, cache it, and create a newer hint.
+   *   <li>Then, when {@code getLastSeenCrcInfo} is called, we will either use the cached CRCInfo
+   *       that we have already read, parsed, and cached; or, if it was never cached (because the
+   *       hint was sufficiently new) we will read it, parse it, and cache it for the first time
+   * </ul>
    */
-  private Tuple2<Optional<SnapshotHint>, Optional<CRCInfo>>
-      maybeGetNewerSnapshotHintAndCurrentCrcInfo(
-          Engine engine,
-          LogSegment logSegment,
-          Optional<SnapshotHint> snapshotHint,
-          long snapshotVersion) {
-
-    // Snapshot hint's version is current.
-    if (snapshotHint.isPresent() && snapshotHint.get().getVersion() == snapshotVersion) {
-      return new Tuple2<>(snapshotHint, Optional.empty());
+  private class CrcInfoContext {
+    private final Engine engine;
+    private Optional<CRCInfo> cachedLastSeenCrcInfo;
+
+    CrcInfoContext(Engine engine) {
+      this.engine = requireNonNull(engine);
+      this.cachedLastSeenCrcInfo = Optional.empty();
     }
 
-    // Ignore the snapshot hint whose version is larger.
-    if (snapshotHint.isPresent() && snapshotHint.get().getVersion() > snapshotVersion) {
-      snapshotHint = Optional.empty();
+    /** Returns the CRC info persisted in the logSegment's lastSeenChecksum File */
+    public Optional<CRCInfo> getLastSeenCrcInfo() {
+      if (!cachedLastSeenCrcInfo.isPresent()) {
+        cachedLastSeenCrcInfo =
+            logSegment
+                .getLastSeenChecksum()
+                .flatMap(crcFile -> ChecksumReader.getCRCInfo(engine, crcFile));
+      }
+      return cachedLastSeenCrcInfo;
     }
 
-    long crcSearchLowerBound =
-        max(
-            asList(
-                // Prefer reading hint over CRC, so start listing from hint's version + 1,
-                // if hint is not present, list from version 0.
-                snapshotHint.map(SnapshotHint::getVersion).orElse(-1L) + 1,
-                logSegment.getCheckpointVersionOpt().orElse(0L),
-                // Only find the CRC within 100 versions.
-                snapshotVersion - 100,
-                0L));
-    Optional<CRCInfo> crcInfoOpt =
-        ChecksumReader.getCRCInfo(
-            engine, logSegment.getLogPath(), snapshotVersion, crcSearchLowerBound);
-    if (!crcInfoOpt.isPresent()) {
-      return new Tuple2<>(snapshotHint, Optional.empty());
+    /**
+     * Attempts to build a newer snapshot hint from CRC that can be used for loading table state
+     * more efficiently. When CRC is read, updates the internal cache.
+     *
+     * @param engine The engine used to read CRC files
+     * @param logSegment The log segment containing checksum information
+     * @param snapshotHint Existing snapshot hint, if any
+     * @param snapshotVersion Target snapshot version
+     * @return An updated snapshot hint if a newer CRC file was found, otherwise the original hint
+     */
+    public Optional<SnapshotHint> maybeGetNewerSnapshotHintAndUpdateCache(
+        Engine engine,
+        LogSegment logSegment,
+        Optional<SnapshotHint> snapshotHint,
+        long snapshotVersion) {
+
+      // Snapshot hint's version is current so we could use it in loading P&M.
+      // No need to read crc.
+      if (snapshotHint.isPresent() && snapshotHint.get().getVersion() == snapshotVersion) {
+        return snapshotHint;
+      }
+
+      // Prefer reading hint over CRC to save 1 io, only read crc if it is newer than snapshot hint.
+      long crcReadLowerBound = snapshotHint.map(SnapshotHint::getVersion).orElse(-1L) + 1;
+
+      Optional<CRCInfo> crcInfoOpt =
+          logSegment
+              .getLastSeenChecksum()
+              .filter(
+                  checksum ->
+                      FileNames.getFileVersion(new Path(checksum.getPath())) >= crcReadLowerBound)
+              .flatMap(checksum -> ChecksumReader.getCRCInfo(engine, checksum));
+
+      if (!crcInfoOpt.isPresent()) {
+        return snapshotHint;
+      }
+
+      CRCInfo crcInfo = crcInfoOpt.get();
+      this.cachedLastSeenCrcInfo = Optional.of(crcInfo);
+      checkArgument(
+          crcInfo.getVersion() >= crcReadLowerBound && crcInfo.getVersion() <= snapshotVersion);
+      // We found a CRCInfo of a version (a) older than the one we are looking for (snapshotVersion)
+      // but (b) newer than the current hint. Use this CRCInfo to create a new hint, and return.
+      return Optional.of(SnapshotHint.fromCrcInfo(crcInfo));
     }
-    CRCInfo crcInfo = crcInfoOpt.get();
-    checkArgument(
-        crcInfo.getVersion() >= crcSearchLowerBound && crcInfo.getVersion() <= snapshotVersion);
-    // We found a CRCInfo of a version (a) older than the one we are looking for (snapshotVersion)
-    // but (b) newer than the current hint. Use this CRCInfo to create a new hint, and return this
-    // crc info if it matches the current version.
-    return new Tuple2<>(
-        Optional.of(SnapshotHint.fromCrcInfo(crcInfo)),
-        crcInfo.getVersion() == snapshotVersion ? crcInfoOpt : Optional.empty());
   }
 }