feat: support for gcs storage

crates/iceberg/Cargo.toml

@@ -29,12 +29,13 @@ license = { workspace = true }
 keywords = ["iceberg"]
 
 [features]
-default = ["storage-memory", "storage-fs", "storage-s3", "tokio"]
-storage-all = ["storage-memory", "storage-fs", "storage-s3"]
+default = ["storage-memory", "storage-fs", "storage-s3", "tokio", "storage-gcs"]
+storage-all = ["storage-memory", "storage-fs", "storage-s3", "storage-gcs"]
 
 storage-memory = ["opendal/services-memory"]
 storage-fs = ["opendal/services-fs"]
 storage-s3 = ["opendal/services-s3"]
+storage-gcs = ["opendal/services-gcs"]
 
 async-std = ["dep:async-std"]
 tokio = ["dep:tokio"]
@@ -84,3 +85,4 @@ iceberg_test_utils = { path = "../test_utils", features = ["tests"] }
 pretty_assertions = { workspace = true }
 tempfile = { workspace = true }
 tera = { workspace = true }
+test-with = "0.13.0"

crates/iceberg/src/io/mod.rs

@@ -82,3 +82,7 @@ pub use storage_s3::*;
 mod storage_fs;
 #[cfg(feature = "storage-fs")]
 use storage_fs::*;
+#[cfg(feature = "storage-gcs")]
+mod storage_gcs;
+#[cfg(feature = "storage-gcs")]
+pub use storage_gcs::*;

crates/iceberg/src/io/storage.rs

@@ -17,6 +17,8 @@
 
 use std::sync::Arc;
 
+#[cfg(feature = "storage-gcs")]
+use opendal::services::GcsConfig;
 #[cfg(feature = "storage-s3")]
 use opendal::services::S3Config;
 use opendal::{Operator, Scheme};
@@ -38,6 +40,8 @@ pub(crate) enum Storage {
         scheme_str: String,
         config: Arc<S3Config>,
     },
+    #[cfg(feature = "storage-gcs")]
+    Gcs { config: Arc<GcsConfig> },
 }
 
 impl Storage {
@@ -56,6 +60,10 @@ impl Storage {
                 scheme_str,
                 config: super::s3_config_parse(props)?.into(),
             }),
+            #[cfg(feature = "storage-gcs")]
+            Scheme::Gcs => Ok(Self::Gcs {
+                config: super::gcs_config_parse(props)?.into(),
+            }),
             _ => Err(Error::new(
                 ErrorKind::FeatureUnsupported,
                 format!("Constructing file io from scheme: {scheme} not supported now",),
@@ -117,7 +125,24 @@ impl Storage {
                     ))
                 }
             }
