[Kernel]Remove two file IO in CRC loading by reusing log listing

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());
   }
 }