[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,16 +29,19 @@
 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.tablefeatures.TableFeatures;
 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;
@@ -342,6 +345,7 @@ private TransactionCommitResult doCommit(
                     dataAndMetadataActions.map(
                         action -> {
                           transactionMetrics.totalActionsCounter.increment();
+                          // TODO: handle RemoveFiles.
                           if (!action.isNullAt(ADD_FILE_ORDINAL)) {
                             transactionMetrics.addFilesCounter.increment();
                             transactionMetrics.addFilesSizeInBytesCounter.increment(
@@ -362,6 +366,10 @@ private TransactionCommitResult doCommit(
         postCommitHooks.add(new CheckpointHook(dataPath, commitAsVersion));
       }
 
+      buildPostCommitCrcInfoIfCurrentCrcAvailable(
+              commitAsVersion, transactionMetrics.captureTransactionMetricsResult())
+          .ifPresent(crcInfo -> postCommitHooks.add(new ChecksumSimpleHook(crcInfo, logPath)));
+
       return new TransactionCommitResult(commitAsVersion, postCommitHooks);
     } catch (FileAlreadyExistsException e) {
       throw e;
@@ -437,6 +445,36 @@ private void recordTransactionReport(
     engine.getMetricsReporters().forEach(reporter -> reporter.report(transactionReport));
   }
 
+  private Optional<CRCInfo> buildPostCommitCrcInfoIfCurrentCrcAvailable(
+      long commitAtVersion, TransactionMetricsResult metricsResult) {
+    if (isNewTable) {
+      return Optional.of(
+          new CRCInfo(
+              commitAtVersion,
+              metadata,
+              protocol,
+              metricsResult.getTotalAddFilesSizeInBytes(),
+              metricsResult.getNumAddFiles(),
+              Optional.of(txnId.toString())));
+    }
+
+    return readSnapshot
+        .getCurrentCrcInfo()
+        // in the case of a conflicting txn and successful retry the readSnapshot may not be
+        // commitVersion - 1
+        .filter(lastCrcInfo -> commitAtVersion == lastCrcInfo.getVersion() + 1)
+        .map(
+            lastCrcInfo ->
+                new CRCInfo(
+                    commitAtVersion,
+                    metadata,
+                    protocol,
+                    // TODO: handle RemoveFiles for calculating table size and num of files.
+                    lastCrcInfo.getTableSizeBytes() + metricsResult.getTotalAddFilesSizeInBytes(),
+                    lastCrcInfo.getNumFiles() + metricsResult.getNumAddFiles(),
+                    Optional.of(txnId.toString())));
+  }
+
   /**
    * 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/checksum/CRCInfo.java

@@ -0,0 +1,164 @@
+/*
+ * 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.util.Preconditions.checkArgument;
+import static java.util.Objects.requireNonNull;
+
+import io.delta.kernel.data.ColumnVector;
+import io.delta.kernel.data.ColumnarBatch;
+import io.delta.kernel.data.Row;
+import io.delta.kernel.internal.actions.Metadata;
+import io.delta.kernel.internal.actions.Protocol;
+import io.delta.kernel.internal.data.GenericRow;
+import io.delta.kernel.internal.util.InternalUtils;
+import io.delta.kernel.types.LongType;
+import io.delta.kernel.types.StringType;
+import io.delta.kernel.types.StructType;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CRCInfo {
+  private static final Logger logger = LoggerFactory.getLogger(CRCInfo.class);
+
+  // 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 static final StructType CRC_FILE_SCHEMA =
+      new StructType()
+          .add(TABLE_SIZE_BYTES, LongType.LONG)
+          .add(NUM_FILES, LongType.LONG)
+          .add(NUM_METADATA, LongType.LONG)
+          .add(NUM_PROTOCOL, LongType.LONG)
+          .add(METADATA, Metadata.FULL_SCHEMA)
+          .add(PROTOCOL, Protocol.FULL_SCHEMA)
+          .add(TXN_ID, StringType.STRING, /*nullable*/ true);
+
+  public static Optional<CRCInfo> fromColumnarBatch(
+      long version, ColumnarBatch batch, int rowId, String crcFilePath) {
+    // Read required fields.
+    Protocol protocol =
+        Protocol.fromColumnVector(batch.getColumnVector(getSchemaIndex(PROTOCOL)), rowId);
+    Metadata metadata =
+        Metadata.fromColumnVector(batch.getColumnVector(getSchemaIndex(METADATA)), rowId);
+    long tableSizeBytes =
+        InternalUtils.requireNonNull(
+                batch.getColumnVector(getSchemaIndex(TABLE_SIZE_BYTES)), rowId, TABLE_SIZE_BYTES)
+            .getLong(rowId);
+    long numFiles =
+        InternalUtils.requireNonNull(
+                batch.getColumnVector(getSchemaIndex(NUM_FILES)), rowId, NUM_FILES)
+            .getLong(rowId);
+
+    // Read optional fields
+    ColumnVector txnIdColumnVector = batch.getColumnVector(getSchemaIndex(TXN_ID));
+    Optional<String> txnId =
+        txnIdColumnVector.isNullAt(rowId)
+            ? Optional.empty()
+            : Optional.of(txnIdColumnVector.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, tableSizeBytes, numFiles, txnId));
+  }
+
+  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;
+
+  public CRCInfo(
+      long version,
+      Metadata metadata,
+      Protocol protocol,
+      long tableSizeBytes,
+      long numFiles,
+      Optional<String> txnId) {
+    checkArgument(tableSizeBytes >= 0);
+    checkArgument(numFiles >= 0);
+    this.version = version;
+    this.metadata = requireNonNull(metadata);
+    this.protocol = requireNonNull(protocol);
+    this.tableSizeBytes = tableSizeBytes;
+    this.numFiles = numFiles;
+    this.txnId = requireNonNull(txnId);
+  }
+
+  /** 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;
+  }
+
+  public long getNumFiles() {
+    return numFiles;
+  }
+
+  public long getTableSizeBytes() {
+    return tableSizeBytes;
+  }
+
+  public Optional<String> getTxnId() {
+    return txnId;
+  }
+
+  /**
+   * Encode as a {@link Row} object with the schema {@link CRCInfo#CRC_FILE_SCHEMA}.
+   *
+   * @return {@link Row} object with the schema {@link CRCInfo#CRC_FILE_SCHEMA}
+   */
+  public Row toRow() {
+    Map<Integer, Object> values = new HashMap<>();
+    // Add required fields
+    values.put(getSchemaIndex(TABLE_SIZE_BYTES), tableSizeBytes);
+    values.put(getSchemaIndex(NUM_FILES), numFiles);
+    values.put(getSchemaIndex(NUM_METADATA), 1L);
+    values.put(getSchemaIndex(NUM_PROTOCOL), 1L);
+    values.put(getSchemaIndex(METADATA), metadata.toRow());
+    values.put(getSchemaIndex(PROTOCOL), protocol.toRow());
+
+    // Add optional fields
+    txnId.ifPresent(txn -> values.put(getSchemaIndex(TXN_ID), txn));
+    return new GenericRow(CRC_FILE_SCHEMA, values);
+  }
+
+  private static int getSchemaIndex(String fieldName) {
+    return CRC_FILE_SCHEMA.indexOf(fieldName);
+  }
+}

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,
+                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

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

@@ -0,0 +1,58 @@
+/*
+ * 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.util.Utils.singletonCloseableIterator;
+import static java.util.Objects.requireNonNull;
+
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.internal.fs.Path;
+import io.delta.kernel.internal.util.FileNames;
+import java.io.IOException;
+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;
+
+  public ChecksumWriter(Path logPath) {
+    this.logPath = requireNonNull(logPath);
+  }
+
+  /** Writes a checksum file */
+  public void writeCheckSum(Engine engine, CRCInfo crcInfo) throws IOException {
+    Path newChecksumPath = FileNames.checksumFile(logPath, crcInfo.getVersion());
+    logger.info("Writing checksum file to path: {}", newChecksumPath);
+    wrapEngineExceptionThrowsIO(
+        () -> {
+          engine
+              .getJsonHandler()
+              .writeJsonFileAtomically(
+                  newChecksumPath.toString(),
+                  singletonCloseableIterator(crcInfo.toRow()),
+                  false /* overwrite */);
+          logger.info("Write checksum file `{}` succeeds", newChecksumPath);
+          return null;
+        },
+        "Write checksum file `%s`",
+        newChecksumPath);
+  }
+}

kernel/kernel-api/src/main/java/io/delta/kernel/internal/hook/ChecksumSimpleHook.java

@@ -0,0 +1,53 @@
+/*
+ * 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.hook;
+
+import static io.delta.kernel.internal.util.Preconditions.checkArgument;
+import static java.util.Objects.requireNonNull;
+
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.hook.PostCommitHook;
+import io.delta.kernel.internal.checksum.CRCInfo;
+import io.delta.kernel.internal.checksum.ChecksumWriter;
+import io.delta.kernel.internal.fs.Path;
+import java.io.IOException;
+
+/**
+ * A post-commit hook that writes a new checksum file at the version committed by the transaction.
+ * This hook performs a simple checksum operation without requiring previous checkpoint or log
+ * reading.
+ */
+public class ChecksumSimpleHook implements PostCommitHook {
+
+  private final CRCInfo crcInfo;
+  private final Path logPath;
+
+  public ChecksumSimpleHook(CRCInfo crcInfo, Path logPath) {
+    this.crcInfo = requireNonNull(crcInfo);
+    this.logPath = requireNonNull(logPath);
+  }
+
+  @Override
+  public void threadSafeInvoke(Engine engine) throws IOException {
+    checkArgument(engine != null);
+    new ChecksumWriter(logPath).writeCheckSum(engine, crcInfo);
+  }
+
+  @Override
+  public PostCommitHookType getType() {
+    return PostCommitHookType.CHECKSUM_SIMPLE;
+  }
+}