Add FileIO implementation for Google Cloud Storage

build.gradle

@@ -354,6 +354,30 @@ project(':iceberg-aws') {
   }
 }
 
+project(':iceberg-gcp') {
+  dependencies {
+    implementation project(path: ':iceberg-bundled-guava', configuration: 'shadow')
+    api project(':iceberg-api')
+    implementation project(':iceberg-common')
+    implementation project(':iceberg-core')
+
+    implementation platform('com.google.cloud:libraries-bom:24.1.0')
+    implementation 'com.google.cloud:google-cloud-storage'
+
+    testImplementation 'com.google.cloud:google-cloud-nio'
+    testImplementation 'org.assertj:assertj-core'
+
+    testImplementation project(path: ':iceberg-api', configuration: 'testArtifacts')
+
+    testImplementation("org.apache.hadoop:hadoop-common") {
+      exclude group: 'org.apache.avro', module: 'avro'
+      exclude group: 'org.slf4j', module: 'slf4j-log4j12'
+      exclude group: 'javax.servlet', module: 'servlet-api'
+      exclude group: 'com.google.code.gson', module: 'gson'
+    }
+  }
+}
+
 project(':iceberg-hive-metastore') {
   dependencies {
     implementation project(path: ':iceberg-bundled-guava', configuration: 'shadow')

gcp/src/main/java/org/apache/iceberg/gcp/GCPProperties.java

@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.iceberg.gcp;
+
+import java.io.Serializable;
+import java.util.Map;
+import java.util.Optional;
+
+public class GCPProperties implements Serializable {
+  // Service Options
+  public static final String GCS_PROJECT_ID = "gcs.project-id";
+  public static final String GCS_CLIENT_LIB_TOKEN = "gcs.client-lib-token";
+  public static final String GCS_SERVICE_HOST = "gcs.service.host";
+
+  // GCS Configuration Properties
+  public static final String GCS_DECRYPTION_KEY = "gcs.decryption-key";
+  public static final String GCS_ENCRYPTION_KEY = "gcs.encryption-key";
+  public static final String GCS_USER_PROJECT = "gcs.user-project";
+
+  public static final String GCS_CHANNEL_READ_CHUNK_SIZE = "gcs.channel.read.chunk-size-bytes";
+  public static final String GCS_CHANNEL_WRITE_CHUNK_SIZE = "gcs.channel.write.chunk-size-bytes";
+
+  private String projectId;
+  private String clientLibToken;
+  private String serviceHost;
+
+  private String gcsDecryptionKey;
+  private String gcsEncryptionKey;
+  private String gcsUserProject;
+
+  private Integer gcsChannelReadChunkSize;
+  private Integer gcsChannelWriteChunkSize;
+
+  public GCPProperties() {
+  }
+
+  public GCPProperties(Map<String, String> properties) {
+    projectId = properties.get(GCS_PROJECT_ID);
+    clientLibToken = properties.get(GCS_CLIENT_LIB_TOKEN);
+    serviceHost = properties.get(GCS_SERVICE_HOST);
+
+    gcsDecryptionKey = properties.get(GCS_DECRYPTION_KEY);
+    gcsEncryptionKey = properties.get(GCS_ENCRYPTION_KEY);
+    gcsUserProject = properties.get(GCS_USER_PROJECT);
+
+    if (properties.containsKey(GCS_CHANNEL_READ_CHUNK_SIZE)) {
+      gcsChannelReadChunkSize = Integer.parseInt(properties.get(GCS_CHANNEL_READ_CHUNK_SIZE));
+    }
+
+    if (properties.containsKey(GCS_CHANNEL_WRITE_CHUNK_SIZE)) {
+      gcsChannelWriteChunkSize = Integer.parseInt(properties.get(GCS_CHANNEL_WRITE_CHUNK_SIZE));
+    }
+  }
+
+  public Optional<Integer> channelReadChunkSize() {
+    return Optional.ofNullable(gcsChannelReadChunkSize);
+  }
+
+  public Optional<Integer> channelWriteChunkSize() {
+    return Optional.ofNullable(gcsChannelWriteChunkSize);
+  }
+
+  public Optional<String> clientLibToken() {
+    return Optional.ofNullable(clientLibToken);
+  }
+
+  public Optional<String> decryptionKey() {
+    return Optional.ofNullable(gcsDecryptionKey);
+  }
+
+  public Optional<String> encryptionKey() {
+    return Optional.ofNullable(gcsEncryptionKey);
+  }
+
+  public Optional<String> projectId() {
+    return Optional.ofNullable(projectId);
+  }
+
+  public Optional<String> serviceHost() {
+    return Optional.ofNullable(serviceHost);
+  }
+
+  public Optional<String> userProject() {
+    return Optional.ofNullable(gcsUserProject);
+  }
+}

gcp/src/main/java/org/apache/iceberg/gcp/gcs/BaseGCSFile.java

@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.iceberg.gcp.gcs;
+
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.Storage;
+import java.net.URI;
+import org.apache.iceberg.gcp.GCPProperties;
+
+abstract class BaseGCSFile {
+  private final Storage storage;
+  private final GCPProperties gcpProperties;
+  private final BlobId blobId;
+  private Blob metadata;
+
+  BaseGCSFile(Storage storage, BlobId blobId, GCPProperties gcpProperties) {
+    this.storage = storage;
+    this.blobId = blobId;
+    this.gcpProperties = gcpProperties;
+  }
+
+  public String location() {
+    return blobId.toGsUtilUri();
+  }
+
+  Storage storage() {
+    return storage;
+  }
+
+  URI uri() {
+    return URI.create(blobId.toGsUtilUri());
+  }
+
+  BlobId blobId() {
+    return blobId;
+  }
+
+  protected GCPProperties gcpProperties() {
+    return gcpProperties;
+  }
+
+  public boolean exists() {
+    return getBlob() != null;
+  }
+
+  protected Blob getBlob() {
+    if (metadata == null) {
+      metadata = storage.get(blobId);
+    }
+
+    return metadata;
+  }
+
+  @Override
+  public String toString() {
+    return blobId.toString();
+  }
+}

gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSFileIO.java

@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.iceberg.gcp.gcs;
+
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.iceberg.gcp.GCPProperties;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.util.SerializableSupplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * FileIO Implementation backed by Google Cloud Storage (GCS)
+ * <p>
+ * Locations follow the conventions used by
+ * {@link com.google.cloud.storage.BlobId#fromGsUtilUri(String) BlobId.fromGsUtilUri}
+ * that follow the convention <pre>{@code gs://<bucket>/<blob_path>}</pre>
+ * <p>
+ * See <a href="https://cloud.google.com/storage/docs/folders#overview">Cloud Storage Overview</a>
+ */
+public class GCSFileIO implements FileIO {
+  private static final Logger LOG = LoggerFactory.getLogger(GCSFileIO.class);
+
+  private SerializableSupplier<Storage> storageSupplier;
+  private GCPProperties gcpProperties;
+  private transient Storage storage;
+  private final AtomicBoolean isResourceClosed = new AtomicBoolean(false);
+
+  /**
+   * No-arg constructor to load the FileIO dynamically.
+   * <p>
+   * All fields are initialized by calling {@link GCSFileIO#initialize(Map)} later.
+   */
+  public GCSFileIO() {
+  }
+
+  /**
+   * Constructor with custom storage supplier and GCP properties.
+   * <p>
+   * Calling {@link GCSFileIO#initialize(Map)} will overwrite information set in this constructor.
+   *
+   * @param storageSupplier storage supplier
+   * @param gcpProperties gcp properties
+   */
+  public GCSFileIO(SerializableSupplier<Storage> storageSupplier, GCPProperties gcpProperties) {
+    this.storageSupplier = storageSupplier;
+    this.gcpProperties = gcpProperties;
+  }
+
+  @Override
+  public InputFile newInputFile(String path) {
+    return GCSInputFile.fromLocation(path, client(), gcpProperties);
+  }
+
+  @Override
+  public OutputFile newOutputFile(String path) {
+    return GCSOutputFile.fromLocation(path, client(), gcpProperties);
+  }
+
+  @Override
+  public void deleteFile(String path) {
+    // There is no specific contract about whether delete should fail
+    // and other FileIO providers ignore failure.  Log the failure for
+    // now as it is not a required operation for Iceberg.
+    if (!client().delete(BlobId.fromGsUtilUri(path))) {
+      LOG.warn("Failed to delete path: {}", path);
+    }
+  }
+
+  private Storage client() {
+    if (storage == null) {
+      storage = storageSupplier.get();
+    }
+    return storage;
+  }
+
+  @Override
+  public void initialize(Map<String, String> properties) {
+    this.gcpProperties = new GCPProperties(properties);
+
+    this.storageSupplier = () -> {
+      StorageOptions.Builder builder = StorageOptions.newBuilder();
+
+      gcpProperties.projectId().ifPresent(builder::setProjectId);
+      gcpProperties.clientLibToken().ifPresent(builder::setClientLibToken);
+      gcpProperties.serviceHost().ifPresent(builder::setHost);
+
+      return builder.build().getService();
+    };
+  }
+
+  @Override
+  public void close() {
+    // handles concurrent calls to close()
+    if (isResourceClosed.compareAndSet(false, true)) {
+      if (storage != null) {
+        // GCS Storage does not appear to be closable, so release the reference
+        storage = null;
+      }
+    }
+  }
+}

gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSInputFile.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.iceberg.gcp.gcs;
+
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.Storage;
+import org.apache.iceberg.gcp.GCPProperties;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.SeekableInputStream;
+
+public class GCSInputFile extends BaseGCSFile implements InputFile {
+  public static GCSInputFile fromLocation(String location, Storage storage, GCPProperties gcpProperties) {
+    return new GCSInputFile(storage, BlobId.fromGsUtilUri(location), gcpProperties);
+  }
+
+  GCSInputFile(Storage storage, BlobId blobId, GCPProperties gcpProperties) {
+    super(storage, blobId, gcpProperties);
+  }
+
+  @Override
+  public long getLength() {
+    return getBlob().getSize();
+  }
+
+  @Override
+  public SeekableInputStream newStream() {
+    return new GCSInputStream(storage(), blobId(), gcpProperties());
+  }
+}