[Kernel] Load the protocol and metadata from the CRC files when avail…

…able (#4077)

<!--
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.
-->
CP crc loading code to master branch from
"kernel-20250115-crc-optimization"
PR created using `git cherry-pick
7d32f66`
## 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.
-->

## 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

---------

Co-authored-by: Venki Korukanti <[email protected]>
Co-authored-by: Allison Portis <[email protected]>

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

@@ -258,7 +258,10 @@ private LogReplay getEmptyLogReplay(
 
       @Override
       protected Tuple2<Protocol, Metadata> loadTableProtocolAndMetadata(
-          Engine engine, Optional<SnapshotHint> snapshotHint, long snapshotVersion) {
+          Engine engine,
+          LogSegment logSegment,
+          Optional<SnapshotHint> snapshotHint,
+          long snapshotVersion) {
         return new Tuple2<>(protocol, metadata);
       }
 

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

@@ -0,0 +1,74 @@
+/*
+ * Copyright (2025) The Delta Lake Project Authors.
+ *
+ * Licensed 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 io.delta.kernel.internal.replay;
+
+import static java.util.Objects.requireNonNull;
+
+import io.delta.kernel.data.ColumnarBatch;
+import io.delta.kernel.internal.actions.Metadata;
+import io.delta.kernel.internal.actions.Protocol;
+import io.delta.kernel.types.StructType;
+import java.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CRCInfo {
+  private static final Logger logger = LoggerFactory.getLogger(CRCInfo.class);
+
+  public static Optional<CRCInfo> fromColumnarBatch(
+      long version, ColumnarBatch batch, int rowId, String crcFilePath) {
+    Protocol protocol = Protocol.fromColumnVector(batch.getColumnVector(PROTOCOL_ORDINAL), rowId);
+    Metadata metadata = Metadata.fromColumnVector(batch.getColumnVector(METADATA_ORDINAL), rowId);
+    //  protocol and metadata are nullable per fromColumnVector's implementation.
+    if (protocol == null || metadata == null) {
+      logger.warn("Invalid checksum file missing protocol and/or metadata: {}", crcFilePath);
+      return Optional.empty();
+    }
+    return Optional.of(new CRCInfo(version, metadata, protocol));
+  }
+
+  // We can add additional fields later
+  public static final StructType FULL_SCHEMA =
+      new StructType().add("protocol", Protocol.FULL_SCHEMA).add("metadata", Metadata.FULL_SCHEMA);
+
+  private static final int PROTOCOL_ORDINAL = 0;
+  private static final int METADATA_ORDINAL = 1;
+
+  private final long version;
+  private final Metadata metadata;
+  private final Protocol protocol;
+
+  protected CRCInfo(long version, Metadata metadata, Protocol protocol) {
+    this.version = version;
+    this.metadata = requireNonNull(metadata);
+    this.protocol = requireNonNull(protocol);
+  }
+
+  /** The version of the Delta table that this CRCInfo represents. */
+  public long getVersion() {
+    return version;
+  }
+
+  /** The {@link Metadata} stored in this CRCInfo. */
+  public Metadata getMetadata() {
+    return metadata;
+  }
+
+  /** The {@link Protocol} stored in this CRCInfo. */
+  public Protocol getProtocol() {
+    return protocol;
+  }
+}

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

@@ -0,0 +1,124 @@
+/*
+ * Copyright (2025) The Delta Lake Project Authors.
+ *
+ * Licensed 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 io.delta.kernel.internal.replay;
+
+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;
+
+/** Utility method to load protocol and metadata from the Delta log checksum files. */
+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.
+   *
+   * @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
+   */
+  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) {
+    try (CloseableIterator<ColumnarBatch> iter =
+        engine
+            .getJsonHandler()
+            .readJsonFiles(
+                singletonCloseableIterator(FileStatus.of(filePath.toString())),
+                CRCInfo.FULL_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);
+        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());
+        return Optional.empty();
+      }
+
+      long crcVersion = FileNames.checksumVersion(filePath);
+
+      return CRCInfo.fromColumnarBatch(crcVersion, batch, 0 /* rowId */, filePath.toString());
+    } catch (Exception e) {
+      // This can happen when the version does not have a checksum file
+      logger.warn("Failed to read checksum file {}", filePath, e);
+      return Optional.empty();
+    }
+  }
+}

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

@@ -17,6 +17,9 @@
 package io.delta.kernel.internal.replay;
 
 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 io.delta.kernel.data.ColumnVector;
 import io.delta.kernel.data.ColumnarBatch;
@@ -38,9 +41,7 @@
 import io.delta.kernel.utils.CloseableIterator;
 import java.io.IOException;
 import java.io.UncheckedIOException;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
