spark: use UC Delta Rest Catalog API credentials for external createTable

Author: timothyw553

spark/src/main/scala/org/apache/spark/sql/delta/catalog/DeltaCatalogClient.scala

@@ -157,6 +157,21 @@ private class DeltaCatalogClient private (
                 schemaName,
                 tableName,
                 staging.getTableId.toString)))
+          case (CatalogTableType.EXTERNAL, Some(externalLocation))
+              if isCloudScheme(externalLocation.getScheme) =>
+            val locationText = externalLocation.toString
+            // External create must write the initial _delta_log, so READ fallback would be wrong.
+            val credentials =
+              client.getTemporaryPathCredentials(locationText, CredentialOperation.READ_WRITE)
+            Some(PreparedUCDeltaRestCatalogApiCreate(
+              location = externalLocation,
+              tableProperties = Map.empty,
+              storageProperties = toPathCredentialProperties(
+                locationText,
+                getStorageCredentials(credentials),
+                externalLocation.getScheme,
+                PathOperation.PATH_CREATE_TABLE,
+                credentialContext)))
           case _ =>
             None
         }
@@ -566,6 +581,33 @@ private[delta] object DeltaCatalogClient {
     }
   }
 
+  private def toPathCredentialProperties(
+      location: String,
+      storageCredentials: Seq[StorageCredential],
+      locationScheme: String,
+      pathOperation: PathOperation,
+      credentialContext: Option[UCDeltaRestCatalogApiCredentialContext]): Map[String, String] = {
+    cloudCredentialProperties(
+      location,
+      storageCredentials,
+      locationScheme,
+      credentialContext) { (context, credential) =>
+      CredPropsUtil.createPathCredProps(
+        context.renewCredentialEnabled,
+        context.credScopedFsEnabled,
+        context.hadoopConf,
+        locationScheme.toLowerCase(Locale.ROOT),
+        context.uri,
+        context.tokenProvider,
+        location,
+        pathOperation,
+        DeltaStorageCredentialUtil.toTemporaryCredentials(
+          toUnityCatalogStorageCredential(credential)))
+        .asScala
+        .toMap
+    }
+  }
+
   private def toTableProperties(staging: StagingTableResponse): Map[String, String] = {
     val stagingTableId = staging.getTableId.toString
     val requiredProperties = Option(staging.getRequiredProperties)

spark/src/test/scala/org/apache/spark/sql/delta/catalog/DeltaCatalogClientSuite.scala

@@ -785,6 +785,46 @@ class DeltaCatalogClientSuite
     }
   }
 
+  test("prepareCreateTable uses temporary path credentials for external cloud tables") {
+    val location = "s3://bucket/external/tbl"
+    configHandler = exchange => {
+      assert(queryParams(exchange)("catalog") === "uc")
+      sendJson(exchange, 200,
+        """{
+          |  "endpoints": [
+          |    "GET /v1/catalogs/{catalog}/schemas/{schema}/tables/{table}",
+          |    "GET /v1/catalogs/{catalog}/schemas/{schema}/tables/{table}/credentials",
+          |    "GET /v1/temporary-path-credentials"
+          |  ],
+          |  "protocol-version": "1.0"
+          |}""".stripMargin)
+    }
+    pathCredentialsHandler = exchange => {
+      credentialRequestCount += 1
+      assert(queryParams(exchange) === Map(
+        "location" -> location,
+        "operation" -> "READ_WRITE"))
+      sendJson(exchange, 200, s3CredentialsResponseJson(location, "READ_WRITE"))
+    }
+
+    val prepared = withUCDeltaRestCatalogApi { catalog =>
+      catalog.prepareCreateTable(
+        Identifier.of(Array("default"), "tbl"),
+        CatalogTableType.EXTERNAL,
+        location = Some(java.net.URI.create(location))).get
+    }
+
+    assert(credentialRequestCount === 1)
+    assert(prepared.location.toString === location)
+    assert(prepared.tableProperties.isEmpty)
+    assert(prepared.storageProperties(S3ACredentialsProviderKey) ===
+      AwsVendedTokenProviderClass)
+    assert(prepared.storageProperties(UCCredentialsTypeKey) === UCCredentialsTypePathValue)
+    assert(prepared.storageProperties(UCPathOperationKey) ===
+      "PATH_CREATE_TABLE")
+    assert(prepared.storageProperties(UCPathKey) === location)
+  }
+
   private def loadWithUCDeltaRestCatalogApi(): V1Table = {
     withUCDeltaRestCatalogApi { catalog =>
       catalog.loadTable(Identifier.of(Array("default"), "tbl")).get.asInstanceOf[V1Table]