Support GCS for delta sharing Server

build.sbt

@@ -96,34 +96,49 @@ lazy val server = (project in file("server")) enablePlugins(JavaAppPackaging) se
     ),
     "org.apache.hadoop" % "hadoop-aws" % "2.10.1" excludeAll(
       ExclusionRule("com.fasterxml.jackson.core"),
-      ExclusionRule("com.fasterxml.jackson.module")
+      ExclusionRule("com.fasterxml.jackson.module"),
+      ExclusionRule("com.google.guava", "guava")
     ),
     "org.apache.hadoop" % "hadoop-azure" % "2.10.1" excludeAll(
+      ExclusionRule("com.fasterxml.jackson.core"),
+      ExclusionRule("com.fasterxml.jackson.module"),
+      ExclusionRule("com.google.guava", "guava")
+    ),
+    "com.google.cloud" % "google-cloud-storage" % "2.2.1" excludeAll(
       ExclusionRule("com.fasterxml.jackson.core"),
       ExclusionRule("com.fasterxml.jackson.module")
     ),
-    "org.apache.hadoop" % "hadoop-common" % "2.10.1" excludeAll(
+    "com.google.cloud.bigdataoss" % "gcs-connector" % "hadoop2-2.2.3" excludeAll(
       ExclusionRule("com.fasterxml.jackson.core"),
       ExclusionRule("com.fasterxml.jackson.module")
     ),
+    "org.apache.hadoop" % "hadoop-common" % "2.10.1" excludeAll(
+      ExclusionRule("com.fasterxml.jackson.core"),
+      ExclusionRule("com.fasterxml.jackson.module"),
+      ExclusionRule("com.google.guava", "guava")
+    ),
     "org.apache.hadoop" % "hadoop-client" % "2.10.1" excludeAll(
       ExclusionRule("com.fasterxml.jackson.core"),
-      ExclusionRule("com.fasterxml.jackson.module")
+      ExclusionRule("com.fasterxml.jackson.module"),
+      ExclusionRule("com.google.guava", "guava")
     ),
     "org.apache.parquet" % "parquet-hadoop" % "1.10.1" excludeAll(
       ExclusionRule("com.fasterxml.jackson.core"),
-      ExclusionRule("com.fasterxml.jackson.module")
+      ExclusionRule("com.fasterxml.jackson.module"),
+      ExclusionRule("com.google.guava", "guava")
     ),
     "io.delta" %% "delta-standalone" % "0.2.0" excludeAll(
       ExclusionRule("com.fasterxml.jackson.core"),
-      ExclusionRule("com.fasterxml.jackson.module")
+      ExclusionRule("com.fasterxml.jackson.module"),
+      ExclusionRule("com.google.guava", "guava")
     ),
     "org.apache.spark" %% "spark-sql" % "2.4.7" excludeAll(
       ExclusionRule("org.slf4j"),
       ExclusionRule("io.netty"),
       ExclusionRule("com.fasterxml.jackson.core"),
       ExclusionRule("com.fasterxml.jackson.module"),
-      ExclusionRule("org.json4s")
+      ExclusionRule("org.json4s"),
+      ExclusionRule("com.google.guava", "guava")
     ),
     "org.slf4j" % "slf4j-api" % "1.6.1",
     "org.slf4j" % "slf4j-simple" % "1.6.1",

server/src/main/scala/io/delta/sharing/server/CloudFileSigner.scala

@@ -22,6 +22,10 @@ import java.util.concurrent.TimeUnit.SECONDS
 
 import com.amazonaws.HttpMethod
 import com.amazonaws.services.s3.model.GeneratePresignedUrlRequest
+import com.google.cloud.storage.BlobId
+import com.google.cloud.storage.BlobInfo
+import com.google.cloud.storage.Storage
+import com.google.cloud.storage.StorageOptions
 import com.microsoft.azure.storage.{CloudStorageAccount, SharedAccessProtocols, StorageCredentialsSharedAccessSignature}
 import com.microsoft.azure.storage.blob.{SharedAccessBlobPermissions, SharedAccessBlobPolicy}
 import org.apache.hadoop.conf.Configuration
@@ -32,6 +36,7 @@ import org.apache.hadoop.fs.azurebfs.services.AuthType
 import org.apache.hadoop.fs.s3a.DefaultS3ClientFactory
 import org.apache.hadoop.util.ReflectionUtils
 
+
 trait CloudFileSigner {
   def sign(path: Path): String
 }
@@ -188,3 +193,22 @@ object AbfsFileSigner {
       getRelativePath(abfsStore, _))
   }
 }
