[Kernel]Incrementally loading domain metadata from CRC (delta-io#4502)

<!--
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.
-->
This PR enhanced Domain metadata loading with CRC - If current CRC not
found and previous found, combined with log replay from current version
until crc's version + 1

## 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.
-->
Adjusted unit tests, by moving the P&M only ones to the shared suite
## 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

Author: huan233usc

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

@@ -165,7 +165,12 @@ public LogReplay(
                 loadTableProtocolAndMetadata(
                     engine, logSegment, newerSnapshotHint, logSegment.getVersion()));
     // Lazy loading of domain metadata only when needed
-    this.activeDomainMetadataMap = new Lazy<>(() -> loadDomainMetadataMap(engine));
+    this.activeDomainMetadataMap =
+        new Lazy<>(
+            () ->
+                loadDomainMetadataMap(engine).entrySet().stream()
+                    .filter(entry -> !entry.getValue().isRemoved())
+                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
   }
 
   /////////////////
@@ -387,18 +392,34 @@ engine, getLogReplayFiles(logSegment), SET_TRANSACTION_READ_SCHEMA, Optional.emp
    */
   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()
+    Optional<CRCInfo> lastSeenCrcInfoOpt = crcInfoContext.getLastSeenCrcInfo();
+    if (!lastSeenCrcInfoOpt.isPresent()
+        || !lastSeenCrcInfoOpt.get().getDomainMetadata().isPresent()) {
+      logger.info("No domain metadata available in CRC info, loading from log");
+      return loadDomainMetadataMapFromLog(engine, Optional.empty());
+    }
+    CRCInfo lastSeenCrcInfo = lastSeenCrcInfoOpt.get();
+    if (lastSeenCrcInfo.getVersion() == logSegment.getVersion()) {
+      return lastSeenCrcInfo.getDomainMetadata().get().stream()
           .collect(Collectors.toMap(DomainMetadata::getDomain, Function.identity()));
     }
-    // TODO https://github.com/delta-io/delta/issues/4454: Incrementally load domain metadata from
-    // CRC when current CRC is not available.
-    // Fall back to loading from the log
-    logger.info("No domain metadata available in CRC info, loading from log");
-    return loadDomainMetadataMapFromLog(engine).entrySet().stream()
-        .filter(entry -> !entry.getValue().isRemoved())
-        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+    Map<String, DomainMetadata> finalDomainMetadataMap =
+        loadDomainMetadataMapFromLog(engine, Optional.of(lastSeenCrcInfo.getVersion() + 1));
+    // Add domains from the CRC that don't exist in the incremental log data
+    // - If a domain is updated to the newer versions or removed, it will exist in
+    // finalDomainMetadataMap, use the one in the map.
+    // - If a domain is only in the CRC file, use the one from CRC.
+    lastSeenCrcInfo
+        .getDomainMetadata()
+        .get()
+        .forEach(
+            domainMetadataInCrc -> {
+              if (!finalDomainMetadataMap.containsKey(domainMetadataInCrc.getDomain())) {
+                finalDomainMetadataMap.put(domainMetadataInCrc.getDomain(), domainMetadataInCrc);
+              }
+            });
+    return finalDomainMetadataMap;
   }
 
   /**
@@ -409,11 +430,15 @@ private Map<String, DomainMetadata> loadDomainMetadataMap(Engine engine) {
    * #loadTableProtocolAndMetadata}.
    *
    * @param engine The engine used to process the log files.
+   * @param minLogVersion The minimum log version to read (inclusive). When provided, only reads log
+   *     files * starting from this version. When not provided, reads the entire log. * For
+   *     incremental loading from crc, this is typically set to (crc version + 1).
    * @return A map where the keys are domain names and the values are the corresponding {@link
    *     DomainMetadata} objects.
    * @throws UncheckedIOException if an I/O error occurs while closing the iterator.
    */
