[Kernel-Spark][DSv2] Implement UCManagedSnapshotManager operations using UCCatalogManagedClient

kernel-spark/src/main/java/io/delta/kernel/spark/snapshot/unitycatalog/UCManagedTableSnapshotManager.java

@@ -21,8 +21,12 @@
 import io.delta.kernel.Snapshot;
 import io.delta.kernel.engine.Engine;
 import io.delta.kernel.internal.DeltaHistoryManager;
+import io.delta.kernel.internal.SnapshotImpl;
+import io.delta.kernel.internal.files.ParsedCatalogCommitData;
+import io.delta.kernel.spark.exception.VersionNotFoundException;
 import io.delta.kernel.spark.snapshot.DeltaSnapshotManager;
 import io.delta.kernel.unitycatalog.UCCatalogManagedClient;
+import java.util.List;
 import java.util.Optional;
 
 /**
@@ -57,35 +61,90 @@ public UCManagedTableSnapshotManager(
 
   @Override
   public Snapshot loadLatestSnapshot() {
-    throw new UnsupportedOperationException(
-        "UCManagedTableSnapshotManager.loadLatestSnapshot is not yet implemented");
+    return ucCatalogManagedClient.loadSnapshot(
+        engine,
+        tableId,
+        tablePath,
+        Optional.empty() /* versionOpt */,
+        Optional.empty() /* timestampOpt */);
   }
 
   @Override
   public Snapshot loadSnapshotAt(long version) {
-    throw new UnsupportedOperationException(
-        "UCManagedTableSnapshotManager.loadSnapshotAt is not yet implemented");
+    return ucCatalogManagedClient.loadSnapshot(
+        engine, tableId, tablePath, Optional.of(version), Optional.empty() /* timestampOpt */);
   }
 
+  /**
+   * Finds the active commit at a specific timestamp.
+   *
+   * <p>For UC-managed tables, this loads the latest snapshot and uses {@link
+   * DeltaHistoryManager#getActiveCommitAtTimestamp} to resolve the timestamp to a commit.
+   */
   @Override
   public DeltaHistoryManager.Commit getActiveCommitAtTime(
       long timestampMillis,
       boolean canReturnLastCommit,
       boolean mustBeRecreatable,
       boolean canReturnEarliestCommit) {
-    throw new UnsupportedOperationException(
-        "UCManagedTableSnapshotManager.getActiveCommitAtTime is not yet implemented");
+    SnapshotImpl snapshot = (SnapshotImpl) loadLatestSnapshot();
+    List<ParsedCatalogCommitData> catalogCommits = snapshot.getLogSegment().getAllCatalogCommits();
+    return DeltaHistoryManager.getActiveCommitAtTimestamp(
+        engine,
+        snapshot,
+        snapshot.getLogPath(),
+        timestampMillis,
+        mustBeRecreatable,
+        canReturnLastCommit,
+        canReturnEarliestCommit,
+        catalogCommits);
   }
 
+  /**
+   * Checks if a specific version exists and is accessible.
+   *
+   * <p>For UC-managed tables with catalogManaged, log files may be cleaned up, so we need to use
+   * DeltaHistoryManager to find the earliest available version based on filesystem state.
+   */
   @Override
-  public void checkVersionExists(long version, boolean mustBeRecreatable, boolean allowOutOfRange) {
-    throw new UnsupportedOperationException(
-        "UCManagedTableSnapshotManager.checkVersionExists is not yet implemented");
+  public void checkVersionExists(long version, boolean mustBeRecreatable, boolean allowOutOfRange)
+      throws VersionNotFoundException {
+    // Load latest to get the current version bounds
+    SnapshotImpl snapshot = (SnapshotImpl) loadLatestSnapshot();
+    long latestRatifiedVersion = snapshot.getVersion();
+
+    // Fast path: check upper bound before expensive filesystem operations
+    if ((version > latestRatifiedVersion) && !allowOutOfRange) {
+      throw new VersionNotFoundException(version, 0, latestRatifiedVersion);
+    }
+
+    // Compute earliestRatifiedCommitVersion from catalog commits
+    List<ParsedCatalogCommitData> catalogCommits = snapshot.getLogSegment().getAllCatalogCommits();
+    Optional<Long> earliestRatifiedCommitVersion =
+        catalogCommits.stream().map(ParsedCatalogCommitData::getVersion).min(Long::compare);
+
+    // Use DeltaHistoryManager to find earliest version based on filesystem state
+    long earliestVersion =
+        mustBeRecreatable
+            ? DeltaHistoryManager.getEarliestRecreatableCommit(
+                engine, snapshot.getLogPath(), earliestRatifiedCommitVersion)
+            : DeltaHistoryManager.getEarliestDeltaFile(
+                engine, snapshot.getLogPath(), earliestRatifiedCommitVersion);
+
+    if (version < earliestVersion) {
+      throw new VersionNotFoundException(version, earliestVersion, latestRatifiedVersion);
+    }
   }
 
   @Override
   public CommitRange getTableChanges(Engine engine, long startVersion, Optional<Long> endVersion) {
-    throw new UnsupportedOperationException(
-        "UCManagedTableSnapshotManager.getTableChanges is not yet implemented");
+    return ucCatalogManagedClient.loadCommitRange(
+        engine,
+        tableId,
+        tablePath,
+        Optional.of(startVersion) /* startVersionOpt */,
+        Optional.empty() /* startTimestampOpt */,
+        endVersion /* endVersionOpt */,
+        Optional.empty() /* endTimestampOpt */);
   }
 }