+
+class GCSFileSigner(
+    name: URI,
+    conf: Configuration,
+    preSignedUrlTimeoutSeconds: Long) extends CloudFileSigner {
+
+  private val storage = StorageOptions.newBuilder.build.getService
+
+  override def sign(path: Path): String = {
+    val absPath = path.toUri
+    val bucketName = absPath.getHost
+    val objectName = absPath.getPath.stripPrefix("/")
+    assert(objectName.nonEmpty, s"cannot get object key from $path")
+    val blobInfo = BlobInfo.newBuilder(BlobId.of(bucketName, objectName)).build
+    storage.signUrl(
+      blobInfo, preSignedUrlTimeoutSeconds, SECONDS, Storage.SignUrlOption.withV4Signature())
+      .toString
+  }
+}

server/src/main/scala/io/delta/standalone/internal/DeltaSharedTableLoader.scala

@@ -21,6 +21,7 @@ import java.net.URI
 import java.nio.charset.StandardCharsets.UTF_8
 import java.util.concurrent.TimeUnit
 
+import com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem
 import com.google.common.cache.CacheBuilder
 import com.google.common.hash.Hashing
 import io.delta.standalone.DeltaLog
@@ -31,7 +32,7 @@ import org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem
 import org.apache.hadoop.fs.s3a.S3AFileSystem
 import org.apache.spark.sql.types.{DataType, MetadataBuilder, StructType}
 
-import io.delta.sharing.server.{model, AbfsFileSigner, S3FileSigner, WasbFileSigner}
+import io.delta.sharing.server.{model, AbfsFileSigner, GCSFileSigner, S3FileSigner, WasbFileSigner}
 import io.delta.sharing.server.config.{ServerConfig, TableConfig}
 
 /**
@@ -88,6 +89,8 @@ class DeltaSharedTable(
         WasbFileSigner(wasb, deltaLog.dataPath.toUri, conf, preSignedUrlTimeoutSeconds)
       case abfs: AzureBlobFileSystem =>
         AbfsFileSigner(abfs, deltaLog.dataPath.toUri, preSignedUrlTimeoutSeconds)
+      case gc: GoogleHadoopFileSystem =>
+        new GCSFileSigner(deltaLog.dataPath.toUri, conf, preSignedUrlTimeoutSeconds)
       case _ =>
         throw new IllegalStateException(s"File system ${fs.getClass} is not supported")
     }

server/src/test/scala/io/delta/sharing/server/DeltaSharingServiceSuite.scala

@@ -551,4 +551,35 @@ class DeltaSharingServiceSuite extends FunSuite with BeforeAndAfterAll {
     }
     assert(e.getMessage.contains("Server returned HTTP response code: 403")) // 403 Forbidden
   }
+
+  ignore("gcs support") {
+    assume(shouldRunIntegrationTest)
+    val gcsTableName = "table_gcs"
+    val response = readNDJson(requestPath(s"/shares/share_gcs/schemas/default/tables/${gcsTableName}/query"), Some("POST"), Some("{}"), Some(0))
+    val lines = response.split("\n")
+    val protocol = lines(0)
+    val metadata = lines(1)
+    val expectedProtocol = Protocol(minReaderVersion = 1).wrap
+    assert(expectedProtocol == JsonUtils.fromJson[SingleAction](protocol))
+    val expectedMetadata = Metadata(
+      id = "de102585-bd69-4bba-bb10-fa92c50a7f85",
+      format = Format(),
+      schemaString = """{"type":"struct","fields":[{"name":"c1","type":"string","nullable":true,"metadata":{}},{"name":"c2","type":"string","nullable":true,"metadata":{}}]}""",
+      partitionColumns = Seq("c2")).wrap
+    assert(expectedMetadata == JsonUtils.fromJson[SingleAction](metadata))
+    val files = lines.drop(2)
+    val actualFiles = files.map(f => JsonUtils.fromJson[SingleAction](f).file)
+    assert(actualFiles.size == 1)
+    val expectedFiles = Seq(
+      AddFile(
+        url = actualFiles(0).url,
+        id = "84f5f9e4de01e99837f77bfc2b7215b0",
+        partitionValues = Map("c2" -> "foo bar"),
+        size = 568,
+        stats = """{"numRecords":1,"minValues":{"c1":"foo bar"},"maxValues":{"c1":"foo bar"},"nullCount":{"c1":0}}"""
+      )
+    )
+    assert(expectedFiles == actualFiles.toList)
+    verifyPreSignedUrl(actualFiles(0).url, 568)
+  }
 }