[Kernel-Spark] Add Unity Catalog connection extraction utilities for CCv2 support

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

@@ -0,0 +1,54 @@
+/*
+ * 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;
+
+/**
+ * Table information for Unity Catalog managed tables.
+ *
+ * <p>This POJO encapsulates all the information needed to interact with a Unity Catalog table
+ * without requiring Spark dependencies.
+ */
+public final class UCTableInfo {
+  private final String tableId;
+  private final String tablePath;
+  private final String ucUri;
+  private final String ucToken;
+
+  public UCTableInfo(String tableId, String tablePath, String ucUri, String ucToken) {
+    this.tableId = requireNonNull(tableId, "tableId is null");
+    this.tablePath = requireNonNull(tablePath, "tablePath is null");
+    this.ucUri = requireNonNull(ucUri, "ucUri is null");
+    this.ucToken = requireNonNull(ucToken, "ucToken is null");
+  }
+
+  public String getTableId() {
+    return tableId;
+  }
+
+  public String getTablePath() {
+    return tablePath;
+  }
+
+  public String getUcUri() {
+    return ucUri;
+  }
+
+  public String getUcToken() {
+    return ucToken;
+  }
+}

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

@@ -0,0 +1,113 @@
+/*
+ * 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.spark.utils.CatalogTableUtils;
+import io.delta.storage.commit.uccommitcoordinator.UCCommitCoordinatorClient;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.catalyst.catalog.CatalogTable;
+import org.apache.spark.sql.delta.coordinatedcommits.UCCatalogConfig;
+import org.apache.spark.sql.delta.coordinatedcommits.UCCommitCoordinatorBuilder$;
+
+/**
+ * Utility class for extracting Unity Catalog table information from Spark catalog metadata.
+ *
+ * <p>This class isolates Spark dependencies, allowing {@link UCManagedSnapshotManager} to be
+ * created without Spark if table info is provided directly via {@link UCTableInfo}.
+ */
+public final class UCUtils {
+
+  // Utility class - no instances
+  private UCUtils() {}
+
+  /**
+   * Extracts Unity Catalog table information from Spark catalog table metadata.
+   *
+   * @param catalogTable Spark catalog table metadata
+   * @param spark SparkSession for resolving Unity Catalog configurations
+   * @return table info if table is UC-managed, empty otherwise
+   * @throws IllegalArgumentException if table is UC-managed but configuration is invalid
+   */
+  public static Optional<UCTableInfo> extractTableInfo(
+      CatalogTable catalogTable, SparkSession spark) {
+    requireNonNull(catalogTable, "catalogTable is null");
+    requireNonNull(spark, "spark is null");
+
+    if (!CatalogTableUtils.isUnityCatalogManagedTable(catalogTable)) {
+      return Optional.empty();
+    }
+
+    String tableId = extractUCTableId(catalogTable);
+    String tablePath = extractTablePath(catalogTable);
+
+    // Get catalog name - require explicit catalog in identifier
+    scala.Option<String> catalogOption = catalogTable.identifier().catalog();
+    if (catalogOption.isEmpty()) {
+      throw new IllegalArgumentException(
+          "Unable to determine Unity Catalog for table "
+              + catalogTable.identifier()
+              + ": catalog name is missing. Use a fully-qualified table name with an explicit "
+              + "catalog (e.g., catalog.schema.table).");
+    }
+    String catalogName = catalogOption.get();
+
+    // Get UC endpoint and token from Spark configs
+    scala.collection.immutable.Map<String, UCCatalogConfig> ucConfigs =
+        UCCommitCoordinatorBuilder$.MODULE$.getCatalogConfigMap(spark);
+
+    scala.Option<UCCatalogConfig> configOpt = ucConfigs.get(catalogName);
+
+    if (configOpt.isEmpty()) {
+      throw new IllegalArgumentException(
+          "Cannot create UC client for table "
+              + catalogTable.identifier()
+              + ": Unity Catalog configuration not found for catalog '"
+              + catalogName
+              + "'.");
+    }
+
+    UCCatalogConfig config = configOpt.get();
+    String ucUri = config.uri();
+    String ucToken = config.token();
+
+    return Optional.of(new UCTableInfo(tableId, tablePath, ucUri, ucToken));
+  }
+
+  private static String extractUCTableId(CatalogTable catalogTable) {
+    Map<String, String> storageProperties =
+        scala.jdk.javaapi.CollectionConverters.asJava(catalogTable.storage().properties());
+
+    // TODO: UC constants should be consolidated in a shared location (future PR)
+    String ucTableId = storageProperties.get(UCCommitCoordinatorClient.UC_TABLE_ID_KEY);
+    if (ucTableId == null || ucTableId.isEmpty()) {
+      throw new IllegalArgumentException(
+          "Cannot extract ucTableId from table " + catalogTable.identifier());
+    }
+    return ucTableId;
+  }
+
+  private static String extractTablePath(CatalogTable catalogTable) {
+    if (catalogTable.location() == null) {
+      throw new IllegalArgumentException(
+          "Cannot extract table path: location is null for table " + catalogTable.identifier());
+    }
+    return catalogTable.location().toString();
+  }
+}

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

