[Kernel]Add simple crc post commit for incremental crc writing.

kernel/kernel-api/src/main/java/io/delta/kernel/hook/PostCommitHook.java

@@ -33,7 +33,14 @@ enum PostCommitHookType {
      * perform this operation, reading previous checkpoint + logs is required to construct a new
      * checkpoint, with latency scaling based on log size (typically seconds to minutes).
      */
-    CHECKPOINT
+    CHECKPOINT,
+    /**
+     * Writes a checksum file at the version committed by the transaction. This hook is present when
+     * all required table statistics(e.g. table size) for checksum file are known when a transaction
+     * commits. This operation has a minimal latency with no requirement of reading previous
+     * checkpoint or logs.
+     */
+    CHECKSUM_SIMPLE
   }
 
   /** Invokes the post commit operation whose implementation must be thread safe. */

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

@@ -25,10 +25,10 @@
 import io.delta.kernel.internal.actions.DomainMetadata;
 import io.delta.kernel.internal.actions.Metadata;
 import io.delta.kernel.internal.actions.Protocol;
+import io.delta.kernel.internal.checksum.CRCInfo;
 import io.delta.kernel.internal.fs.Path;
 import io.delta.kernel.internal.metrics.SnapshotQueryContext;
 import io.delta.kernel.internal.metrics.SnapshotReportImpl;
-import io.delta.kernel.internal.replay.CRCInfo;
 import io.delta.kernel.internal.replay.CreateCheckpointIterator;
 import io.delta.kernel.internal.replay.LogReplay;
 import io.delta.kernel.internal.snapshot.LogSegment;

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

@@ -29,15 +29,18 @@
 import io.delta.kernel.expressions.Column;
 import io.delta.kernel.hook.PostCommitHook;
 import io.delta.kernel.internal.actions.*;
+import io.delta.kernel.internal.checksum.CRCInfo;
 import io.delta.kernel.internal.data.TransactionStateRow;
 import io.delta.kernel.internal.fs.Path;
 import io.delta.kernel.internal.hook.CheckpointHook;
+import io.delta.kernel.internal.hook.ChecksumSimpleHook;
 import io.delta.kernel.internal.metrics.TransactionMetrics;
 import io.delta.kernel.internal.metrics.TransactionReportImpl;
 import io.delta.kernel.internal.replay.ConflictChecker;
 import io.delta.kernel.internal.replay.ConflictChecker.TransactionRebaseState;
 import io.delta.kernel.internal.rowtracking.RowTracking;
 import io.delta.kernel.internal.util.*;
+import io.delta.kernel.metrics.TransactionMetricsResult;
 import io.delta.kernel.metrics.TransactionReport;
 import io.delta.kernel.types.StructType;
 import io.delta.kernel.utils.CloseableIterable;
@@ -361,6 +364,9 @@ private TransactionCommitResult doCommit(
         postCommitHooks.add(new CheckpointHook(dataPath, commitAsVersion));
       }
 