+import java.util.*;
 
 /**
  * Replays a history of actions, resolving them to produce the current state of the table. The
@@ -135,7 +136,7 @@ public LogReplay(
     this.logSegment = logSegment;
     this.protocolAndMetadata =
         snapshotMetrics.loadInitialDeltaActionsTimer.time(
-            () -> loadTableProtocolAndMetadata(engine, snapshotHint, snapshotVersion));
+            () -> loadTableProtocolAndMetadata(engine, logSegment, snapshotHint, snapshotVersion));
     // Lazy loading of domain metadata only when needed
     this.domainMetadataMap = new Lazy<>(() -> loadDomainMetadataMap(engine));
   }
@@ -201,13 +202,47 @@ public CloseableIterator<FilteredColumnarBatch> getAddFilesAsColumnarBatches(
    * use the P and/or M from the hint.
    */
   protected Tuple2<Protocol, Metadata> loadTableProtocolAndMetadata(
-      Engine engine, Optional<SnapshotHint> snapshotHint, long snapshotVersion) {
+      Engine engine,
+      LogSegment logSegment,
+      Optional<SnapshotHint> snapshotHint,
+      long snapshotVersion) {
 
-    // Exit early if the hint already has the info we need
+    // Exit early if the hint already has the info we need.
     if (snapshotHint.isPresent() && snapshotHint.get().getVersion() == snapshotVersion) {
       return new Tuple2<>(snapshotHint.get().getProtocol(), snapshotHint.get().getMetadata());
     }
 
+    // Snapshot hit is not use-able in this case for determine the lower bound.
+    if (snapshotHint.isPresent() && snapshotHint.get().getVersion() > snapshotVersion) {
+      snapshotHint = Optional.empty();
+    }
+
+    long crcSearchLowerBound =
+        max(
+            asList(
+                // Prefer reading hint over CRC, so start listing from hint's version + 1.
+                snapshotHint.map(SnapshotHint::getVersion).orElse(0L) + 1,
+                logSegment.checkpointVersionOpt.orElse(0L),
+                // Only find the CRC within 100 versions.
+                snapshotVersion - 100,
+                0L));
+    Optional<CRCInfo> crcInfoOpt =
+        ChecksumReader.getCRCInfo(engine, logSegment.logPath, snapshotVersion, crcSearchLowerBound);
+    if (crcInfoOpt.isPresent()) {
+      CRCInfo crcInfo = crcInfoOpt.get();
+      if (crcInfo.getVersion() == snapshotVersion) {
+        // CRC is related to the desired snapshot version. Load protocol and metadata from CRC.
+        return new Tuple2<>(crcInfo.getProtocol(), crcInfo.getMetadata());
+      }
+      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
+      snapshotHint =
+          Optional.of(
+              new SnapshotHint(crcInfo.getVersion(), crcInfo.getProtocol(), crcInfo.getMetadata()));
+    }
+
     Protocol protocol = null;
     Metadata metadata = null;
 

kernel/kernel-api/src/main/java/io/delta/kernel/internal/util/FileNames.java

@@ -57,6 +57,9 @@ public enum DeltaLogFileType {
   private static final Pattern CLASSIC_CHECKPOINT_FILE_PATTERN =
       Pattern.compile("\\d+\\.checkpoint\\.parquet");
 
+  /** Example: 00000000000000000001.crc */
+  private static final Pattern CHECK_SUM_FILE_PATTERN = Pattern.compile("(\\d+)\\.crc");
+
   /**
    * Examples:
    *
@@ -89,8 +92,8 @@ public static long getFileVersion(Path path) {
       return checkpointVersion(path);
     } else if (isCommitFile(path.getName())) {
       return deltaVersion(path);
-      // } else if (isChecksumFile(path)) {
-      //    checksumVersion(path);
+    } else if (isChecksumFile(path.getName())) {
+      return checksumVersion(path);
     } else {
       throw new IllegalArgumentException(
           String.format("Unexpected file type found in transaction log: %s", path));
@@ -133,6 +136,15 @@ public static String sidecarFile(Path path, String sidecar) {
     return String.format("%s/%s/%s", path.toString(), SIDECAR_DIRECTORY, sidecar);
   }
 
+  /** Returns the path to the checksum file for the given version. */
+  public static Path checksumFile(Path path, long version) {
+    return new Path(path, String.format("%020d.crc", version));
+  }
+
+  public static long checksumVersion(Path path) {
+    return Long.parseLong(path.getName().split("\\.")[0]);
+  }
+
   /**
    * Returns the prefix of all delta log files for the given version.
    *
@@ -211,4 +223,8 @@ public static boolean isCommitFile(String path) {
     return DELTA_FILE_PATTERN.matcher(fileName).matches()
         || UUID_DELTA_FILE_REGEX.matcher(fileName).matches();
   }
+
+  public static boolean isChecksumFile(String checksumFilePath) {
+    return CHECK_SUM_FILE_PATTERN.matcher(new Path(checksumFilePath).getName()).matches();
+  }
 }

kernel/kernel-api/src/main/java/io/delta/kernel/internal/util/InternalUtils.java

@@ -30,9 +30,7 @@
 import java.time.LocalDateTime;
 import java.time.ZoneOffset;
 import java.time.temporal.ChronoUnit;
-import java.util.Collection;
-import java.util.Optional;
-import java.util.Set;
+import java.util.*;
 import java.util.stream.Collectors;
 
 public class InternalUtils {

kernel/kernel-api/src/main/java/io/delta/kernel/utils/FileStatus.java

@@ -75,6 +75,16 @@ public static FileStatus of(String path, long size, long modificationTime) {
     return new FileStatus(path, size, modificationTime);
   }
 
+  /**
+   * Create a {@link FileStatus} with the given path with size and modification time set to 0.
+   *
+   * @param path Fully qualified file path.
+   * @return {@link FileStatus} object
+   */
+  public static FileStatus of(String path) {
+    return new FileStatus(path, 0 /* size */, 0 /* modTime */);
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {