add checksum complex

Author: huan233usc

kernel/kernel-api/src/main/java/io/delta/kernel/Table.java

@@ -143,4 +143,16 @@ TransactionBuilder createTransactionBuilder(
    */
   void checkpoint(Engine engine, long version)
       throws TableNotFoundException, CheckpointAlreadyExistsException, IOException;
+
+  /**
+   * Checkpoint the table at given version. It writes a single checkpoint file.
+   *
+   * @param engine {@link Engine} instance to use.
+   * @param version Version to checkpoint.
+   * @throws TableNotFoundException if the table is not found
+   * @throws CheckpointAlreadyExistsException if a checkpoint already exists at the given version
+   * @throws IOException for any I/O error.
+   * @since 3.2.0
+   */
+  void checksum(Engine engine, long version) throws TableNotFoundException, IOException;
 }

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

@@ -40,7 +40,8 @@ enum PostCommitHookType {
      * transaction commits. This operation has a minimal latency with no requirement of reading
      * previous checkpoint or logs.
      */
-    CHECKSUM_SIMPLE
+    CHECKSUM_SIMPLE,
+    CHECKSUM_FULL
   }
 
   /** Invokes the post commit operation whose implementation must be thread safe. */

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

@@ -134,6 +134,12 @@ public void checkpoint(Engine engine, long version)
     snapshotManager.checkpoint(engine, clock, version);
   }
 
+  @Override
+  public void checksum(Engine engine, long version)
+      throws TableNotFoundException, CheckpointAlreadyExistsException, IOException {
+    snapshotManager.checksum(engine, version);
+  }
+
   @Override
   public TransactionBuilder createTransactionBuilder(
       Engine engine, String engineInfo, Operation operation) {

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

@@ -33,6 +33,7 @@
 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.ChecksumFullHook;
 import io.delta.kernel.internal.hook.ChecksumSimpleHook;
 import io.delta.kernel.internal.metrics.TransactionMetrics;
 import io.delta.kernel.internal.metrics.TransactionReportImpl;
@@ -364,8 +365,14 @@ private TransactionCommitResult doCommit(
         postCommitHooks.add(new CheckpointHook(dataPath, commitAsVersion));
       }
 
-      buildPostCommitCrcInfo(commitAsVersion, transactionMetrics.captureTransactionMetricsResult())
-          .ifPresent(crcInfo -> postCommitHooks.add(new ChecksumSimpleHook(crcInfo, logPath)));
+      Optional<CRCInfo> postCommitCrcInfo =
+          buildPostCommitCrcInfo(
+              commitAsVersion, transactionMetrics.captureTransactionMetricsResult());
+      if (postCommitCrcInfo.isPresent()) {
+        postCommitHooks.add(new ChecksumSimpleHook(postCommitCrcInfo.get(), logPath));
+      } else {
+        postCommitHooks.add(new ChecksumFullHook(dataPath, commitAsVersion));
+      }
 
       return new TransactionCommitResult(commitAsVersion, postCommitHooks);
     } catch (FileAlreadyExistsException e) {

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

@@ -51,9 +51,7 @@ public void writeCheckSum(Engine engine, CRCInfo crcInfo) throws IOException {
           engine
               .getJsonHandler()
               .writeJsonFileAtomically(
-                  newChecksumPath.toString(),
-                  singletonCloseableIterator(toRow(crcInfo)),
-                  false /* overwrite */);
+                  newChecksumPath.toString(), singletonCloseableIterator(toRow(crcInfo)), false);
           logger.info("Write checksum file `{}` succeeds", newChecksumPath);
           return null;
         },

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

