feat(io): add OSS storage implementation (#1153)

## What changes are included in this PR?

- [x] Support AliyunOSS backend by  OpenDAL
- [x] Update doc explains why feature `storage-oss` is not included in
`storage-all`.
- [x] Fix typo

<!--
Provide a summary of the modifications in this PR. List the main changes
such as new features, bug fixes, refactoring, or any other updates.
-->

## Are these changes tested?

<!--
Specify what test covers (unit test, integration test, etc.).

If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->

The support for AliyunOSS is based on OpenDAL, a production-verified
project. Therefore, no additional testing has been added.

Add a new example.

Signed-off-by: divinerapier <[email protected]>

Author: divinerapier

Cargo.lock

@@ -28,6 +28,7 @@ version = { workspace = true }
 iceberg = { workspace = true }
 iceberg-catalog-rest = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
+futures = { workspace = true }
 
 [[example]]
 name = "rest-catalog-namespace"
@@ -36,3 +37,13 @@ path = "src/rest_catalog_namespace.rs"
 [[example]]
 name = "rest-catalog-table"
 path = "src/rest_catalog_table.rs"
+
+[[example]]
+name = "oss-backend"
+path = "src/oss_backend.rs"
+required-features = ["storage-oss"]
+
+
+[features]
+default = []
+storage-oss = ["iceberg/storage-oss"]

crates/examples/Cargo.toml

@@ -0,0 +1,85 @@
+// 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.
+
+use std::collections::HashMap;
+
+use futures::stream::StreamExt;
+use iceberg::{Catalog, NamespaceIdent, TableIdent};
+use iceberg_catalog_rest::{RestCatalog, RestCatalogConfig};
+
+// Configure these values according to your environment
+
+static REST_URI: &str = "http://127.0.0.1:8181";
+static NAMESPACE: &str = "default";
+static TABLE_NAME: &str = "t1";
+static OSS_ENDPOINT: &str = "https://oss-cn-hangzhou.aliyuncs.com/bucket";
+static OSS_ACCESS_KEY_ID: &str = "LTAI5t999999999999999";
+static OSS_ACCESS_KEY_SECRET: &str = "99999999999999999999999999999999";
+
+/// This is a simple example that demonstrates how to use [`RestCatalog`] to read data from OSS.
+///
+/// The demo reads data from an existing table in OSS storage.
+///
+/// A running instance of the iceberg-rest catalog on port 8181 is required. You can find how to run
+/// the iceberg-rest catalog with `docker compose` in the official
+/// [quickstart documentation](https://iceberg.apache.org/spark-quickstart/).
+///
+/// The example also requires valid OSS credentials and endpoint to be configured.
+#[tokio::main]
+async fn main() {
+    // Create the REST iceberg catalog.
+    let config = RestCatalogConfig::builder()
+        .uri(REST_URI.to_string())
+        .props(HashMap::from([
+            (
+                iceberg::io::OSS_ENDPOINT.to_string(),
+                OSS_ENDPOINT.to_string(),
+            ),
+            (
+                iceberg::io::OSS_ACCESS_KEY_ID.to_string(),
+                OSS_ACCESS_KEY_ID.to_string(),
+            ),
+            (
+                iceberg::io::OSS_ACCESS_KEY_SECRET.to_string(),
+                OSS_ACCESS_KEY_SECRET.to_string(),
+            ),
+        ]))
+        .build();
+    let catalog = RestCatalog::new(config);
+
+    // Create the table identifier.
+    let namespace_ident = NamespaceIdent::from_vec(vec![NAMESPACE.to_string()]).unwrap();
+    let table_ident = TableIdent::new(namespace_ident.clone(), TABLE_NAME.to_string());
+
+    // Check if the table exists.
+    if !catalog.table_exists(&table_ident).await.unwrap() {
+        println!("Table {TABLE_NAME} must exists.");
+        return;
+    }
+
+    let table = catalog.load_table(&table_ident).await.unwrap();
+    println!("Table {TABLE_NAME} loaded!");
+
+    let scan = table.scan().select_all().build().unwrap();
+    let reader = scan.to_arrow().await.unwrap();
+    let buf = reader.collect::<Vec<_>>().await;
+
+    println!("Table {TABLE_NAME} has {} batches.", buf.len());
+
+    assert!(!buf.is_empty());
+    assert!(buf.iter().all(|x| x.is_ok()));
+}

crates/examples/src/oss_backend.rs

@@ -36,6 +36,7 @@ storage-fs = ["opendal/services-fs"]
 storage-gcs = ["opendal/services-gcs"]
 storage-memory = ["opendal/services-memory"]
 storage-s3 = ["opendal/services-s3"]
+storage-oss = ["opendal/services-oss"]
 
 async-std = ["dep:async-std"]
 tokio = ["tokio/rt-multi-thread"]

crates/iceberg/Cargo.toml

@@ -57,3 +57,26 @@ async fn main() -> Result<()> {
     Ok(())
 }
 ```
+
+## IO Support
+
+Iceberg Rust provides various storage backends through feature flags. Here are the currently supported storage backends:
+
+| Storage Backend | Feature Flag | Status | Description |
+|----------------|--------------|--------|-------------|
+| Memory | `storage-memory` | ✅ Stable | In-memory storage for testing and development |
+| Local Filesystem | `storage-fs` | ✅ Stable | Local filesystem storage |
+| Amazon S3 | `storage-s3` | ✅ Stable | Amazon S3 storage |
+| Google Cloud Storage | `storage-gcs` | ✅ Stable | Google Cloud Storage |
+| Alibaba Cloud OSS | `storage-oss` | 🧪 Experimental | Alibaba Cloud Object Storage Service |
+
+You can enable all stable storage backends at once using the `storage-all` feature flag. 
+
+> Note that `storage-oss` is currently experimental and not included in `storage-all`.
+
+Example usage in `Cargo.toml`:
+
+```toml
+[dependencies]
+iceberg = { version = "x.y.z", features = ["storage-s3", "storage-fs"] }
+```

crates/iceberg/README.md

@@ -268,7 +268,7 @@ define_from_err!(
 define_from_err!(
     std::array::TryFromSliceError,
     ErrorKind::DataInvalid,
-    "failed to convert byte slive to array"
+    "failed to convert byte slice to array"
 );
 
 define_from_err!(

crates/iceberg/src/error.rs

@@ -89,6 +89,11 @@ mod storage_gcs;
 #[cfg(feature = "storage-gcs")]
 pub use storage_gcs::*;
 
+#[cfg(feature = "storage-oss")]
+mod storage_oss;
+#[cfg(feature = "storage-oss")]
+pub use storage_oss::*;
+
 pub(crate) fn is_truthy(value: &str) -> bool {
     ["true", "t", "1", "on"].contains(&value.to_lowercase().as_str())
 }

crates/iceberg/src/io/mod.rs

@@ -20,6 +20,8 @@ use std::sync::Arc;
 use opendal::layers::RetryLayer;
 #[cfg(feature = "storage-gcs")]
 use opendal::services::GcsConfig;
+#[cfg(feature = "storage-oss")]
+use opendal::services::OssConfig;
 #[cfg(feature = "storage-s3")]
 use opendal::services::S3Config;
 use opendal::{Operator, Scheme};
@@ -41,6 +43,8 @@ pub(crate) enum Storage {
         scheme_str: String,
         config: Arc<S3Config>,
     },
+    #[cfg(feature = "storage-oss")]
+    Oss { config: Arc<OssConfig> },
     #[cfg(feature = "storage-gcs")]
     Gcs { config: Arc<GcsConfig> },
 }
@@ -65,6 +69,10 @@ impl Storage {
             Scheme::Gcs => Ok(Self::Gcs {
                 config: super::gcs_config_parse(props)?.into(),
             }),
+            #[cfg(feature = "storage-oss")]
+            Scheme::Oss => Ok(Self::Oss {
+                config: super::oss_config_parse(props)?.into(),
+            }),
             // Update doc on [`FileIO`] when adding new schemes.
             _ => Err(Error::new(
                 ErrorKind::FeatureUnsupported,
@@ -125,6 +133,7 @@ impl Storage {
                     ))
                 }
             }
+
             #[cfg(feature = "storage-gcs")]
             Storage::Gcs { config } => {
                 let operator = super::gcs_config_build(config, path)?;
@@ -138,10 +147,26 @@ impl Storage {
                     ))
                 }
             }
+            #[cfg(feature = "storage-oss")]
+            Storage::Oss { config } => {
+                let op = super::oss_config_build(config, path)?;
+
+                // Check prefix of oss path.
+                let prefix = format!("oss://{}/", op.info().name());
+                if path.starts_with(&prefix) {
+                    Ok((op, &path[prefix.len()..]))
+                } else {
+                    Err(Error::new(
+                        ErrorKind::DataInvalid,
+                        format!("Invalid oss url: {}, should start with {}", path, prefix),
+                    ))
+                }
+            }
             #[cfg(all(
                 not(feature = "storage-s3"),
                 not(feature = "storage-fs"),
-                not(feature = "storage-gcs")
+                not(feature = "storage-gcs"),
+                not(feature = "storage-oss")
             ))]
             _ => Err(Error::new(
                 ErrorKind::FeatureUnsupported,
@@ -163,6 +188,7 @@ impl Storage {
             "file" | "" => Ok(Scheme::Fs),
             "s3" | "s3a" => Ok(Scheme::S3),
             "gs" | "gcs" => Ok(Scheme::Gcs),
+            "oss" => Ok(Scheme::Oss),
             s => Ok(s.parse::<Scheme>()?),
         }
     }

crates/iceberg/src/io/storage.rs

@@ -0,0 +1,66 @@
+// 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.
+
+use std::collections::HashMap;
+
+use opendal::services::OssConfig;
+use opendal::{Configurator, Operator};
+use url::Url;
+
+use crate::{Error, ErrorKind, Result};
+
+/// Required configuration arguments for creating an Aliyun OSS Operator with OpenDAL:
+/// - `oss.endpoint`: The OSS service endpoint URL
+/// - `oss.access-key-id`: The access key ID for authentication
+/// - `oss.access-key-secret`: The access key secret for authentication
+///   Aliyun oss endpoint.
+pub const OSS_ENDPOINT: &str = "oss.endpoint";
+/// Aliyun oss access key id.
+pub const OSS_ACCESS_KEY_ID: &str = "oss.access-key-id";
+/// Aliyun oss access key secret.
+pub const OSS_ACCESS_KEY_SECRET: &str = "oss.access-key-secret";
+
+/// Parse iceberg props to oss config.
+pub(crate) fn oss_config_parse(mut m: HashMap<String, String>) -> Result<OssConfig> {
+    let mut cfg: OssConfig = OssConfig::default();
+    if let Some(endpoint) = m.remove(OSS_ENDPOINT) {
+        cfg.endpoint = Some(endpoint);
+    };
+    if let Some(access_key_id) = m.remove(OSS_ACCESS_KEY_ID) {
+        cfg.access_key_id = Some(access_key_id);
+    };
+    if let Some(access_key_secret) = m.remove(OSS_ACCESS_KEY_SECRET) {
+        cfg.access_key_secret = Some(access_key_secret);
+    };
+
+    Ok(cfg)
+}
+
+/// Build new opendal operator from give path.
+pub(crate) fn oss_config_build(cfg: &OssConfig, path: &str) -> Result<Operator> {
+    let url = Url::parse(path)?;
+    let bucket = url.host_str().ok_or_else(|| {
+        Error::new(
+            ErrorKind::DataInvalid,
+            format!("Invalid oss url: {}, missing bucket", path),
+        )
+    })?;
+
+    let builder = cfg.clone().into_builder().bucket(bucket);
+
+    Ok(Operator::new(builder)?.finish())
+}