-  private Map<String, DomainMetadata> loadDomainMetadataMapFromLog(Engine engine) {
+  private Map<String, DomainMetadata> loadDomainMetadataMapFromLog(
+      Engine engine, Optional<Long> minLogVersion) {
     try (CloseableIterator<ActionWrapper> reverseIter =
         new ActionsIterator(
             engine,
@@ -422,14 +447,19 @@ private Map<String, DomainMetadata> loadDomainMetadataMapFromLog(Engine engine)
             Optional.empty() /* checkpointPredicate */)) {
       Map<String, DomainMetadata> domainMetadataMap = new HashMap<>();
       while (reverseIter.hasNext()) {
-        final ColumnarBatch columnarBatch = reverseIter.next().getColumnarBatch();
+        final ActionWrapper nextElem = reverseIter.next();
+        final long version = nextElem.getVersion();
+        final ColumnarBatch columnarBatch = nextElem.getColumnarBatch();
         assert (columnarBatch.getSchema().equals(DOMAIN_METADATA_READ_SCHEMA));
 
         final ColumnVector dmVector = columnarBatch.getColumnVector(0);
 
         // We are performing a reverse log replay. This function ensures that only the first
         // encountered domain metadata for each domain is added to the map.
         DomainMetadataUtils.populateDomainMetadataMap(dmVector, domainMetadataMap);
+        if (minLogVersion.isPresent() && minLogVersion.get() == version) {
+          break;
+        }
       }
       return domainMetadataMap;
     } catch (IOException ex) {

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

@@ -145,4 +145,61 @@ trait ChecksumLogReplayMetricsTestBase extends LogReplayBaseSuite {
         readVersion = 10)
     }
   }
+
+  test("checksum not found at the read version, but found at a previous version") {
+    withTableWithCrc { (table, tablePath, engine) =>
+      deleteChecksumFileForTable(tablePath, Seq(10, 11, 5, 6))
+
+      loadSnapshotFieldsCheckMetrics(
+        table,
+        engine,
+        expJsonVersionsRead = Seq(11),
+        expParquetVersionsRead = Seq(10),
+        expParquetReadSetSizes = getExpectedCheckpointReadSize(Seq(1)),
+        expChecksumReadSet = Nil)
+
+      loadSnapshotFieldsCheckMetrics(
+        table,
+        engine,
+        // We find the checksum from crc at version 4, but still read commit files 5 and 6
+        // to find the P&M which could have been updated in version 5 and 6.
+        expJsonVersionsRead = Seq(6, 5),
+        expParquetVersionsRead = Nil,
+        expParquetReadSetSizes = Nil,
+        expChecksumReadSet = Seq(4),
+        readVersion = 6)
+
+      // now try to load version 3 and it should get P&M from checksum files only
+      loadSnapshotFieldsCheckMetrics(
+        table,
+        engine,
+        // We find the checksum from crc at version 3, so shouldn't read anything else
+        expJsonVersionsRead = Nil,
+        expParquetVersionsRead = Nil,
+        expParquetReadSetSizes = Nil,
+        expChecksumReadSet = Seq(3),
+        readVersion = 3)
+    }
+  }
+
+  test(
+    "checksum missing read version, " +
+      "both checksum and checkpoint exist the read version the previous version => use checksum") {
+    withTableWithCrc { (table, tablePath, engine) =>
+      val checkpointVersion = 10
+      val readVersion = checkpointVersion + 1
+      deleteChecksumFileForTable(tablePath, Seq(checkpointVersion + 1))
+
+      // 11.crc, missing, 10.crc and 10.checkpoint.parquet exist.
+      // read 10.crc and 11.json.
+      loadSnapshotFieldsCheckMetrics(
+        table,
+        engine,
+        expJsonVersionsRead = Seq(readVersion),
+        expParquetVersionsRead = Nil,
+        expParquetReadSetSizes = Nil,
+        expChecksumReadSet = Seq(checkpointVersion),
+        readVersion = readVersion)
+    }
+  }
 }

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