+      buildPostCommitCrcInfo(commitAsVersion, transactionMetrics.captureTransactionMetricsResult())
+          .ifPresent(crcInfo -> postCommitHooks.add(new ChecksumSimpleHook(crcInfo, logPath)));
+
       return new TransactionCommitResult(commitAsVersion, postCommitHooks);
     } catch (FileAlreadyExistsException e) {
       throw e;
@@ -436,6 +442,53 @@ private void recordTransactionReport(
     engine.getMetricsReporters().forEach(reporter -> reporter.report(transactionReport));
   }
 
+  private Optional<CRCInfo> buildPostCommitCrcInfo(
+      long commitAtVersion, TransactionMetricsResult metricsResult) {
+    // Create table
+    if (isNewTable) {
+      return Optional.of(
+          new CRCInfo(
+              commitAtVersion,
+              metadata,
+              protocol,
+              metricsResult.getTotalAddFilesSizeInBytes(),
+              metricsResult.getNumAddFiles(),
+              Optional.of(txnId.toString())));
+    }
+    // We cannot compute the table statistic if the crc info of commitAtVersion is missing
+    if (!readSnapshot.getCurrentCrcInfo().isPresent()
+        || commitAtVersion != readSnapshot.getCurrentCrcInfo().get().getVersion() + 1) {
+      return Optional.empty();
+    }
+
+    return calculateNewCrcInfo(commitAtVersion, metricsResult);
+  }
+
+  private Optional<CRCInfo> calculateNewCrcInfo(
+      long commitAtVersion, TransactionMetricsResult metricsResult) {
+    CRCInfo lastCrcInfo =
+        readSnapshot
+            .getCurrentCrcInfo()
+            .orElseThrow(() -> new IllegalStateException("CRC info must be present at this point"));
+
+    return Optional.of(
+        new CRCInfo(
+            commitAtVersion,
+            metadata,
+            protocol,
+            calculateNewTableSize(lastCrcInfo, metricsResult),
+            calculateNewFileCount(lastCrcInfo, metricsResult),
+            Optional.of(txnId.toString())));
+  }
+
+  private long calculateNewTableSize(CRCInfo lastCrcInfo, TransactionMetricsResult metricsResult) {
+    return lastCrcInfo.getTableSizeBytes() + metricsResult.getTotalAddFilesSizeInBytes();
+  }
+
+  private long calculateNewFileCount(CRCInfo lastCrcInfo, TransactionMetricsResult metricsResult) {
+    return lastCrcInfo.getNumFiles() + metricsResult.getNumAddFiles();
+  }
+
   /**
    * Get the part of the schema of the table that needs the statistics to be collected per file.
    *

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

@@ -13,14 +13,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.delta.kernel.internal.replay;
+package io.delta.kernel.internal.checksum;
 
+import static io.delta.kernel.internal.checksum.ChecksumUtils.CRC_FILE_SCHEMA;
 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;
@@ -30,31 +30,46 @@ public class CRCInfo {
 
   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 protocol =
+        Protocol.fromColumnVector(
+            batch.getColumnVector(CRC_FILE_SCHEMA.indexOf("protocol")), rowId);
+    Metadata metadata =
+        Metadata.fromColumnVector(
+            batch.getColumnVector(CRC_FILE_SCHEMA.indexOf("metadata")), rowId);
+    long tableSizeBytes =
+        batch.getColumnVector(CRC_FILE_SCHEMA.indexOf("tableSizeBytes")).getLong(rowId);
+    long numFiles = batch.getColumnVector(CRC_FILE_SCHEMA.indexOf("numFiles")).getLong(rowId);
+    Optional<String> txnId =
+        Optional.ofNullable(
+            batch.getColumnVector(CRC_FILE_SCHEMA.indexOf("txnId")).getString(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));
+    return Optional.of(new CRCInfo(version, metadata, protocol, tableSizeBytes, numFiles, txnId));
   }
 
-  // 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;
+  private final long tableSizeBytes;
+  private final long numFiles;
+  private final Optional<String> txnId;
 
-  protected CRCInfo(long version, Metadata metadata, Protocol protocol) {
+  public CRCInfo(
+      long version,
+      Metadata metadata,
+      Protocol protocol,
+      long tableSizeBytes,
+      long numFiles,
+      Optional<String> txnId) {
     this.version = version;
     this.metadata = requireNonNull(metadata);
     this.protocol = requireNonNull(protocol);
+    this.tableSizeBytes = tableSizeBytes;
+    this.numFiles = numFiles;
+    this.txnId = txnId;
   }
 
   /** The version of the Delta table that this CRCInfo represents. */
@@ -71,4 +86,16 @@ public Metadata getMetadata() {
   public Protocol getProtocol() {
     return protocol;
   }
+
+  public long getNumFiles() {
+    return numFiles;
+  }
+
+  public long getTableSizeBytes() {
+    return tableSizeBytes;
+  }
+
+  public Optional<String> getTxnId() {
+    return txnId;
+  }
 }

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

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.delta.kernel.internal.replay;
+package io.delta.kernel.internal.checksum;
 
 import static io.delta.kernel.internal.util.FileNames.*;
 import static io.delta.kernel.internal.util.Utils.singletonCloseableIterator;
@@ -96,7 +96,7 @@ private static Optional<CRCInfo> readChecksumFile(Engine engine, Path filePath)
             .getJsonHandler()
             .readJsonFiles(
                 singletonCloseableIterator(FileStatus.of(filePath.toString())),
-                CRCInfo.FULL_SCHEMA,
+                ChecksumUtils.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

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

@@ -0,0 +1,38 @@
+/*
+ * 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.checksum;
+
+import io.delta.kernel.internal.actions.Metadata;
+import io.delta.kernel.internal.actions.Protocol;
+import io.delta.kernel.types.LongType;
+import io.delta.kernel.types.StringType;
+import io.delta.kernel.types.StructType;
+
+/** Helper class for shared components in checksum file read and write */
+public final class ChecksumUtils {
+
+  public static StructType CRC_FILE_SCHEMA =
+      new StructType()
+          .add("tableSizeBytes", LongType.LONG)
+          .add("numFiles", LongType.LONG)
+          .add("numMetadata", LongType.LONG)
+          .add("numProtocol", LongType.LONG)
+          .add("metadata", Metadata.FULL_SCHEMA)
+          .add("protocol", Protocol.FULL_SCHEMA)
+          .add("txnId", StringType.STRING, /*nullable*/ true);
+
+  private ChecksumUtils() {}
+}

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

@@ -0,0 +1,90 @@
+/*
+ * 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.checksum;
+
+import static io.delta.kernel.internal.DeltaErrors.wrapEngineExceptionThrowsIO;
+import static io.delta.kernel.internal.checksum.ChecksumUtils.CRC_FILE_SCHEMA;
+import static io.delta.kernel.internal.util.Utils.singletonCloseableIterator;
+import static java.util.Objects.requireNonNull;
+
+import io.delta.kernel.data.Row;
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.internal.data.GenericRow;
+import io.delta.kernel.internal.fs.Path;
+import io.delta.kernel.internal.util.FileNames;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Writers for writing checksum files from a snapshot */
+public class ChecksumWriter {
+
+  private static final Logger logger = LoggerFactory.getLogger(ChecksumWriter.class);
+
+  private final Path logPath;
+  // Constants for schema field names
+  private static final String TABLE_SIZE_BYTES = "tableSizeBytes";
+  private static final String NUM_FILES = "numFiles";
+  private static final String NUM_METADATA = "numMetadata";
+  private static final String NUM_PROTOCOL = "numProtocol";
+  private static final String METADATA = "metadata";
+  private static final String PROTOCOL = "protocol";
+  private static final String TXN_ID = "txnId";
+
+  public ChecksumWriter(Path logPath) {
+    this.logPath = requireNonNull(logPath);
+  }
+
+  /** Writes a checksum file */
+  public void writeCheckSum(Engine engine, CRCInfo crcInfo) throws IOException {
+    // No sufficient information to write checksum file.
+    Path newChecksumPath = FileNames.checksumFile(logPath, crcInfo.getVersion());
+    logger.debug("Writing checksum file to path: {}", newChecksumPath);
+    wrapEngineExceptionThrowsIO(
+        () -> {
+          engine
+              .getJsonHandler()
+              .writeJsonFileAtomically(
+                  newChecksumPath.toString(),
+                  singletonCloseableIterator(buildCheckSumRow(crcInfo)),
+                  false /* overwrite */);
+          return null;
+        },
+        "Write checksum file `%s`",
+        newChecksumPath);
+  }
+
+  private Row buildCheckSumRow(CRCInfo crcInfo) {
+    Map<Integer, Object> values = new HashMap<>();
+    // Add required fields
+    values.put(getSchemaIndex(TABLE_SIZE_BYTES), crcInfo.getTableSizeBytes());
+    values.put(getSchemaIndex(NUM_FILES), crcInfo.getNumFiles());
+    values.put(getSchemaIndex(NUM_METADATA), 1L);
+    values.put(getSchemaIndex(NUM_PROTOCOL), 1L);
+    values.put(getSchemaIndex(METADATA), crcInfo.getMetadata().toRow());
+    values.put(getSchemaIndex(PROTOCOL), crcInfo.getProtocol().toRow());
+
+    // Add optional fields
+    crcInfo.getTxnId().ifPresent(txn -> values.put(getSchemaIndex(TXN_ID), txn));
+    return new GenericRow(CRC_FILE_SCHEMA, values);
+  }
+
+  private int getSchemaIndex(String fieldName) {
+    return CRC_FILE_SCHEMA.indexOf(fieldName);
+  }
+}