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

build.sbt

@@ -462,6 +462,7 @@ lazy val sparkV1Filtered = (project in file("spark-v1-filtered"))
 lazy val sparkV2 = (project in file("kernel-spark"))
   .dependsOn(sparkV1Filtered)
   .dependsOn(kernelDefaults)
+  .dependsOn(kernelUnityCatalog % "compile->compile;test->test")
   .dependsOn(goldenTables % "test")
   .settings(
     name := "delta-spark-v2",

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

@@ -0,0 +1,135 @@
+/*
+ * 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.spark.snapshot.unitycatalog;
+
+import static java.util.Objects.requireNonNull;
+
+import io.delta.kernel.CommitRange;
+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;
+
+/**
+ * Snapshot manager for Unity Catalog managed tables.
+ *
+ * <p>Used for tables with the catalog-managed commit feature enabled. Unity Catalog serves as the
+ * source of truth for the table's commit history.
+ */
+public class UCManagedTableSnapshotManager implements DeltaSnapshotManager {
+
+  private final UCCatalogManagedClient ucCatalogManagedClient;
+  private final String tableId;
+  private final String tablePath;
+  private final Engine engine;
+
+  /**
+   * Creates a new UCManagedTableSnapshotManager.
+   *
+   * @param ucCatalogManagedClient the UC client for catalog-managed operations
+   * @param tableInfo the UC table information (tableId, tablePath, etc.)
+   * @param engine the Kernel engine for table operations
+   */
+  public UCManagedTableSnapshotManager(
+      UCCatalogManagedClient ucCatalogManagedClient, UCTableInfo tableInfo, Engine engine) {
+    this.ucCatalogManagedClient =
+        requireNonNull(ucCatalogManagedClient, "ucCatalogManagedClient is null");
+    requireNonNull(tableInfo, "tableInfo is null");
+    this.tableId = tableInfo.getTableId();
+    this.tablePath = tableInfo.getTablePath();
+    this.engine = requireNonNull(engine, "engine is null");
+  }
+
+  @Override
+  public Snapshot loadLatestSnapshot() {
+    return ucCatalogManagedClient.loadSnapshot(
+        engine,
+        tableId,
+        tablePath,
+        Optional.empty() /* versionOpt */,
+        Optional.empty() /* timestampOpt */);
+  }
+
+  @Override
+  public Snapshot loadSnapshotAt(long version) {
+    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) {
+    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, all ratified commits are available, so the earliest version is
+   * typically 0. This method validates that the requested version is within the valid range.
+   */
+  @Override
+  public void checkVersionExists(long version, boolean mustBeRecreatable, boolean allowOutOfRange)
+      throws VersionNotFoundException {
+    // Load latest to get the current version bounds
+    Snapshot latestSnapshot = loadLatestSnapshot();
+    long latestVersion = latestSnapshot.getVersion();
+
+    // For UC tables, earliest recreatable version is 0 (all ratified commits are available)
+    long earliestVersion = 0;
+
+    if (version < earliestVersion || ((version > latestVersion) && !allowOutOfRange)) {
+      throw new VersionNotFoundException(version, earliestVersion, latestVersion);
+    }
+  }
+
+  @Override
+  public CommitRange getTableChanges(Engine engine, long startVersion, Optional<Long> endVersion) {
+    return ucCatalogManagedClient.loadCommitRange(
+        engine,
+        tableId,
+        tablePath,
+        Optional.of(startVersion) /* startVersionOpt */,
+        Optional.empty() /* startTimestampOpt */,
+        endVersion /* endVersionOpt */,
+        Optional.empty() /* endTimestampOpt */);
+  }
+}

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

@@ -29,7 +29,7 @@
 /**
  * Utility class for extracting Unity Catalog table information from Spark catalog metadata.
  *
- * <p>This class isolates Spark dependencies, allowing {@link UCManagedSnapshotManager} to be
+ * <p>This class isolates Spark dependencies, allowing {@link UCManagedTableSnapshotManager} to be
  * created without Spark if table info is provided directly via {@link UCTableInfo}.
  */
 public final class UCUtils {

kernel-spark/src/test/java/io/delta/kernel/spark/snapshot/unitycatalog/UCManagedTableSnapshotManagerTest.java

@@ -0,0 +1,100 @@
+/*
+ * 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.spark.snapshot.unitycatalog;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import io.delta.kernel.defaults.engine.DefaultEngine;
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.unitycatalog.InMemoryUCClient;
+import io.delta.kernel.unitycatalog.UCCatalogManagedClient;
+import org.apache.hadoop.conf.Configuration;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for {@link UCManagedTableSnapshotManager}.
+ *
+ * <p>These tests focus on constructor validation and basic instantiation. Integration tests for
+ * actual functionality are in {@code UCManagedTableSnapshotManagerSuite}.
+ */
+public class UCManagedTableSnapshotManagerTest {
+
+  private static final String TEST_TABLE_ID = "test_uc_table_id";
+  private static final String TEST_TABLE_PATH = "/test/path/to/table";
+  private static final String TEST_UC_URI = "https://uc.example.com";
+  private static final String TEST_UC_TOKEN = "test_token";
+
+  private UCTableInfo createTestTableInfo() {
+    return new UCTableInfo(TEST_TABLE_ID, TEST_TABLE_PATH, TEST_UC_URI, TEST_UC_TOKEN);
+  }
+
+  private UCCatalogManagedClient createTestClient() {
+    InMemoryUCClient ucClient = new InMemoryUCClient("test_metastore");
+    return new UCCatalogManagedClient(ucClient);
+  }
+
+  private Engine createTestEngine() {
+    return DefaultEngine.create(new Configuration());
+  }
+
+  @Test
+  public void testConstructor_NullClient_ThrowsNPE() {
+    UCTableInfo tableInfo = createTestTableInfo();
+    Engine engine = createTestEngine();
+
+    NullPointerException exception =
+        assertThrows(
+            NullPointerException.class,
+            () -> new UCManagedTableSnapshotManager(null, tableInfo, engine));
+    assertEquals("ucCatalogManagedClient is null", exception.getMessage());
+  }
+
+  @Test
+  public void testConstructor_NullTableInfo_ThrowsNPE() {
+    UCCatalogManagedClient client = createTestClient();
+    Engine engine = createTestEngine();
+
+    NullPointerException exception =
+        assertThrows(
+            NullPointerException.class,
+            () -> new UCManagedTableSnapshotManager(client, null, engine));
+    assertEquals("tableInfo is null", exception.getMessage());
+  }
+
+  @Test
+  public void testConstructor_NullEngine_ThrowsNPE() {
+    UCCatalogManagedClient client = createTestClient();
+    UCTableInfo tableInfo = createTestTableInfo();
+
+    NullPointerException exception =
+        assertThrows(
+            NullPointerException.class,
+            () -> new UCManagedTableSnapshotManager(client, tableInfo, null));
+    assertEquals("engine is null", exception.getMessage());
+  }
+
+  @Test
+  public void testConstructor_ValidInputs_CreatesManager() {
+    UCCatalogManagedClient client = createTestClient();
+    UCTableInfo tableInfo = createTestTableInfo();
+    Engine engine = createTestEngine();
+
+    UCManagedTableSnapshotManager manager =
+        new UCManagedTableSnapshotManager(client, tableInfo, engine);
+
+    assertNotNull(manager);
+  }
+}