@@ -54,26 +54,6 @@ class DomainMetadataCheckSumReplayMetricsSuite extends ChecksumLogReplayMetricsT
     sizes.flatMap(size => Seq(size, size))
   }
 
-  // TODO: Remove to ChecksumLogReplayMetricsTestBase
-  // after incremental CRC loading for domain metadata implemented
-  test("checksum missing at read version and checkpoint version (readVersion - 1) " +
-    " => use checkpoint") {
-    withTableWithCrc { (table, tablePath, engine) =>
-      val checkpointVersion = 10
-      val readVersion = checkpointVersion + 1
-      deleteChecksumFileForTable(tablePath, Seq(readVersion))
-
-      loadSnapshotFieldsCheckMetrics(
-        table,
-        engine,
-        expJsonVersionsRead = Seq(readVersion),
-        expParquetVersionsRead = Seq(checkpointVersion),
-        expParquetReadSetSizes = Seq(1),
-        expChecksumReadSet = Seq(checkpointVersion),
-        readVersion = readVersion)
-    }
-  }
-
   test("read domain metadata fro checksum even if snapshot hint exists") {
     withTableWithCrc { (table, _, engine) =>
       val readVersion = 11

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

@@ -96,66 +96,4 @@ class PandMCheckSumLogReplayMetricsSuite extends ChecksumLogReplayMetricsTestBas
         readVersion = 6)
     }
   }
-
-  // TODO: Move to ChecksumLogReplayMetricsTestBase
-  // after incremental CRC loading for domain metadata implemented
-  test("checksum not found at the read version, but found at a previous version") {
-    withTableWithCrc { (table, tablePath, engine) =>
-      deleteChecksumFileForTable(tablePath, Seq(10, 11, 5, 6))
-
-      loadSnapshotFieldsCheckMetrics(
-        table,
-        engine,
-        // 10.checkpoint found, so use it and combined with 11.crc
-        expJsonVersionsRead = Seq(11),
-        expParquetVersionsRead = Seq(10),
-        expParquetReadSetSizes = getExpectedCheckpointReadSize(Seq(1)),
-        expChecksumReadSet = Nil)
-
-      loadSnapshotFieldsCheckMetrics(
-        table,
-        engine,
-        // We find the checksum from crc at version 4, but still read commit files 5 and 6
-        // to find the P&M which could have been updated in version 5 and 6.
-        expJsonVersionsRead = Seq(6, 5),
-        expParquetVersionsRead = Nil,
-        expParquetReadSetSizes = Nil,
-        expChecksumReadSet = Seq(4),
-        readVersion = 6)
-
-      // now try to load version 3 and it should get P&M from checksum files only
-      loadSnapshotFieldsCheckMetrics(
-        table,
-        engine,
-        // We find the checksum from crc at version 3, so shouldn't read anything else
-        expJsonVersionsRead = Nil,
-        expParquetVersionsRead = Nil,
-        expParquetReadSetSizes = Nil,
-        expChecksumReadSet = Seq(3),
-        readVersion = 3)
-    }
-  }
-
-  // TODO: Move to ChecksumLogReplayMetricsTestBase
-  // after incremental CRC loading for domain metadata implemented
-  test(
-    "checksum missing read version, " +
-      "both checksum and checkpoint exist the read version the previous version => use checksum") {
-    withTableWithCrc { (table, tablePath, engine) =>
-      val checkpointVersion = 10
-      val readVersion = checkpointVersion + 1
-      deleteChecksumFileForTable(tablePath, Seq(checkpointVersion + 1))
-
-      // 11.crc, missing, 10.crc and 10.checkpoint.parquet exist.
-      // read 10.crc and 11.json.
-      loadSnapshotFieldsCheckMetrics(
-        table,
-        engine,
-        expJsonVersionsRead = Seq(readVersion),
-        expParquetVersionsRead = Nil,
-        expParquetReadSetSizes = Nil,
-        expChecksumReadSet = Seq(checkpointVersion),
-        readVersion = readVersion)
-    }
-  }
 }