-            #[cfg(all(not(feature = "storage-s3"), not(feature = "storage-fs")))]
+            #[cfg(feature = "storage-gcs")]
+            Storage::Gcs { config } => {
+                let operator = super::gcs_config_build(config)?;
+                let prefix = format!("gs://{}/", operator.info().name());
+                if path.starts_with(&prefix) {
+                    Ok((operator, &path[prefix.len()..]))
+                } else {
+                    Err(Error::new(
+                        ErrorKind::DataInvalid,
+                        format!("Invalid gcs url: {}, should start with {}", path, prefix),
+                    ))
+                }
+            }
+            #[cfg(all(
+                not(feature = "storage-s3"),
+                not(feature = "storage-fs"),
+                not(feature = "storage-gcs")
+            ))]
             _ => Err(Error::new(
                 ErrorKind::FeatureUnsupported,
                 "No storage service has been enabled",
@@ -131,6 +156,7 @@ impl Storage {
             "memory" => Ok(Scheme::Memory),
             "file" | "" => Ok(Scheme::Fs),
             "s3" | "s3a" => Ok(Scheme::S3),
+            "gs" => Ok(Scheme::Gcs),
             s => Ok(s.parse::<Scheme>()?),
         }
     }

crates/iceberg/src/io/storage_gcs.rs

@@ -0,0 +1,74 @@
+// 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.
+//! Google Cloud Storage properties
+
+use std::collections::HashMap;
+
+use opendal::services::GcsConfig;
+use opendal::Operator;
+
+use crate::{Error, ErrorKind, Result};
+
+// Reference: https://github.com/apache/iceberg/blob/main/gcp/src/main/java/org/apache/iceberg/gcp/GCPProperties.java
+
+/// Google Cloud Storage bucket name
+pub const GCS_BUCKET: &str = "gcs.bucket";
+/// Google Cloud Storage endpoint
+pub const GCS_ENDPOINT: &str = "gcs.endpoint";
+/// Google Cloud Storage OAuth token
+pub const GCS_OAUTH2_TOKEN: &str = "gcs.oauth2.token";
+/// Google Cloud Storage working (root) directory
+pub const GCS_ROOT: &str = "gcs.root";
+/// Google Cloud Storage working (root) directory
+pub const GCS_CREDENTIAL_PATH: &str = "gcs.credential-path";
+
+/// Parse iceberg properties to [`GcsConfig`].
+pub(crate) fn gcs_config_parse(mut m: HashMap<String, String>) -> Result<GcsConfig> {
+    let mut cfg = GcsConfig::default();
+
+    if let Some(bucket) = m.remove(GCS_BUCKET) {
+        cfg.bucket = bucket;
+    } else {
+        return Err(Error::new(
+            ErrorKind::DataInvalid,
+            "Bucket name is required for GCS",
+        ));
+    }
+
+    if let Some(root) = m.remove(GCS_ROOT) {
+        cfg.root = Some(root)
+    }
+
+    if let Some(endpoint) = m.remove(GCS_ENDPOINT) {
+        cfg.endpoint = Some(endpoint);
+    }
+
+    if let Some(cred_path) = m.remove(GCS_CREDENTIAL_PATH) {
+        cfg.credential_path = Some(cred_path);
+    }
+
+    if let Some(token) = m.remove(GCS_OAUTH2_TOKEN) {
+        cfg.credential = Some(token);
+    }
+
+    Ok(cfg)
+}
+
+/// Build a new OpenDAL [`Operator`] based on a provided [`GcsConfig`].
+pub(crate) fn gcs_config_build(cfg: &GcsConfig) -> Result<Operator> {
+    Ok(Operator::from_config(cfg.clone())?.finish())
+}

crates/iceberg/testdata/file_io_gcs/docker-compose.yaml

@@ -0,0 +1,22 @@
+# 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.
+
+services:
+  gcs-server:
+    image: fsouza/fake-gcs-server@sha256:36b0116fae5236e8def76ccb07761a9ca323e476f366a5f4bf449cac19deaf2d
+    expose:
+      - 4443

crates/iceberg/tests/file_io_gcs_test.rs

@@ -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.
+
+//! Integration tests for FileIO Google Cloud Storage (GCS).
+
+use bytes::Bytes;
+use iceberg::io::{FileIO, FileIOBuilder, GCS_BUCKET, GCS_CREDENTIAL_PATH};
+use iceberg_test_utils::set_up;
+
+// static DOCKER_COMPOSE_ENV: RwLock<Option<DockerCompose>> = RwLock::new(None);
+
+// TODO: use compose with fake-gcs-server
+//#[ctor]
+//fn before_all() {
+//    let mut guard = DOCKER_COMPOSE_ENV.write().unwrap();
+//    let docker_compose = DockerCompose::new(
+//        normalize_test_name(module_path!()),
+//        format!("{}/testdata/file_io_gcs", env!("CARGO_MANIFEST_DIR")),
+//    );
+//    docker_compose.run();
+//    guard.replace(docker_compose);
+//}
+//
+//#[dtor]
+//fn after_all() {
+//    let mut guard = DOCKER_COMPOSE_ENV.write().unwrap();
+//    guard.take();
+//}
+
+async fn get_file_io_gcs() -> FileIO {
+    set_up();
+
+    FileIOBuilder::new("gcs")
+        .with_props(vec![
+            (GCS_BUCKET, std::env::var("GCS_BUCKET").unwrap().to_string()),
+            (
+                GCS_CREDENTIAL_PATH,
+                std::env::var("GCS_CREDENTIAL_PATH").unwrap().to_string(),
+            ),
+        ])
+        .build()
+        .unwrap()
+}
+
+fn get_gs_path() -> String {
+    format!(
+        "gs://{}",
+        std::env::var("GCS_BUCKET").expect("Only runs with var enabled")
+    )
+}
+
+#[tokio::test]
+#[test_with::env(GCS_BUCKET, GCS_CREDENTIAL_PATH)]
+async fn gcs_exists() {
+    let file_io = get_file_io_gcs().await;
+    assert!(file_io
+        .is_exist(format!("{}/", get_gs_path()))
+        .await
+        .unwrap());
+}
+
+#[tokio::test]
+#[test_with::env(GCS_BUCKET, GCS_CREDENTIAL_PATH)]
+async fn gcs_write() {
+    let gs_file = format!("{}/write-file", get_gs_path());
+    let file_io = get_file_io_gcs().await;
+    let output = file_io.new_output(&gs_file).unwrap();
+    output
+        .write(bytes::Bytes::from_static(b"iceberg-gcs!"))
+        .await
+        .expect("Write to test output file");
+    assert!(file_io.is_exist(gs_file).await.unwrap())
+}
+
+#[tokio::test]
+#[test_with::env(GCS_BUCKET, GCS_CREDENTIAL_PATH)]
+async fn gcs_read() {
+    let gs_file = format!("{}/read-gcs", get_gs_path());
+    let file_io = get_file_io_gcs().await;
+    let output = file_io.new_output(&gs_file).unwrap();
+    output
+        .write(bytes::Bytes::from_static(b"iceberg!"))
+        .await
+        .expect("Write to test output file");
+    assert!(file_io.is_exist(&gs_file).await.unwrap());
+
+    let input = file_io.new_input(gs_file).unwrap();
+    assert_eq!(input.read().await.unwrap(), Bytes::from_static(b"iceberg!"));
+}