Update interfaces to better reflect CCv2 spec

Author: TimothyW553

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

@@ -15,49 +15,66 @@
  */
 package io.delta.kernel.spark.snapshot;
 
-import io.delta.kernel.CommitRange;
-import io.delta.kernel.Snapshot;
-import io.delta.kernel.engine.Engine;
-import io.delta.kernel.internal.files.ParsedLogData;
-import java.util.List;
+import io.delta.storage.commit.GetCommitsResponse;
 import java.util.Optional;
 
 /**
- * Adapter for catalog-managed tables that knows how to load snapshots and commit ranges for a
- * specific table.
+ * Adapter interface for catalog-managed Delta tables.
+ *
+ * <p>This is a thin, protocol-aligned interface that adapters implement to fetch commit metadata
+ * from a catalog's commit coordinator API. Adapters are responsible only for communication with the
+ * catalog - they don't know about Delta snapshots or Kernel internals.
+ *
+ * <p>The {@link CatalogManagedSnapshotManager} uses this interface to retrieve commits and then
+ * builds Delta snapshots and commit ranges using Kernel's TableManager APIs.
+ *
+ * <p>Implementations should be catalog-specific (e.g., UnityCatalogAdapter, GlueCatalogAdapter) but
+ * share this common interface so the snapshot manager can work with any catalog.
  */
 public interface ManagedCatalogAdapter extends AutoCloseable {
 
-  Snapshot loadSnapshot(Engine engine, Optional<Long> versionOpt, Optional<Long> timestampOpt);
+  /**
+   * Returns the unique identifier for this table in the catalog.
+   *
+   * @return the catalog-assigned table identifier
+   */
+  String getTableId();
 
-  CommitRange loadCommitRange(
-      Engine engine,
-      Optional<Long> startVersionOpt,
-      Optional<Long> startTimestampOpt,
-      Optional<Long> endVersionOpt,
-      Optional<Long> endTimestampOpt);
+  /**
+   * Returns the storage path for this table.
+   *
+   * @return the filesystem path to the Delta table root
+   */
+  String getTablePath();
 
   /**
-   * Gets the ratified commits from the catalog up to the specified version.
+   * Retrieves commits from the catalog's commit coordinator.
    *
-   * <p>The returned list contains {@link ParsedLogData} representing each ratified commit, sorted
-   * by version in ascending order. These are typically {@code ParsedCatalogCommitData} instances
-   * for catalog-managed tables.
+   * <p>This is the primary method that adapters must implement. It calls the catalog's API to get
+   * the list of ratified commits within the specified version range.
    *
-   * @param endVersionOpt optional end version (inclusive); if empty, returns commits up to latest
-   * @return list of parsed log data representing ratified commits, sorted by version ascending
+   * @param startVersion the starting version (inclusive), typically 0 for initial load
+   * @param endVersion optional ending version (inclusive); if empty, returns up to latest
+   * @return response containing the list of commits and the latest ratified table version
    */
-  List<ParsedLogData> getRatifiedCommits(Optional<Long> endVersionOpt);
+  GetCommitsResponse getCommits(long startVersion, Optional<Long> endVersion);
 
   /**
-   * Gets the latest ratified table version from the catalog.
+   * Returns the latest ratified table version from the catalog.
+   *
+   * <p>For catalog-managed tables, this is the highest version that has been successfully ratified
+   * by the catalog coordinator. Returns -1 if the catalog hasn't registered any commits yet (which
+   * can happen when version 0 exists but hasn't been ratified).
    *
-   * <p>For catalog-managed tables, this is the highest version that has been ratified by the
-   * catalog coordinator.
+   * <p>Default implementation calls {@link #getCommits} with no end version and extracts the latest
+   * version from the response. Implementations may override for efficiency if the catalog provides
+   * a dedicated API.
    *
-   * @return the latest version ratified by the catalog, or 0 if only the initial commit exists
+   * @return the latest version ratified by the catalog, or -1 if none registered
    */
-  long getLatestRatifiedVersion();
+  default long getLatestRatifiedVersion() {
+    return getCommits(0, Optional.empty()).getLatestTableVersion();
+  }
 
   @Override
   void close();

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

@@ -17,21 +17,21 @@
 
 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.files.ParsedLogData;
 import io.delta.kernel.spark.snapshot.ManagedCatalogAdapter;
-import java.util.List;
+import io.delta.storage.commit.GetCommitsResponse;
 import java.util.Optional;
 import org.apache.spark.sql.SparkSession;
 import org.apache.spark.sql.catalyst.catalog.CatalogTable;
 
 /**
- * UC-backed implementation shell of {@link ManagedCatalogAdapter}.
+ * Unity Catalog implementation 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.
+ * <p>This adapter is responsible only for fetching commit metadata from Unity Catalog's commit
+ * coordinator API. It does not contain any Delta/Kernel snapshot building logic - that
+ * responsibility belongs to the snapshot manager layer.
+ *
+ * <p>Methods are stubbed in this wireframe PR and will be implemented in a follow-up once UC
+ * operations are enabled.
  */
 public final class UnityCatalogAdapter implements ManagedCatalogAdapter {
 
@@ -48,16 +48,21 @@ public UnityCatalogAdapter(String tableId, String tablePath, String endpoint, St
   }
 
   /**
-   * Creates adapter from Spark catalog table (convenience method).
+   * Creates adapter from Spark catalog table.
+   *
+   * <p>Extracts UC connection info from Spark metadata and creates the adapter.
    *
-   * <p>Extracts UC connection info from Spark metadata and delegates to {@link
-   * #fromConnectionInfo}.
+   * @param catalogTable the catalog table metadata
+   * @param spark the active SparkSession
+   * @return Optional containing the adapter if this is a UC-managed table, empty otherwise
    */
   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). */
@@ -67,50 +72,39 @@ public static ManagedCatalogAdapter fromConnectionInfo(UnityCatalogConnectionInf
         info.getTableId(), info.getTablePath(), info.getEndpoint(), info.getToken());
   }
 
