Implement UC snapshot manager and adapter wiring

Author: TimothyW553

kernel-spark/src/main/java/io/delta/kernel/spark/snapshot/CatalogManagedSnapshotManager.java

@@ -20,7 +20,13 @@
 
 import io.delta.kernel.CommitRange;
 import io.delta.kernel.Snapshot;
+import io.delta.kernel.defaults.engine.DefaultEngine;
 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 java.util.List;
 import java.util.Optional;
 
 import io.delta.kernel.internal.DeltaHistoryManager;
@@ -30,10 +36,10 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * Wireframe implementation of DeltaSnapshotManager for catalog-managed tables (e.g., UC).
+ * Implementation of DeltaSnapshotManager for catalog-managed tables (e.g., UC).
  *
- * <p>All operations are intentionally stubbed in this PR. Functionality will be implemented in a
- * follow-up PR.
+ * <p>This snapshot manager is agnostic to the underlying catalog implementation. It delegates to a
+ * {@link ManagedCatalogAdapter}, keeping catalog-specific wiring out of the manager itself.
  */
 @Experimental
 public class CatalogManagedSnapshotManager implements DeltaSnapshotManager, AutoCloseable {
@@ -43,6 +49,7 @@ public class CatalogManagedSnapshotManager implements DeltaSnapshotManager, Auto
   private final ManagedCatalogAdapter catalogAdapter;
   private final String tableId;
   private final String tablePath;
+  private final Engine kernelEngine;
 
   public CatalogManagedSnapshotManager(
       ManagedCatalogAdapter catalogAdapter,
@@ -54,44 +61,133 @@ public CatalogManagedSnapshotManager(
     this.tablePath = requireNonNull(tablePath, "tablePath is null");
     requireNonNull(hadoopConf, "hadoopConf is null");
 
+    this.kernelEngine = DefaultEngine.create(hadoopConf);
     logger.info(
         "Created CatalogManagedSnapshotManager for table {} at path {}", tableId, tablePath);
   }
 
+  /** Loads the latest snapshot of the catalog-managed Delta table. */
   @Override
   public Snapshot loadLatestSnapshot() {
-    throw new UnsupportedOperationException("loadLatestSnapshot not implemented yet");
+    return catalogAdapter.loadSnapshot(
+        kernelEngine, /* versionOpt = */ Optional.empty(), /* timestampOpt = */ Optional.empty());
   }
 
+  /**
+   * Loads a specific version of the Unity Catalog managed Delta table.
+   *
+   * @param version the version to load (must be >= 0)
+   * @return the snapshot at the specified version
+   */
   @Override
   public Snapshot loadSnapshotAt(long version) {
     checkArgument(version >= 0, "version must be non-negative");
-    throw new UnsupportedOperationException("loadSnapshotAt not implemented yet");
+    return catalogAdapter.loadSnapshot(
+        kernelEngine, Optional.of(version), /* timestampOpt = */ Optional.empty());
   }
 
+  /**
+   * Finds the active commit at a specific timestamp.
+   *
+   * <p>For catalog-managed tables, this method retrieves ratified commits from the catalog and uses
+   * {@link DeltaHistoryManager#getActiveCommitAtTimestamp} to find the commit that was active at
+   * the specified timestamp.
+   *
+   * @param timestampMillis the timestamp in milliseconds since epoch (UTC)
+   * @param canReturnLastCommit if true, returns the last commit if the timestamp is after all
+   *     commits; if false, throws an exception
+   * @param mustBeRecreatable if true, only considers commits that can be fully recreated from
+   *     available log files; if false, considers all commits
+   * @param canReturnEarliestCommit if true, returns the earliest commit if the timestamp is before
+   *     all commits; if false, throws an exception
+   * @return the commit that was active at the specified timestamp
+   */
   @Override
   public DeltaHistoryManager.Commit getActiveCommitAtTime(
       long timestampMillis,
       boolean canReturnLastCommit,
       boolean mustBeRecreatable,
       boolean canReturnEarliestCommit) {
-    throw new UnsupportedOperationException("getActiveCommitAtTime not implemented yet");
+    // Load the latest snapshot for timestamp resolution
+    SnapshotImpl latestSnapshot = (SnapshotImpl) loadLatestSnapshot();
+
+    // Extract catalog commits from the snapshot's log segment (avoids redundant UC call)
+    List<ParsedCatalogCommitData> catalogCommits =
+        latestSnapshot.getLogSegment().getAllCatalogCommits();
+
+    return DeltaHistoryManager.getActiveCommitAtTimestamp(
+        kernelEngine,
+        latestSnapshot,
+        latestSnapshot.getLogPath(),
+        timestampMillis,
+        mustBeRecreatable,
+        canReturnLastCommit,
+        canReturnEarliestCommit,
+        catalogCommits);
   }
 
+  /**
+   * Checks if a specific version exists and is accessible.
+   *
+   * <p>For catalog-managed tables, versions are assumed to be contiguous (enforced by the catalog
+   * coordinator). This method performs a lightweight check by verifying the version is within the
+   * valid range [0, latestRatifiedVersion].
+   *
+   * <p>This approach is consistent with the existing Spark Delta behavior in {@code
+   * DeltaHistoryManager.checkVersionExists} which also assumes contiguous commits.
+   *
+   * @param version the version to check
+   * @param mustBeRecreatable if true, requires that the version can be fully recreated from
+   *     available log files. For catalog-managed tables, all versions are recreatable since the
+   *     catalog maintains the complete commit history.
+   * @param allowOutOfRange if true, allows versions greater than the latest version without
+   *     throwing an exception; if false, throws exception for out-of-range versions
+   * @throws VersionNotFoundException if the version is not available
+   */
   @Override
-  public void checkVersionExists(long version, boolean mustBeRecreatable, boolean allowOutOfRange) {
+  public void checkVersionExists(long version, boolean mustBeRecreatable, boolean allowOutOfRange)
+      throws VersionNotFoundException {
     checkArgument(version >= 0, "version must be non-negative");
-    throw new UnsupportedOperationException("checkVersionExists not implemented yet");
+
+    // For catalog-managed tables, the earliest recreatable version is 0 since the catalog
+    // maintains the complete commit history
+    long earliestVersion = 0;
+    long latestVersion = catalogAdapter.getLatestRatifiedVersion();
+
+    if (version < earliestVersion || ((version > latestVersion) && !allowOutOfRange)) {
+      throw new VersionNotFoundException(version, earliestVersion, latestVersion);
+    }
   }
 
+  /**
+   * Gets a range of table changes between versions.
+   *
+   * <p><strong>Note:</strong> This operation delegates to the managed commit client.
+   *
+   * @throws UnsupportedOperationException if not yet implemented for catalog-managed tables
+   */
   @Override
   public CommitRange getTableChanges(Engine engine, long startVersion, Optional<Long> endVersion) {
-    throw new UnsupportedOperationException("getTableChanges not implemented yet");
+    requireNonNull(engine, "engine is null");
+    checkArgument(startVersion >= 0, "startVersion must be non-negative");
+    endVersion.ifPresent(v -> checkArgument(v >= 0, "endVersion must be non-negative"));
+
+    return catalogAdapter.loadCommitRange(
+        engine,
+        Optional.of(startVersion),
+        /* startTimestampOpt = */ Optional.empty(),
+        endVersion,
+        /* endTimestampOpt = */ Optional.empty());
   }
 
+  /**
+   * Closes the UC client and releases resources.
+   *
+   * <p>This method should be called when the snapshot manager is no longer needed. Prefer using
+   * try-with-resources to ensure proper cleanup.
+   */
   @Override
   public void close() {
-    // no-op in wireframe; adapter may implement close in the future
     try {
       catalogAdapter.close();
       logger.info("Closed CatalogManagedSnapshotManager for table {}", tableId);

kernel-spark/src/main/java/io/delta/kernel/spark/snapshot/DeltaSnapshotManagerFactory.java

@@ -17,6 +17,8 @@
 
 import static java.util.Objects.requireNonNull;
 
+import io.delta.kernel.spark.snapshot.unitycatalog.UnityCatalogAdapter;
+import java.util.Optional;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.spark.annotation.Experimental;
 import org.apache.spark.sql.SparkSession;
@@ -78,16 +80,36 @@ public static DeltaSnapshotManager fromPath(String tablePath, Configuration hado
   /**
    * Creates a snapshot manager from catalog table metadata.
    *
-   * <p>Wire-up is intentionally deferred; this skeleton method currently throws until implemented
-   * in a follow-up PR.
+   * <p>Automatically selects {@link CatalogManagedSnapshotManager} for Unity Catalog managed
+   * tables, or falls back to {@link PathBasedSnapshotManager} for regular tables.
    *
-   * @throws UnsupportedOperationException always, until UC wiring is added
+   * @param catalogTable Spark catalog table metadata
+   * @param spark SparkSession for resolving Unity Catalog configurations
+   * @param hadoopConf Hadoop configuration for the Delta Kernel engine
+   * @return appropriate snapshot manager implementation
+   * @throws NullPointerException if catalogTable, spark, or hadoopConf is null
+   * @throws IllegalArgumentException if catalogTable is UC-managed but configuration is invalid
    */
   public static DeltaSnapshotManager fromCatalogTable(
       CatalogTable catalogTable, SparkSession spark, Configuration hadoopConf) {
     requireNonNull(catalogTable, "catalogTable is null");
     requireNonNull(spark, "spark is null");
     requireNonNull(hadoopConf, "hadoopConf is null");
-    throw new UnsupportedOperationException("UC catalog wiring not implemented in skeleton");
+
+    Optional<ManagedCatalogAdapter> adapterOpt =
+        UnityCatalogAdapter.fromCatalog(catalogTable, spark);
+
+    if (adapterOpt.isPresent()) {
+      ManagedCatalogAdapter adapter = adapterOpt.get();
+      // Cast to UnityCatalogAdapter to access tableId and tablePath
+      UnityCatalogAdapter ucAdapter = (UnityCatalogAdapter) adapter;
+      String tableId = ucAdapter.getTableId();
+      String tablePath = ucAdapter.getTablePath();
+      return new CatalogManagedSnapshotManager(adapter, tableId, tablePath, hadoopConf);
+    }
+
+    // Fallback to path-based snapshot manager
+    String tablePath = catalogTable.location().toString();
+    return new PathBasedSnapshotManager(tablePath, hadoopConf);
   }
 }

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

@@ -20,46 +20,73 @@
 import io.delta.kernel.CommitRange;
 import io.delta.kernel.Snapshot;
 import io.delta.kernel.engine.Engine;
+import io.delta.kernel.internal.files.ParsedCatalogCommitData;
 import io.delta.kernel.internal.files.ParsedLogData;
 import io.delta.kernel.spark.snapshot.ManagedCatalogAdapter;
+import io.delta.kernel.unitycatalog.UCCatalogManagedClient;
+import io.delta.storage.commit.Commit;
+import io.delta.storage.commit.GetCommitsResponse;
+import io.delta.storage.commit.uccommitcoordinator.UCClient;
+import io.delta.storage.commit.uccommitcoordinator.UCCommitCoordinatorException;
+import io.delta.storage.commit.uccommitcoordinator.UCTokenBasedRestClient;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Optional;
+import java.util.stream.Collectors;
+import org.apache.hadoop.fs.Path;
 import org.apache.spark.sql.SparkSession;
 import org.apache.spark.sql.catalyst.catalog.CatalogTable;
 
-/**
- * UC-backed implementation shell of {@link ManagedCatalogAdapter}.
- *
- * <p>Methods are intentionally stubbed in this wireframe PR and will be implemented in a follow-up
- * once UC operations are enabled.
- */
+/** UC-backed implementation of {@link ManagedCatalogAdapter}. */
 public final class UnityCatalogAdapter implements ManagedCatalogAdapter {
 
   private final String tableId;
   private final String tablePath;
+  private final UCClient ucClient;
+  private final UCCatalogManagedClient ucManagedClient;
 
-  public UnityCatalogAdapter(String tableId, String tablePath) {
+  public UnityCatalogAdapter(String tableId, String tablePath, UCClient ucClient) {
     this.tableId = requireNonNull(tableId, "tableId is null");
     this.tablePath = requireNonNull(tablePath, "tablePath is null");
+    this.ucClient = requireNonNull(ucClient, "ucClient is null");
+    this.ucManagedClient = new UCCatalogManagedClient(ucClient);
   }
 
   /**
    * Creates adapter from Spark catalog table (convenience method).
    *
    * <p>Extracts UC connection info from Spark metadata and delegates to {@link
    * #fromConnectionInfo}.
+   *
+   * @param catalogTable Spark catalog table metadata
+   * @param spark SparkSession for resolving Unity Catalog configurations
+   * @return adapter if table is UC-managed, empty otherwise
+   * @throws IllegalArgumentException if table is UC-managed but configuration is invalid
    */
   public static Optional<ManagedCatalogAdapter> fromCatalog(
       CatalogTable catalogTable, SparkSession spark) {
     requireNonNull(catalogTable, "catalogTable is null");
     requireNonNull(spark, "spark is null");
-    throw new UnsupportedOperationException("UC wiring deferred to implementation PR");
+
+    return SparkUnityCatalogUtils.extractConnectionInfo(catalogTable, spark)
+        .map(UnityCatalogAdapter::fromConnectionInfo);
   }
 
-  /** Creates adapter from connection info (no Spark dependency). */
+  /**
+   * Creates adapter from connection info (no Spark dependency).
+   *
+   * <p>This method allows creating a UC adapter without Spark dependencies if you have connection
+   * information directly.
+   *
+   * @param info Unity Catalog connection information
+   * @return adapter instance
+   */
   public static ManagedCatalogAdapter fromConnectionInfo(UnityCatalogConnectionInfo info) {
     requireNonNull(info, "info is null");
-    return new UnityCatalogAdapter(info.getTableId(), info.getTablePath());
+    UCClient client = new UCTokenBasedRestClient(info.getEndpoint(), info.getToken());
+    return new UnityCatalogAdapter(info.getTableId(), info.getTablePath(), client);
   }
 
   public String getTableId() {
@@ -73,7 +100,7 @@ public String getTablePath() {
   @Override
   public Snapshot loadSnapshot(
       Engine engine, Optional<Long> versionOpt, Optional<Long> timestampOpt) {
-    throw new UnsupportedOperationException("UC snapshot loading not implemented yet");
+    return ucManagedClient.loadSnapshot(engine, tableId, tablePath, versionOpt, timestampOpt);
   }
 
   @Override
@@ -83,21 +110,62 @@ public CommitRange loadCommitRange(
       Optional<Long> startTimestampOpt,
       Optional<Long> endVersionOpt,
       Optional<Long> endTimestampOpt) {
-    throw new UnsupportedOperationException("UC commit range loading not implemented yet");
+    return ucManagedClient.loadCommitRange(
+        engine,
+        tableId,
+        tablePath,
+        startVersionOpt,
+        startTimestampOpt,
+        endVersionOpt,
+        endTimestampOpt);
   }
 
   @Override
   public List<ParsedLogData> getRatifiedCommits(Optional<Long> endVersionOpt) {
-    throw new UnsupportedOperationException("UC commit listing not implemented yet");
+    GetCommitsResponse response = getCommitsFromUC(endVersionOpt);
+    return response.getCommits().stream()
+        .sorted(Comparator.comparingLong(Commit::getVersion))
+        .map(
+            commit ->
+                ParsedCatalogCommitData.forFileStatus(
+                    hadoopFileStatusToKernelFileStatus(commit.getFileStatus())))
+        .collect(Collectors.toList());
   }
 
   @Override
   public long getLatestRatifiedVersion() {
-    throw new UnsupportedOperationException("UC ratified version lookup not implemented yet");
+    GetCommitsResponse response = getCommitsFromUC(Optional.empty());
+    long maxRatified = response.getLatestTableVersion();
+    // UC returns -1 when only 0.json exists (CREATE not yet registered with UC)
+    return maxRatified == -1 ? 0 : maxRatified;
   }
 
   @Override
   public void close() {
-    // no-op in wireframe
+    try {
+      ucClient.close();
+    } catch (Exception e) {
+      // Swallow close errors to avoid disrupting caller cleanup
+    }
+  }
+
+  private GetCommitsResponse getCommitsFromUC(Optional<Long> endVersionOpt) {
+    try {
+      return ucClient.getCommits(
+          tableId,
+          new Path(tablePath).toUri(),
+          Optional.empty(), // startVersion
+          endVersionOpt);
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    } catch (UCCommitCoordinatorException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private static io.delta.kernel.utils.FileStatus hadoopFileStatusToKernelFileStatus(
+      org.apache.hadoop.fs.FileStatus hadoopFS) {
+    return io.delta.kernel.utils.FileStatus.of(
+        hadoopFS.getPath().toString(), hadoopFS.getLen(), hadoopFS.getModificationTime());
   }
 }