@@ -0,0 +1,40 @@
+/*
+ * 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.assertEquals;
+
+import org.junit.jupiter.api.Test;
+
+/** Tests for {@link UCTableInfo}. */
+class UCTableInfoTest {
+
+  @Test
+  void testConstructor_ValidInputs_StoresAllFields() {
+    // Use distinctive values that would fail if implementation had hardcoded defaults
+    String tableId = "uc_tbl_7f3a9b2c-e8d1-4f6a";
+    String tablePath = "abfss://[email protected]/delta/v2";
+    String ucUri = "https://uc-server.example.net/api/2.1/uc";
+    String ucToken = "dapi_Kx9mN$2pQr#7vWz";
+
+    UCTableInfo info = new UCTableInfo(tableId, tablePath, ucUri, ucToken);
+
+    assertEquals(tableId, info.getTableId(), "Table ID should be stored correctly");
+    assertEquals(tablePath, info.getTablePath(), "Table path should be stored correctly");
+    assertEquals(ucUri, info.getUcUri(), "UC URI should be stored correctly");
+    assertEquals(ucToken, info.getUcToken(), "UC token should be stored correctly");
+  }
+}

kernel-spark/src/test/java/io/delta/kernel/spark/utils/CatalogTableUtilsTest.java

@@ -16,14 +16,15 @@
 package io.delta.kernel.spark.utils;
 
 import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import io.delta.storage.commit.uccommitcoordinator.UCCommitCoordinatorClient;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import org.apache.spark.sql.catalyst.catalog.CatalogTable;
 import org.junit.jupiter.api.Test;
+import scala.Option;
 
 /** Tests for {@link CatalogTableUtils}. */
 class CatalogTableUtilsTest {
@@ -97,22 +98,6 @@ void testIsUnityCatalogManaged_PreviewFlagMissingId_ReturnsFalse() {
         "Preview flag without ID should not be considered Unity managed");
   }
 
-  @Test
-  void testIsCatalogManaged_NullTable_ThrowsException() {
-    assertThrows(
-        NullPointerException.class,
-        () -> CatalogTableUtils.isCatalogManaged(null),
-        "Null table should throw NullPointerException");
-  }
-
-  @Test
-  void testIsUnityCatalogManaged_NullTable_ThrowsException() {
-    assertThrows(
-        NullPointerException.class,
-        () -> CatalogTableUtils.isUnityCatalogManagedTable(null),
-        "Null table should throw NullPointerException");
-  }
-
   @Test
   void testIsCatalogManaged_NullStorage_ReturnsFalse() {
     CatalogTable table = catalogTableWithNullStorage(Collections.emptyMap());
@@ -151,15 +136,36 @@ void testIsUnityCatalogManaged_NullStorageProperties_ReturnsFalse() {
 
   private static CatalogTable catalogTable(
       Map<String, String> properties, Map<String, String> storageProperties) {
-    return CatalogTableTestUtils$.MODULE$.catalogTableWithProperties(properties, storageProperties);
+    return CatalogTableTestUtils$.MODULE$.createCatalogTable(
+        "tbl" /* tableName */,
+        Option.empty() /* catalogName */,
+        properties,
+        storageProperties,
+        Option.empty() /* locationUri */,
+        false /* nullStorage */,
+        false /* nullStorageProperties */);
   }
 
   private static CatalogTable catalogTableWithNullStorage(Map<String, String> properties) {
-    return CatalogTableTestUtils$.MODULE$.catalogTableWithNullStorage(properties);
+    return CatalogTableTestUtils$.MODULE$.createCatalogTable(
+        "tbl" /* tableName */,
+        Option.empty() /* catalogName */,
+        properties,
+        new HashMap<>() /* storageProperties */,
+        Option.empty() /* locationUri */,
+        true /* nullStorage */,
+        false /* nullStorageProperties */);
   }
 
   private static CatalogTable catalogTableWithNullStorageProperties(
       Map<String, String> properties) {
-    return CatalogTableTestUtils$.MODULE$.catalogTableWithNullStorageProperties(properties);
+    return CatalogTableTestUtils$.MODULE$.createCatalogTable(
+        "tbl" /* tableName */,
+        Option.empty() /* catalogName */,
+        properties,
+        new HashMap<>() /* storageProperties */,
+        Option.empty() /* locationUri */,
+        false /* nullStorage */,
+        true /* nullStorageProperties */);
   }
 }