+  @Override
   public String getTableId() {
     return tableId;
   }
 
+  @Override
   public String getTablePath() {
     return tablePath;
   }
 
+  /** Returns the UC endpoint URL. */
   public String getEndpoint() {
     return endpoint;
   }
 
+  /** Returns the UC authentication token. */
   public String getToken() {
     return token;
   }
 
   @Override
-  public Snapshot loadSnapshot(
-      Engine engine, Optional<Long> versionOpt, Optional<Long> timestampOpt) {
-    throw new UnsupportedOperationException("UC snapshot loading not implemented yet");
-  }
-
-  @Override
-  public CommitRange loadCommitRange(
-      Engine engine,
-      Optional<Long> startVersionOpt,
-      Optional<Long> startTimestampOpt,
-      Optional<Long> endVersionOpt,
-      Optional<Long> endTimestampOpt) {
-    throw new UnsupportedOperationException("UC commit range loading not implemented yet");
-  }
-
-  @Override
-  public List<ParsedLogData> getRatifiedCommits(Optional<Long> endVersionOpt) {
-    throw new UnsupportedOperationException("UC commit listing not implemented yet");
+  public GetCommitsResponse getCommits(long startVersion, Optional<Long> endVersion) {
+    requireNonNull(endVersion, "endVersion is null");
+    throw new UnsupportedOperationException("UC getCommits not implemented yet");
   }
 
   @Override
   public long getLatestRatifiedVersion() {
-    throw new UnsupportedOperationException("UC ratified version lookup not implemented yet");
+    throw new UnsupportedOperationException("UC getLatestRatifiedVersion not implemented yet");
   }
 
   @Override
   public void close() {
-    // no-op in wireframe
+    // no-op in wireframe; will close UCClient in implementation
   }
 }