@@ -0,0 +1,43 @@
+/*
+ * 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 io.delta.kernel.Table;
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.hook.PostCommitHook;
+import io.delta.kernel.internal.fs.Path;
+import java.io.IOException;
+
+public class ChecksumFullHook implements PostCommitHook {
+
+  private final Path tablePath;
+  private final long checksumVersion;
+
+  public ChecksumFullHook(Path tablePath, long checksumVersion) {
+    this.tablePath = tablePath;
+    this.checksumVersion = checksumVersion;
+  }
+
+  @Override
+  public void threadSafeInvoke(Engine engine) throws IOException {
+    Table.forPath(engine, tablePath.toString()).checksum(engine, checksumVersion);
+  }
+
+  @Override
+  public PostCommitHookType getType() {
+    return PostCommitHookType.CHECKPOINT;
+  }
+}

kernel/kernel-api/src/main/java/io/delta/kernel/internal/snapshot/SnapshotManager.java

@@ -20,6 +20,7 @@
 import static io.delta.kernel.internal.TableConfig.EXPIRED_LOG_CLEANUP_ENABLED;
 import static io.delta.kernel.internal.TableConfig.LOG_RETENTION;
 import static io.delta.kernel.internal.TableFeatures.validateWriteSupportedTable;
+import static io.delta.kernel.internal.actions.SingleAction.CHECKPOINT_SCHEMA;
 import static io.delta.kernel.internal.replay.LogReplayUtils.assertLogFilesBelongToTable;
 import static io.delta.kernel.internal.snapshot.MetadataCleanup.cleanupExpiredLogs;
 import static io.delta.kernel.internal.util.Preconditions.checkArgument;
@@ -31,9 +32,12 @@
 import io.delta.kernel.exceptions.InvalidTableException;
 import io.delta.kernel.exceptions.TableNotFoundException;
 import io.delta.kernel.internal.*;
+import io.delta.kernel.internal.actions.AddFile;
 import io.delta.kernel.internal.actions.Metadata;
 import io.delta.kernel.internal.annotation.VisibleForTesting;
 import io.delta.kernel.internal.checkpoints.*;
+import io.delta.kernel.internal.checksum.CRCInfo;
+import io.delta.kernel.internal.checksum.ChecksumWriter;
 import io.delta.kernel.internal.fs.Path;
 import io.delta.kernel.internal.lang.ListUtils;
 import io.delta.kernel.internal.metrics.SnapshotQueryContext;
@@ -48,6 +52,7 @@
 import java.nio.file.FileAlreadyExistsException;
 import java.util.*;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.LongAdder;
 import java.util.stream.Collectors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -207,6 +212,48 @@ public void checkpoint(Engine engine, Clock clock, long version)
     }
   }
 
+  public void checksum(Engine engine, long version) throws IOException {
+    ChecksumWriter checksumWriter = new ChecksumWriter(logPath);
+    SnapshotImpl snapshot =
+        (SnapshotImpl)
+            getSnapshotAt(
+                engine,
+                version,
+                SnapshotQueryContext.forVersionSnapshot(tablePath.toString(), version));
+    if (snapshot.getCurrentCrcInfo().isPresent()) {
+      checksumWriter.writeCheckSum(engine, snapshot.getCurrentCrcInfo().get());
+      return;
+    }
+
+    // TODO: Optimize using crc after https://github.com/delta-io/delta/pull/4112
+    LongAdder tableSizeByte = new LongAdder();
+    LongAdder fileCount = new LongAdder();
+    CreateCheckpointIterator currentTableStateIterator =
+        snapshot.getCreateCheckpointIterator(engine);
+    currentTableStateIterator.forEachRemaining(
+        batch ->
+            batch
+                .getRows()
+                .forEachRemaining(
+                    // TODO: Collect domain metadata, ict etc.
+                    row -> {
+                      if (!row.isNullAt(CHECKPOINT_SCHEMA.indexOf("add"))) {
+                        tableSizeByte.add(
+                            new AddFile(row.getStruct(CHECKPOINT_SCHEMA.indexOf("add"))).getSize());
+                        fileCount.add(1);
+                      }
+                    }));
+    checksumWriter.writeCheckSum(
+        engine,
+        new CRCInfo(
+            version,
+            snapshot.getMetadata(),
+            snapshot.getProtocol(),
+            tableSizeByte.longValue(),
+            fileCount.longValue(),
+            Optional.empty()));
+  }
+
   ////////////////////
   // Helper Methods //
   ////////////////////

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/checksum/ChecksumWriterSuite.scala

@@ -59,7 +59,8 @@ class ChecksumWriterSuite extends AnyFunSuite with MockEngineUtils {
 
       checksumWriter.writeCheckSum(
         mockEngine(jsonHandler = jsonHandler),
-        new CRCInfo(version, metadata, protocol, tableSizeBytes, numFiles, txn)
+        new CRCInfo(version, metadata, protocol, tableSizeBytes, numFiles, txn),
+        false
       )
 
       verifyChecksumFile(jsonHandler, version)

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/ChecksumSimpleComparisonSuite.scala

@@ -115,6 +115,8 @@ class ChecksumSimpleComparisonSuite extends AnyFunSuite with TestUtils {
     val kernelCrc = readCrcInfo(engine, kernelTablePath, version)
 
     assertCrcInfoEquals(sparkCrc, kernelCrc)
+    var checksumFullCrc = readCrcInfoFromChecksumFull(engine, kernelTablePath, version)
+    assertCrcInfoEquals(sparkCrc, checksumFullCrc)
   }
 
   private def readCrcInfo(engine: Engine, path: String, version: Long): CRCInfo = {
@@ -123,6 +125,17 @@ class ChecksumSimpleComparisonSuite extends AnyFunSuite with TestUtils {
       .orElseThrow(() => new IllegalStateException(s"CRC info not found for version $version"))
   }
 
+  private def readCrcInfoFromChecksumFull(engine: Engine, path: String, version: Long): CRCInfo = {
+//    assert(Files.deleteIfExists(buildCrcPath(path, version)))
+    defaultEngine.getFileSystemClient.delete(buildCrcPath(path, version).toString)
+    Table
+      .forPath(engine, path)
+      .checksum(engine, version);
+    ChecksumReader
+      .getCRCInfo(engine, new Path(s"$path/_delta_log"), version, version)
+      .orElseThrow(() => new IllegalStateException(s"CRC info not found for version $version"))
+  }
+
   private def buildCrcPath(basePath: String, version: Long): java.nio.file.Path = {
     new File(f"$basePath/_delta_log/$version%020d.crc").toPath
   }