[KERNEL] Read the _last_checkpoint file for latest queries with maxRatifiedVersion (#5529)

allisonport-db · web-flow · commit 7a8c2428fc89 · 2025-11-21T13:02:27.000-08:00
#### Which Delta project/connector is this regarding?  - [ ] Spark - [ ] Standalone - [ ] Flink - [X] Kernel - [ ] Other (fill in here) ## Description Based off of #5443 Review the commits in this PR [here](https://github.com/delta-io/delta/pull/5529/files/d3ac1fbbf6c1b4ba89729375067fddfc822075ca..0783f0d8a99339c3e13c99032d90c3de37ebba96) Adds the logic in log segment construction to correctly use the _last_checkpoint file for latest queries for catalog managed tables. ## How was this patch tested? Adds tests to check we use the _last_checkpoint file. Tests in the previous PR test that we correctly load the right version. ## Does this PR introduce _any_ user-facing changes? No
diff --git a/kernel/kernel-api/src/main/java/io/delta/kernel/internal/snapshot/SnapshotManager.java b/kernel/kernel-api/src/main/java/io/delta/kernel/internal/snapshot/SnapshotManager.java
@@ -200,28 +200,45 @@ private SnapshotImpl createSnapshot(
    * </ol>
    */
   public LogSegment getLogSegmentForVersion(Engine engine, Optional<Long> versionToLoadOpt) {
-    return getLogSegmentForVersion(engine, versionToLoadOpt, Collections.emptyList());
+    return getLogSegmentForVersion(
+        engine,
+        versionToLoadOpt,
+        Collections.emptyList() /* parsedLogDatas */,
+        Optional.empty() /* maxCatalogVersionOpt */);
   }
 
   /**
    * [delta-io/delta#4765]: Right now, we only support sorted and contiguous ratified commit log
    * data.
+   *
+   * @param timeTravelVersionOpt the version to time-travel to for a time-travel query
+   * @param parsedLogDatas the parsed log data from the catalog
+   * @param maxCatalogVersionOpt the maximum version ratified by the catalog for catalog managed
+   *     tables. Empty for file-system managed tables.
    */
   public LogSegment getLogSegmentForVersion(
-      Engine engine, Optional<Long> versionToLoadOpt, List<ParsedLogData> parsedLogDatas) {
+      Engine engine,
+      Optional<Long> timeTravelVersionOpt,
+      List<ParsedLogData> parsedLogDatas,
+      Optional<Long> maxCatalogVersionOpt) {
+    // This is the actual version we want to load. For "latest" (aka non-time-travel) queries for
+    // catalogManaged tables we want to load the maxCatalogVersion
+    final Optional<Long> versionToLoadOpt =
+        timeTravelVersionOpt.isPresent() ? timeTravelVersionOpt : maxCatalogVersionOpt;
     final long versionToLoad = versionToLoadOpt.orElse(Long.MAX_VALUE);
     final String versionToLoadStr = versionToLoadOpt.map(String::valueOf).orElse("latest");
     logger.info("Loading log segment for version {}", versionToLoadStr);
     final long logSegmentBuildingStartTimeMillis = System.currentTimeMillis();
 
-    ////////////////////////////////////////////////////////////////////////////////////////////////
-    // Step 1: Find the latest checkpoint version. If $versionToLoadOpt is empty, use the version //
-    //         referenced by the _LAST_CHECKPOINT file. If $versionToLoad is present, search for  //
-    //         the previous latest complete checkpoint at or before $versionToLoad.               //
-    ////////////////////////////////////////////////////////////////////////////////////////////////
+    ///////////////////////////////////////////////////////////////////////////////////////////
+    // Step 1: Find the latest checkpoint version. If timeTravelVersionOpt is empty, use the //
+    //         version referenced by the _LAST_CHECKPOINT file. If timeTravelVersionOpt is   //
+    //         present, search for the previous latest complete checkpoint at or before the  //
+    //         version to load                                                               //
+    ///////////////////////////////////////////////////////////////////////////////////////////
 
     final Optional<Long> startCheckpointVersionOpt =
-        getStartCheckpointVersion(engine, versionToLoadOpt);
+        getStartCheckpointVersion(engine, timeTravelVersionOpt, maxCatalogVersionOpt);
 
     /////////////////////////////////////////////////////////////////
     // Step 2: Determine the actual version to start listing from. //
@@ -593,39 +610,82 @@ private List<ParsedDeltaData> getAllDeltasAfterCheckpointWithCatalogPriority(
   }
 
   /**
-   * Determine the starting checkpoint version that is at or before `versionToLoadOpt`. If no
-   * `versionToLoadOpt` is provided, will use the checkpoint pointed to by the _last_checkpoint
-   * file.
+   * Determine the starting checkpoint version that is at or before the version to load.
+   *
+   * <p>Version to load: For time-travel queries, this is the time-travel version. For latest
+   * queries on catalog maanged tables, this is the max ratified catalog version. For latest queries
+   * on file-system managed tables, this is the latest available version on the file-system.
+   *
+   * <p>For non-time travel queries we will use the checkpoint pointed to by the _last_checkpoint
+   * file (except for when it is after the maxRatifiedCatalogVersion, in which case we will search
+   * backwards for a checkpoint).
    */
-  private Optional<Long> getStartCheckpointVersion(Engine engine, Optional<Long> versionToLoadOpt) {
-    return versionToLoadOpt
+  private Optional<Long> getStartCheckpointVersion(
+      Engine engine, Optional<Long> timeTravelVersionOpt, Optional<Long> maxCatalogVersionOpt) {
+    // This is a "latest" query, let's try to use the _last_checkpoint file if possible
+    if (!timeTravelVersionOpt.isPresent()) {
+      logger.info("Reading the _last_checkpoint file for 'latest' query");
+      Optional<Long> lastCheckpointFileVersionOpt =
+          new Checkpointer(logPath).readLastCheckpointFile(engine).map(x -> x.version);
+
+      if (!lastCheckpointFileVersionOpt.isPresent()) {
+        logger.info("No _last_checkpoint file found, default to listing from 0");
+        return Optional.empty();
+      }
+
+      long lastCheckpointFileVersion = lastCheckpointFileVersionOpt.get();
+
+      if (!maxCatalogVersionOpt.isPresent()) {
+        // If there is no maxCatalogVersion we don't have to do anything special --> just return
+        return Optional.of(lastCheckpointFileVersion);
+      } else {
+        // When there is a maxCatalogVersion we only want to return the version from the
+        // _last_checkpoint file if it is less than or equal to the maxCatalogVersion. Otherwise,
+        // we should revert to listing backwards from the version to load.
+
+        // This situation is possible due to race conditions. Since fetching the maxCatalogVersion
+        // from the catalog, it is possible that a concurrent writer has committed, published
+        // and checkpointed before this listing code is executed. Thus, it is possible that the
+        // _last_checkpoint file points to a checkpoint later than the maxCatalogVersion.
+        if (lastCheckpointFileVersion <= maxCatalogVersionOpt.get()) {
+          return Optional.of(lastCheckpointFileVersion);
+        }
+        logger.info(
+            "Found checkpoint at version {} in _last_checkpoint file but cannot be used because "
+                + "maxCatalogVersion = {}.",
+            lastCheckpointFileVersion,
+            maxCatalogVersionOpt.get());
+      }
+    }
+
+    long versionToLoad =
+        timeTravelVersionOpt.orElseGet(
+            () ->
+                maxCatalogVersionOpt.orElseThrow(
+                    () ->
+                        new IllegalStateException(
+                            "Impossible state: If timeTravelToVersionOpt and maxCatalogVersion "
+                                + "is empty we should always have returned earlier")));
+    logger.info("Finding last complete checkpoint at or before version {}", versionToLoad);
+    final long startTimeMillis = System.currentTimeMillis();
+    return Checkpointer.findLastCompleteCheckpointBefore(engine, logPath, versionToLoad + 1)
+        .map(checkpointInstance -> checkpointInstance.version)
         .map(
-            versionToLoad -> {
+            checkpointVersion -> {
+              checkArgument(
+                  checkpointVersion <= versionToLoad,
+                  "Last complete checkpoint version %s was not <= targetVersion %s",
+                  checkpointVersion,
+                  versionToLoad);
+
               logger.info(
-                  "Finding last complete checkpoint at or before version {}", versionToLoad);
-              final long startTimeMillis = System.currentTimeMillis();
-              return Checkpointer.findLastCompleteCheckpointBefore(
-                      engine, logPath, versionToLoad + 1)
-                  .map(checkpointInstance -> checkpointInstance.version)
-                  .map(
-                      checkpointVersion -> {
-                        checkArgument(
-                            checkpointVersion <= versionToLoad,
-                            "Last complete checkpoint version %s was not <= targetVersion %s",
-                            checkpointVersion,
-                            versionToLoad);
-
-                        logger.info(
-                            "{}: Took {}ms to find last complete checkpoint <= targetVersion {}",
-                            tablePath,
-                            System.currentTimeMillis() - startTimeMillis,
-                            versionToLoad);
-
-                        return checkpointVersion;
-                      });
-            })
-        .orElseGet(
-            () -> new Checkpointer(logPath).readLastCheckpointFile(engine).map(x -> x.version));
+                  "{}: Took {}ms to find last complete checkpoint <= targetVersion {}",
+                  tablePath,
+                  System.currentTimeMillis() - startTimeMillis,
+                  versionToLoad);
+
+              return checkpointVersion;
+            });
   }
 
   private void logDebugFileStatuses(String varName, List<FileStatus> fileStatuses) {
diff --git a/kernel/kernel-api/src/main/java/io/delta/kernel/internal/table/SnapshotFactory.java b/kernel/kernel-api/src/main/java/io/delta/kernel/internal/table/SnapshotFactory.java
@@ -248,7 +248,8 @@ private Lazy<LogSegment> getLazyLogSegment(
                   .time(
                       () ->
                           new SnapshotManager(tablePath)
-                              .getLogSegmentForVersion(engine, versionToLoad, ctx.logDatas));
+                              .getLogSegmentForVersion(
+                                  engine, versionToLoad, ctx.logDatas, ctx.maxCatalogVersion));
 
           snapshotCtx.setResolvedVersion(logSegment.getVersion());
           snapshotCtx.setCheckpointVersion(logSegment.getCheckpointVersionOpt());
diff --git a/kernel/kernel-api/src/test/scala/io/delta/kernel/internal/SnapshotManagerSuite.scala b/kernel/kernel-api/src/test/scala/io/delta/kernel/internal/SnapshotManagerSuite.scala
@@ -738,6 +738,168 @@ class SnapshotManagerSuite extends AnyFunSuite with MockFileSystemClientUtils {
     assert(exMsg.contains("Missing checkpoint at version 1"))
   }
 
+  /* ------------------- CATALOG MANAGED TABLE TESTS ------------------ */
+
+  test("catalog managed: latest query, we load the maxCatalogVersion even if other deltas exist") {
+    val deltas = deltaFileStatuses(0L to 20)
+    val checkpoints = singularCheckpointFileStatuses(Seq(10))
+
+    val logSegment = snapshotManager.getLogSegmentForVersion(
+      createMockFSListFromEngine(deltas ++ checkpoints),
+      Optional.empty(), // timeTravelVersionOpt
+      Collections.emptyList(), // parsedLogDatas
+      Optional.of(15L) // maxCatalogVersionOpt
+    )
+
+    checkLogSegment(
+      logSegment,
+      expectedVersion = 15,
+      expectedDeltas = deltaFileStatuses(11L to 15),
+      expectedCompactions = Seq.empty,
+      expectedCheckpoints = singularCheckpointFileStatuses(Seq(10)),
+      expectedCheckpointVersion = Some(10),
+      expectedLastCommitTimestamp = 150L)
+  }
+
+  test("catalog managed: latest query, _last_checkpoint does not exist") {
+    val deltas = deltaFileStatuses(0L to 20)
+    val checkpoints = singularCheckpointFileStatuses(Seq(10))
+
+    val logSegment = snapshotManager.getLogSegmentForVersion(
+      createMockFSListFromEngine(deltas ++ checkpoints),
+      Optional.empty(), // timeTravelVersionOpt
+      Collections.emptyList(), // parsedLogDatas
+      Optional.of(20L) // maxCatalogVersionOpt
+    )
+
+    // Should find checkpoint at version 10 by searching backwards from version 20
+    checkLogSegment(
+      logSegment,
+      expectedVersion = 20,
+      expectedDeltas = deltaFileStatuses(11L to 20),
+      expectedCompactions = Seq.empty,
+      expectedCheckpoints = singularCheckpointFileStatuses(Seq(10)),
+      expectedCheckpointVersion = Some(10),
+      expectedLastCommitTimestamp = 200L)
+  }
+
+  test("catalog managed: latest query, when _last_checkpoint exists and " +
+    "is <= maxCatalogVersion we use it") {
+    val deltas = deltaFileStatuses(0L to 30)
+    val checkpoints = singularCheckpointFileStatuses(Seq(10, 20, 25))
+    val lastCheckpointFileStatus = FileStatus.of(s"$logPath/_last_checkpoint", 2, 2)
+    val files = deltas ++ checkpoints ++ Seq(lastCheckpointFileStatus)
+
+    // Create mocked engine that fails if we try to list before the version stored in
+    // _last_checkpoint
+    def listFrom(filePath: String): Seq[FileStatus] = {
+      if (filePath < FileNames.listingPrefix(logPath, 25)) {
+        throw new RuntimeException(
+          s"Listing from before the checkpoint version referenced by _last_checkpoint.")
+      }
+      listFromProvider(files)(filePath)
+    }
+    val mockedEngine = mockEngine(
+      jsonHandler = new MockReadLastCheckpointFileJsonHandler(
+        lastCheckpointFileStatus.getPath,
+        25
+      ), // _last_checkpoint points to version 25
+      fileSystemClient = new MockListFromFileSystemClient(listFrom))
+
+    // Latest query with catalog managed table (maxCatalogVersion = 30)
+    val logSegment = snapshotManager.getLogSegmentForVersion(
+      mockedEngine,
+      Optional.empty(), // timeTravelVersionOpt
+      Collections.emptyList(), // parsedLogDatas
+      Optional.of(30L) // maxCatalogVersionOpt
+    )
+
+    checkLogSegment(
+      logSegment,
+      expectedVersion = 30,
+      expectedDeltas = deltaFileStatuses(26L to 30),
+      expectedCompactions = Seq.empty,
+      expectedCheckpoints = singularCheckpointFileStatuses(Seq(25)),
+      expectedCheckpointVersion = Some(25),
+      expectedLastCommitTimestamp = 300L)
+  }
+
+  test("catalog managed:" +
+    "latest query, ignore _last_checkpoint if it's newer than maxCatalogVersion") {
+    val deltas = deltaFileStatuses(0L to 26)
+    val checkpoints = singularCheckpointFileStatuses(Seq(10, 20, 25))
+    val lastCheckpointFileStatus = FileStatus.of(s"$logPath/_last_checkpoint", 2, 2)
+    val files = deltas ++ checkpoints ++ Seq(lastCheckpointFileStatus)
+
+    // Latest query with catalog managed table where maxCatalogVersion < _last_checkpoint version
+    val logSegment = snapshotManager.getLogSegmentForVersion(
+      mockEngine(
+        jsonHandler = new MockReadLastCheckpointFileJsonHandler(
+          lastCheckpointFileStatus.getPath,
+          25
+        ), // _last_checkpoint points to version 25
+        fileSystemClient = new MockListFromFileSystemClient(listFromProvider(files))),
+      Optional.empty(), // timeTravelVersionOpt
+      Collections.emptyList(), // parsedLogDatas
+      Optional.of(24L) // maxCatalogVersionOpt is 24, which is < 25
+    )
+
+    // Should use checkpoint at version 20 not 25, and should load maxCatalogVersion
+    checkLogSegment(
+      logSegment,
+      expectedVersion = 24,
+      expectedDeltas = deltaFileStatuses(21L to 24),
+      expectedCompactions = Seq.empty,
+      expectedCheckpoints = singularCheckpointFileStatuses(Seq(20)),
+      expectedCheckpointVersion = Some(20),
+      expectedLastCommitTimestamp = 240L)
+  }
+
+  test("catalog managed: time travel query ignores _last_checkpoint") {
+    val deltas = deltaFileStatuses(0L to 30)
+    val checkpoints = singularCheckpointFileStatuses(Seq(10, 20, 25))
+    val lastCheckpointFileStatus = FileStatus.of(s"$logPath/_last_checkpoint", 2, 2)
+    val files = deltas ++ checkpoints ++ Seq(lastCheckpointFileStatus)
+
+    val jsonHandler = new BaseMockJsonHandler {
+      override def readJsonFiles(
+          fileIter: CloseableIterator[FileStatus],
+          physicalSchema: StructType,
+          predicate: Optional[Predicate]): CloseableIterator[ColumnarBatch] = {
+        assert(fileIter.hasNext)
+        if (fileIter.next.getPath == lastCheckpointFileStatus.getPath) {
+          throw new RuntimeException(
+            "We should not be reading the _last_checkpoint file for time-travel queries")
+        } else {
+          throw new RuntimeException("We should not be reading JSON files besides " +
+            "_last_checkpoint during log segment construction")
+        }
+      }
+    }
+
+    // Time travel query with catalog managed table
+    val logSegment = snapshotManager.getLogSegmentForVersion(
+      mockEngine(
+        jsonHandler = jsonHandler,
+        fileSystemClient = new MockListFromFileSystemClient(listFromProvider(files))),
+      Optional.of(15L), // timeTravelVersionOpt = 15
+      Collections.emptyList(), // parsedLogDatas
+      Optional.of(30L) // maxCatalogVersionOpt
+    )
+
+    // Should use checkpoint at version 10 for time travel to version 15
+    checkLogSegment(
+      logSegment,
+      expectedVersion = 15,
+      expectedDeltas = deltaFileStatuses(11L to 15),
+      expectedCompactions = Seq.empty,
+      expectedCheckpoints = singularCheckpointFileStatuses(Seq(10)),
+      expectedCheckpointVersion = Some(10),
+      expectedLastCommitTimestamp = 150L)
+  }
+
+  /* ------------------- Compaction tests ------------------ */
+
   test("One compaction") {
     testWithCompactionsNoCheckpoint(
       deltaVersions = 0L until 5L,
