laskoviymishka commented on code in PR #3165: URL: https://github.com/apache/iceberg-rust/pull/3165#discussion_r4008118773
########## crates/storage/object_store/src/lib.rs: ########## @@ -0,0 +1,364 @@ +// 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. + +//! `object_store`-based storage implementation for Apache Iceberg. +//! +//! This crate provides [`ObjectStoreStorage`] and [`ObjectStoreStorageFactory`], +//! which implement the [`Storage`] and +//! [`StorageFactory`] traits from the `iceberg` crate +//! using the [`object_store`](https://docs.rs/object_store) crate as the backend. +//! +//! Currently only S3 storage is supported (via the `object_store-s3` feature flag, +//! enabled by default). + +#[cfg(feature = "object_store-s3")] +mod s3; + +use std::ops::Range; +use std::sync::Arc; + +use async_trait::async_trait; +use bytes::Bytes; +use dashmap::DashMap; +use futures::stream::BoxStream; +use futures::{StreamExt, TryStreamExt}; +#[cfg(feature = "object_store-s3")] +use iceberg::io::S3Config; +use iceberg::io::{ + FileMetadata, FileRead, FileWrite, InputFile, OutputFile, Storage, StorageConfig, + StorageFactory, +}; +use iceberg::{Error, ErrorKind, Result}; +use object_store::path::Path as ObjectStorePath; +use object_store::{ObjectStore, ObjectStoreExt, PutPayload, WriteMultipart}; +#[cfg(feature = "object_store-s3")] +use s3::{build_s3_store, parse_s3_url}; +use serde::{Deserialize, Serialize}; + +/// Convert an `object_store::Error` into an `iceberg::Error`. +fn from_object_store_error(e: object_store::Error) -> Error { + Error::new(ErrorKind::Unexpected, "Failure in doing io operation").with_source(e) +} + +/// `object_store`-based storage factory. +/// +/// Use this factory with `FileIOBuilder::new(factory)` to create FileIO instances +/// backed by the `object_store` crate. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum ObjectStoreStorageFactory { + /// S3 storage factory. + #[cfg(feature = "object_store-s3")] + S3, +} + +#[typetag::serde(name = "ObjectStoreStorageFactory")] +impl StorageFactory for ObjectStoreStorageFactory { + #[allow(unused_variables)] + fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>> { + match self { + #[cfg(feature = "object_store-s3")] + ObjectStoreStorageFactory::S3 => { + let s3_config = S3Config::try_from(config)?; + Ok(Arc::new(ObjectStoreStorage::S3 { + config: Arc::new(s3_config), + store_cache: Arc::new(DashMap::new()), + })) + } + } + } +} + +/// `object_store`-based storage implementation. +/// +/// Stores are cached per bucket to avoid rebuilding the client on every operation. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum ObjectStoreStorage { + /// S3 storage variant. + #[cfg(feature = "object_store-s3")] + S3 { + /// Parsed S3 configuration from iceberg core. + config: Arc<S3Config>, + /// Per-bucket store cache. + #[serde(skip, default)] + store_cache: Arc<DashMap<String, Arc<dyn ObjectStore>>>, Review Comment: Since this crate is `publish = true`, these two variant fields become part of the public API the moment it hits crates.io — `public-api.txt` already records both as `pub`. `store_cache` especially is pure implementation detail; exposing `Arc<DashMap<...>>` as a public field locks the cache structure into semver, so we couldn't later switch to `Mutex<HashMap>` or add a store abstraction without a breaking change. I'd wrap the variant data in a struct with private fields and a `pub fn new(config)` constructor, exposing the config through a getter if callers need it. Better to lock this down before the first publish than after. ########## crates/storage/object_store/src/s3.rs: ########## @@ -0,0 +1,156 @@ +// 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::sync::Arc; + +use iceberg::io::S3Config; +use iceberg::{Error, ErrorKind, Result}; +use object_store::ObjectStore; +use object_store::aws::AmazonS3Builder; +use url::Url; + +/// Parse an absolute S3 URL into (scheme, bucket, relative_path). +/// +/// Accepts `s3://` and `s3a://` `s3n://` schemes. +pub(crate) fn parse_s3_url(path: &str) -> Result<(&str, &str, &str)> { + let url = Url::parse(path).map_err(|e| { + Error::new(ErrorKind::DataInvalid, format!("Invalid URL: {path}")).with_source(e) + })?; + + let scheme = &path[..url.scheme().len()]; Review Comment: This parses with `Url` but then slices the raw input using lengths taken from the normalized parsed fields, and the two don't always line up. Two concrete failures: an uppercase scheme like `S3://bucket/key` gets sliced as `&path[..2]` = `"S3"`, which falls through the match to the unsupported-scheme error and rejects a valid URL. And `url.host_str()` is percent-decoded, so `s3://my%2Dbucket/key` gives `bucket_str = "my-bucket"` (9 bytes) while the raw span is 11 bytes — the bucket slice at line 67 returns the wrong bytes and `prefix_len` is off, which can panic on a char boundary. I'd match on `url.scheme()` directly (it's already lowercased) and pull bucket/relative from `url.host_str()` / `url.path().trim_start_matches('/')`, returning owned `String`s instead of slicing the input — that's what the opendal sibling does. wdyt? ########## crates/storage/object_store/src/s3.rs: ########## @@ -0,0 +1,156 @@ +// 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::sync::Arc; + +use iceberg::io::S3Config; +use iceberg::{Error, ErrorKind, Result}; +use object_store::ObjectStore; +use object_store::aws::AmazonS3Builder; +use url::Url; + +/// Parse an absolute S3 URL into (scheme, bucket, relative_path). +/// +/// Accepts `s3://` and `s3a://` `s3n://` schemes. +pub(crate) fn parse_s3_url(path: &str) -> Result<(&str, &str, &str)> { + let url = Url::parse(path).map_err(|e| { + Error::new(ErrorKind::DataInvalid, format!("Invalid URL: {path}")).with_source(e) + })?; + + let scheme = &path[..url.scheme().len()]; + match scheme { + "s3" | "s3a" | "s3n" => {} + _ => { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("Unsupported S3 scheme: {scheme} in url: {path}"), + )); + } + } + + let bucket_str = url.host_str().ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid s3 url: {path}, missing bucket"), + ) + })?; + + if bucket_str.is_empty() { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("Invalid s3 url: {path}, missing bucket"), + )); + }; + + let prefix_len = scheme.len() + "://".len() + bucket_str.len() + "/".len(); + let relative = if path.len() > prefix_len { + &path[prefix_len..] + } else { + "" + }; + + let bucket_start = scheme.len() + "://".len(); + let bucket = &path[bucket_start..bucket_start + bucket_str.len()]; + + Ok((scheme, bucket, relative)) +} + +/// Build an `AmazonS3` store from iceberg's `S3Config` for a given bucket. +pub(crate) fn build_s3_store(config: &S3Config, bucket: &str) -> Result<Arc<dyn ObjectStore>> { Review Comment: This maps 7 of the 16 `S3Config` fields and silently drops the rest, and the SSE fields are the dangerous ones: when an operator sets `s3.sse.type=kms` or `custom`, `TryFrom` populates the SSE fields on `S3Config` but nothing here forwards them, so we write data unencrypted even though encryption was explicitly required. `AmazonS3Builder` exposes `with_sse_kms_encryption` / `with_ssec_encryption`, and the opendal sibling maps all three SSE types — I'd mirror that. The assume-role fields (`role_arn`, `external_id`, `role_session_name`) and `disable_ec2_metadata` / `disable_config_load` are also dropped, and object_store has no builder API for those. Silently ignoring them is worse than not supporting them — a role-based config falls through to the credential chain and only fails at first I/O. I'd return a `FeatureUnsupported`/`DataInvalid` error listing the unsupported non-default fields rather than dropping them. Fix the SSE forwarding and error on the fields we can't express, and this one's resolved. ########## crates/storage/object_store/src/lib.rs: ########## @@ -0,0 +1,364 @@ +// 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. + +//! `object_store`-based storage implementation for Apache Iceberg. +//! +//! This crate provides [`ObjectStoreStorage`] and [`ObjectStoreStorageFactory`], +//! which implement the [`Storage`] and +//! [`StorageFactory`] traits from the `iceberg` crate +//! using the [`object_store`](https://docs.rs/object_store) crate as the backend. +//! +//! Currently only S3 storage is supported (via the `object_store-s3` feature flag, +//! enabled by default). + +#[cfg(feature = "object_store-s3")] +mod s3; + +use std::ops::Range; +use std::sync::Arc; + +use async_trait::async_trait; +use bytes::Bytes; +use dashmap::DashMap; +use futures::stream::BoxStream; +use futures::{StreamExt, TryStreamExt}; +#[cfg(feature = "object_store-s3")] +use iceberg::io::S3Config; +use iceberg::io::{ + FileMetadata, FileRead, FileWrite, InputFile, OutputFile, Storage, StorageConfig, + StorageFactory, +}; +use iceberg::{Error, ErrorKind, Result}; +use object_store::path::Path as ObjectStorePath; +use object_store::{ObjectStore, ObjectStoreExt, PutPayload, WriteMultipart}; +#[cfg(feature = "object_store-s3")] +use s3::{build_s3_store, parse_s3_url}; +use serde::{Deserialize, Serialize}; + +/// Convert an `object_store::Error` into an `iceberg::Error`. +fn from_object_store_error(e: object_store::Error) -> Error { + Error::new(ErrorKind::Unexpected, "Failure in doing io operation").with_source(e) Review Comment: This collapses every `object_store::Error` to `ErrorKind::Unexpected`, so a `NotFound` coming back from `read`/`metadata`/`delete` is indistinguishable from a network failure without downcasting the source. `exists` special-cases `NotFound` itself, but the others don't, and callers rely on `ErrorKind::NotFound` for control flow like commit-conflict detection and manifest reads. I'd dispatch on the `object_store::Error` variant here — `NotFound` → `ErrorKind::NotFound`, `PermissionDenied` → the closest matching kind, else `Unexpected` — so every caller gets the right kind for free. ########## crates/storage/object_store/src/lib.rs: ########## @@ -0,0 +1,364 @@ +// 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. + +//! `object_store`-based storage implementation for Apache Iceberg. +//! +//! This crate provides [`ObjectStoreStorage`] and [`ObjectStoreStorageFactory`], +//! which implement the [`Storage`] and +//! [`StorageFactory`] traits from the `iceberg` crate +//! using the [`object_store`](https://docs.rs/object_store) crate as the backend. +//! +//! Currently only S3 storage is supported (via the `object_store-s3` feature flag, +//! enabled by default). + +#[cfg(feature = "object_store-s3")] +mod s3; + +use std::ops::Range; +use std::sync::Arc; + +use async_trait::async_trait; +use bytes::Bytes; +use dashmap::DashMap; +use futures::stream::BoxStream; +use futures::{StreamExt, TryStreamExt}; +#[cfg(feature = "object_store-s3")] +use iceberg::io::S3Config; +use iceberg::io::{ + FileMetadata, FileRead, FileWrite, InputFile, OutputFile, Storage, StorageConfig, + StorageFactory, +}; +use iceberg::{Error, ErrorKind, Result}; +use object_store::path::Path as ObjectStorePath; +use object_store::{ObjectStore, ObjectStoreExt, PutPayload, WriteMultipart}; +#[cfg(feature = "object_store-s3")] +use s3::{build_s3_store, parse_s3_url}; +use serde::{Deserialize, Serialize}; + +/// Convert an `object_store::Error` into an `iceberg::Error`. +fn from_object_store_error(e: object_store::Error) -> Error { + Error::new(ErrorKind::Unexpected, "Failure in doing io operation").with_source(e) +} + +/// `object_store`-based storage factory. +/// +/// Use this factory with `FileIOBuilder::new(factory)` to create FileIO instances +/// backed by the `object_store` crate. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum ObjectStoreStorageFactory { + /// S3 storage factory. + #[cfg(feature = "object_store-s3")] + S3, +} + +#[typetag::serde(name = "ObjectStoreStorageFactory")] +impl StorageFactory for ObjectStoreStorageFactory { + #[allow(unused_variables)] + fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>> { + match self { + #[cfg(feature = "object_store-s3")] + ObjectStoreStorageFactory::S3 => { + let s3_config = S3Config::try_from(config)?; + Ok(Arc::new(ObjectStoreStorage::S3 { + config: Arc::new(s3_config), + store_cache: Arc::new(DashMap::new()), + })) + } + } + } +} + +/// `object_store`-based storage implementation. +/// +/// Stores are cached per bucket to avoid rebuilding the client on every operation. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum ObjectStoreStorage { + /// S3 storage variant. + #[cfg(feature = "object_store-s3")] + S3 { + /// Parsed S3 configuration from iceberg core. + config: Arc<S3Config>, + /// Per-bucket store cache. + #[serde(skip, default)] + store_cache: Arc<DashMap<String, Arc<dyn ObjectStore>>>, + }, +} + +impl ObjectStoreStorage { + /// Get or create a cached store and extract the relative `ObjectStorePath`. + fn get_store_and_path(&self, path: &str) -> Result<(Arc<dyn ObjectStore>, ObjectStorePath)> { + match self { + #[cfg(feature = "object_store-s3")] + ObjectStoreStorage::S3 { + config, + store_cache, + } => { + let (_scheme, bucket, relative) = parse_s3_url(path)?; + + let store = store_cache + .entry(bucket.to_string()) + .or_try_insert_with(|| build_s3_store(config, bucket))? + .value() + .clone(); + + Ok((store, ObjectStorePath::from(relative))) + } + } + } +} + +#[typetag::serde(name = "ObjectStoreStorage")] +#[async_trait] +impl Storage for ObjectStoreStorage { + async fn exists(&self, path: &str) -> Result<bool> { + let (store, object_path) = self.get_store_and_path(path)?; + match store.head(&object_path).await { + Ok(_) => Ok(true), + Err(object_store::Error::NotFound { .. }) => Ok(false), + Err(e) => Err(from_object_store_error(e)), + } + } + + async fn metadata(&self, path: &str) -> Result<FileMetadata> { + let (store, object_path) = self.get_store_and_path(path)?; + let meta = store + .head(&object_path) + .await + .map_err(from_object_store_error)?; + Ok(FileMetadata { + size: meta.size as u64, + }) + } + + async fn read(&self, path: &str) -> Result<Bytes> { + let (store, object_path) = self.get_store_and_path(path)?; + let result = store + .get(&object_path) + .await + .map_err(from_object_store_error)?; + result.bytes().await.map_err(from_object_store_error) + } + + async fn reader(&self, path: &str) -> Result<Box<dyn FileRead>> { + let (store, object_path) = self.get_store_and_path(path)?; + Ok(Box::new(ObjectStoreReader { + store, + path: object_path, + })) + } + + async fn write(&self, path: &str, bs: Bytes) -> Result<()> { + let (store, object_path) = self.get_store_and_path(path)?; + store + .put(&object_path, PutPayload::from_bytes(bs)) + .await + .map_err(from_object_store_error)?; + Ok(()) + } + + async fn writer(&self, path: &str) -> Result<Box<dyn FileWrite>> { + let (store, object_path) = self.get_store_and_path(path)?; + let upload = store + .put_multipart(&object_path) + .await + .map_err(from_object_store_error)?; + let writer = WriteMultipart::new(upload); + Ok(Box::new(ObjectStoreWriter { + writer: Some(writer), + })) + } + + async fn delete(&self, path: &str) -> Result<()> { + let (store, object_path) = self.get_store_and_path(path)?; + store + .delete(&object_path) + .await + .map_err(from_object_store_error)?; + Ok(()) + } + + async fn delete_prefix(&self, path: &str) -> Result<()> { + let (store, object_path) = self.get_store_and_path(path)?; + let prefix = if object_path.as_ref().ends_with('/') { + object_path + } else { + ObjectStorePath::from(format!("{}/", object_path.as_ref())) + }; + + let mut list_stream = store.list(Some(&prefix)); + while let Some(entry) = list_stream.next().await { + let entry = entry.map_err(from_object_store_error)?; + store + .delete(&entry.location) + .await + .map_err(from_object_store_error)?; + } + Ok(()) + } + + async fn delete_stream(&self, paths: BoxStream<'static, String>) -> Result<()> { + paths + .map(Ok) + .try_for_each_concurrent(16, |path| async move { + let (store, object_path) = self.get_store_and_path(&path)?; + store + .delete(&object_path) + .await + .map_err(from_object_store_error)?; + Ok(()) + }) + .await + } + + fn new_input(&self, path: &str) -> Result<InputFile> { + Ok(InputFile::new(Arc::new(self.clone()), path.to_string())) + } + + fn new_output(&self, path: &str) -> Result<OutputFile> { + Ok(OutputFile::new(Arc::new(self.clone()), path.to_string())) + } +} + +/// Reader that implements `FileRead` using `object_store`. +struct ObjectStoreReader { + store: Arc<dyn ObjectStore>, + path: ObjectStorePath, +} + +#[async_trait] +impl FileRead for ObjectStoreReader { + async fn read(&self, range: Range<u64>) -> Result<Bytes> { + let opts = object_store::GetOptions { + range: Some((range.start..range.end).into()), + ..Default::default() + }; + let result = self + .store + .get_opts(&self.path, opts) + .await + .map_err(from_object_store_error)?; + result.bytes().await.map_err(from_object_store_error) + } +} + +/// Writer that implements `FileWrite` using `object_store` multipart upload. +struct ObjectStoreWriter { + writer: Option<WriteMultipart>, Review Comment: `WriteMultipart` doesn't complete or abort on drop, and there's no `Drop` impl here, so if an `ObjectStoreWriter` is dropped without `close()` — panic unwind, an early `?` return, a cancelled future — the uploaded parts are orphaned in the bucket, billed indefinitely and never committed. I'd add a `Drop` that best-effort aborts the inner `WriteMultipart` via `take()`. Worth flagging that no test will catch this since it only surfaces as leaked S3 state. wdyt? ########## crates/storage/object_store/src/lib.rs: ########## @@ -0,0 +1,364 @@ +// 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. + +//! `object_store`-based storage implementation for Apache Iceberg. +//! +//! This crate provides [`ObjectStoreStorage`] and [`ObjectStoreStorageFactory`], +//! which implement the [`Storage`] and +//! [`StorageFactory`] traits from the `iceberg` crate +//! using the [`object_store`](https://docs.rs/object_store) crate as the backend. +//! +//! Currently only S3 storage is supported (via the `object_store-s3` feature flag, +//! enabled by default). + +#[cfg(feature = "object_store-s3")] +mod s3; + +use std::ops::Range; +use std::sync::Arc; + +use async_trait::async_trait; +use bytes::Bytes; +use dashmap::DashMap; +use futures::stream::BoxStream; +use futures::{StreamExt, TryStreamExt}; +#[cfg(feature = "object_store-s3")] +use iceberg::io::S3Config; +use iceberg::io::{ + FileMetadata, FileRead, FileWrite, InputFile, OutputFile, Storage, StorageConfig, + StorageFactory, +}; +use iceberg::{Error, ErrorKind, Result}; +use object_store::path::Path as ObjectStorePath; +use object_store::{ObjectStore, ObjectStoreExt, PutPayload, WriteMultipart}; +#[cfg(feature = "object_store-s3")] +use s3::{build_s3_store, parse_s3_url}; +use serde::{Deserialize, Serialize}; + +/// Convert an `object_store::Error` into an `iceberg::Error`. +fn from_object_store_error(e: object_store::Error) -> Error { + Error::new(ErrorKind::Unexpected, "Failure in doing io operation").with_source(e) +} + +/// `object_store`-based storage factory. +/// +/// Use this factory with `FileIOBuilder::new(factory)` to create FileIO instances +/// backed by the `object_store` crate. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum ObjectStoreStorageFactory { + /// S3 storage factory. + #[cfg(feature = "object_store-s3")] + S3, +} + +#[typetag::serde(name = "ObjectStoreStorageFactory")] +impl StorageFactory for ObjectStoreStorageFactory { + #[allow(unused_variables)] + fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>> { + match self { + #[cfg(feature = "object_store-s3")] + ObjectStoreStorageFactory::S3 => { + let s3_config = S3Config::try_from(config)?; + Ok(Arc::new(ObjectStoreStorage::S3 { + config: Arc::new(s3_config), + store_cache: Arc::new(DashMap::new()), + })) + } + } + } +} + +/// `object_store`-based storage implementation. +/// +/// Stores are cached per bucket to avoid rebuilding the client on every operation. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum ObjectStoreStorage { + /// S3 storage variant. + #[cfg(feature = "object_store-s3")] + S3 { + /// Parsed S3 configuration from iceberg core. + config: Arc<S3Config>, + /// Per-bucket store cache. + #[serde(skip, default)] + store_cache: Arc<DashMap<String, Arc<dyn ObjectStore>>>, + }, +} + +impl ObjectStoreStorage { + /// Get or create a cached store and extract the relative `ObjectStorePath`. + fn get_store_and_path(&self, path: &str) -> Result<(Arc<dyn ObjectStore>, ObjectStorePath)> { + match self { + #[cfg(feature = "object_store-s3")] + ObjectStoreStorage::S3 { + config, + store_cache, + } => { + let (_scheme, bucket, relative) = parse_s3_url(path)?; + + let store = store_cache + .entry(bucket.to_string()) + .or_try_insert_with(|| build_s3_store(config, bucket))? + .value() + .clone(); + + Ok((store, ObjectStorePath::from(relative))) + } + } + } +} + +#[typetag::serde(name = "ObjectStoreStorage")] +#[async_trait] +impl Storage for ObjectStoreStorage { + async fn exists(&self, path: &str) -> Result<bool> { + let (store, object_path) = self.get_store_and_path(path)?; + match store.head(&object_path).await { + Ok(_) => Ok(true), + Err(object_store::Error::NotFound { .. }) => Ok(false), + Err(e) => Err(from_object_store_error(e)), + } + } + + async fn metadata(&self, path: &str) -> Result<FileMetadata> { + let (store, object_path) = self.get_store_and_path(path)?; + let meta = store + .head(&object_path) + .await + .map_err(from_object_store_error)?; + Ok(FileMetadata { + size: meta.size as u64, Review Comment: `ObjectMeta::size` is already `u64` in object_store 0.13, so this cast is a no-op that trips `clippy::useless_conversion`. Just `size: meta.size,`. ########## crates/storage/object_store/src/lib.rs: ########## @@ -0,0 +1,364 @@ +// 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. + +//! `object_store`-based storage implementation for Apache Iceberg. +//! +//! This crate provides [`ObjectStoreStorage`] and [`ObjectStoreStorageFactory`], +//! which implement the [`Storage`] and +//! [`StorageFactory`] traits from the `iceberg` crate +//! using the [`object_store`](https://docs.rs/object_store) crate as the backend. +//! +//! Currently only S3 storage is supported (via the `object_store-s3` feature flag, +//! enabled by default). + +#[cfg(feature = "object_store-s3")] +mod s3; + +use std::ops::Range; +use std::sync::Arc; + +use async_trait::async_trait; +use bytes::Bytes; +use dashmap::DashMap; +use futures::stream::BoxStream; +use futures::{StreamExt, TryStreamExt}; +#[cfg(feature = "object_store-s3")] +use iceberg::io::S3Config; +use iceberg::io::{ + FileMetadata, FileRead, FileWrite, InputFile, OutputFile, Storage, StorageConfig, + StorageFactory, +}; +use iceberg::{Error, ErrorKind, Result}; +use object_store::path::Path as ObjectStorePath; +use object_store::{ObjectStore, ObjectStoreExt, PutPayload, WriteMultipart}; +#[cfg(feature = "object_store-s3")] +use s3::{build_s3_store, parse_s3_url}; +use serde::{Deserialize, Serialize}; + +/// Convert an `object_store::Error` into an `iceberg::Error`. +fn from_object_store_error(e: object_store::Error) -> Error { + Error::new(ErrorKind::Unexpected, "Failure in doing io operation").with_source(e) +} + +/// `object_store`-based storage factory. +/// +/// Use this factory with `FileIOBuilder::new(factory)` to create FileIO instances +/// backed by the `object_store` crate. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum ObjectStoreStorageFactory { + /// S3 storage factory. + #[cfg(feature = "object_store-s3")] + S3, +} + +#[typetag::serde(name = "ObjectStoreStorageFactory")] +impl StorageFactory for ObjectStoreStorageFactory { + #[allow(unused_variables)] + fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>> { + match self { + #[cfg(feature = "object_store-s3")] + ObjectStoreStorageFactory::S3 => { + let s3_config = S3Config::try_from(config)?; + Ok(Arc::new(ObjectStoreStorage::S3 { + config: Arc::new(s3_config), + store_cache: Arc::new(DashMap::new()), + })) + } + } + } +} + +/// `object_store`-based storage implementation. +/// +/// Stores are cached per bucket to avoid rebuilding the client on every operation. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum ObjectStoreStorage { + /// S3 storage variant. + #[cfg(feature = "object_store-s3")] + S3 { + /// Parsed S3 configuration from iceberg core. + config: Arc<S3Config>, + /// Per-bucket store cache. + #[serde(skip, default)] + store_cache: Arc<DashMap<String, Arc<dyn ObjectStore>>>, + }, +} + +impl ObjectStoreStorage { + /// Get or create a cached store and extract the relative `ObjectStorePath`. + fn get_store_and_path(&self, path: &str) -> Result<(Arc<dyn ObjectStore>, ObjectStorePath)> { + match self { + #[cfg(feature = "object_store-s3")] + ObjectStoreStorage::S3 { + config, + store_cache, + } => { + let (_scheme, bucket, relative) = parse_s3_url(path)?; + + let store = store_cache + .entry(bucket.to_string()) + .or_try_insert_with(|| build_s3_store(config, bucket))? + .value() + .clone(); + + Ok((store, ObjectStorePath::from(relative))) + } + } + } +} + +#[typetag::serde(name = "ObjectStoreStorage")] +#[async_trait] +impl Storage for ObjectStoreStorage { + async fn exists(&self, path: &str) -> Result<bool> { + let (store, object_path) = self.get_store_and_path(path)?; + match store.head(&object_path).await { + Ok(_) => Ok(true), + Err(object_store::Error::NotFound { .. }) => Ok(false), + Err(e) => Err(from_object_store_error(e)), + } + } + + async fn metadata(&self, path: &str) -> Result<FileMetadata> { + let (store, object_path) = self.get_store_and_path(path)?; + let meta = store + .head(&object_path) + .await + .map_err(from_object_store_error)?; + Ok(FileMetadata { + size: meta.size as u64, + }) + } + + async fn read(&self, path: &str) -> Result<Bytes> { + let (store, object_path) = self.get_store_and_path(path)?; + let result = store + .get(&object_path) + .await + .map_err(from_object_store_error)?; + result.bytes().await.map_err(from_object_store_error) + } + + async fn reader(&self, path: &str) -> Result<Box<dyn FileRead>> { + let (store, object_path) = self.get_store_and_path(path)?; + Ok(Box::new(ObjectStoreReader { + store, + path: object_path, + })) + } + + async fn write(&self, path: &str, bs: Bytes) -> Result<()> { + let (store, object_path) = self.get_store_and_path(path)?; + store + .put(&object_path, PutPayload::from_bytes(bs)) + .await + .map_err(from_object_store_error)?; + Ok(()) + } + + async fn writer(&self, path: &str) -> Result<Box<dyn FileWrite>> { + let (store, object_path) = self.get_store_and_path(path)?; + let upload = store + .put_multipart(&object_path) + .await + .map_err(from_object_store_error)?; + let writer = WriteMultipart::new(upload); + Ok(Box::new(ObjectStoreWriter { + writer: Some(writer), + })) + } + + async fn delete(&self, path: &str) -> Result<()> { + let (store, object_path) = self.get_store_and_path(path)?; + store + .delete(&object_path) + .await + .map_err(from_object_store_error)?; + Ok(()) + } + + async fn delete_prefix(&self, path: &str) -> Result<()> { + let (store, object_path) = self.get_store_and_path(path)?; + let prefix = if object_path.as_ref().ends_with('/') { + object_path + } else { + ObjectStorePath::from(format!("{}/", object_path.as_ref())) + }; + + let mut list_stream = store.list(Some(&prefix)); + while let Some(entry) = list_stream.next().await { + let entry = entry.map_err(from_object_store_error)?; + store + .delete(&entry.location) + .await + .map_err(from_object_store_error)?; + } + Ok(()) + } + + async fn delete_stream(&self, paths: BoxStream<'static, String>) -> Result<()> { + paths + .map(Ok) + .try_for_each_concurrent(16, |path| async move { + let (store, object_path) = self.get_store_and_path(&path)?; + store + .delete(&object_path) + .await + .map_err(from_object_store_error)?; + Ok(()) + }) + .await + } + + fn new_input(&self, path: &str) -> Result<InputFile> { + Ok(InputFile::new(Arc::new(self.clone()), path.to_string())) + } + + fn new_output(&self, path: &str) -> Result<OutputFile> { + Ok(OutputFile::new(Arc::new(self.clone()), path.to_string())) + } +} + +/// Reader that implements `FileRead` using `object_store`. +struct ObjectStoreReader { + store: Arc<dyn ObjectStore>, + path: ObjectStorePath, +} + +#[async_trait] +impl FileRead for ObjectStoreReader { + async fn read(&self, range: Range<u64>) -> Result<Bytes> { + let opts = object_store::GetOptions { + range: Some((range.start..range.end).into()), + ..Default::default() + }; + let result = self + .store + .get_opts(&self.path, opts) + .await + .map_err(from_object_store_error)?; + result.bytes().await.map_err(from_object_store_error) + } +} + +/// Writer that implements `FileWrite` using `object_store` multipart upload. +struct ObjectStoreWriter { + writer: Option<WriteMultipart>, +} + +#[async_trait] +impl FileWrite for ObjectStoreWriter { + async fn write(&mut self, bs: Bytes) -> Result<()> { + let writer = self + .writer + .as_mut() + .ok_or_else(|| Error::new(ErrorKind::Unexpected, "Writer has already been closed"))?; + writer.put(bs); + Ok(()) + } + + async fn close(&mut self) -> Result<()> { + let writer = self + .writer + .take() + .ok_or_else(|| Error::new(ErrorKind::Unexpected, "Writer has already been closed"))?; + writer.finish().await.map_err(from_object_store_error)?; + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[cfg(feature = "object_store-s3")] + fn make_s3_storage() -> ObjectStoreStorage { Review Comment: The tests here all run against `S3Config::default()`, and `AmazonS3Builder::build()` doesn't validate eagerly, so the cache and roundtrip tests pass without ever touching a backend — none of write/read/reader/delete/delete_prefix/delete_stream/metadata is actually exercised. That's false confidence about exactly the paths most likely to break (multipart lifecycle, serial-vs-batch delete, range reads). The opendal sibling has a localstack-backed test in CI. I'd add a feature/env-gated integration target covering a write+read roundtrip, a range read, and `delete_prefix` over 10+ objects before the follow-up backends lean on this crate. wdyt? ########## crates/storage/object_store/Cargo.toml: ########## @@ -0,0 +1,50 @@ +# 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] +edition = { workspace = true } +license = { workspace = true } +name = "iceberg-storage-object_store" +publish = true +repository = { workspace = true } +version = { workspace = true } + +categories = ["database"] +description = "Apache Iceberg object_store storage implementation" +keywords = ["iceberg", "object-store", "storage", "s3"] + +[features] +default = ["object_store-s3"] +object_store-s3 = ["object_store/aws"] + +[dependencies] +async-trait = { workspace = true } +bytes = { workspace = true } +dashmap = { workspace = true } +futures = { workspace = true } +iceberg = { workspace = true } +object_store = { workspace = true } +serde = { workspace = true } Review Comment: `serde = { workspace = true }` inherits only `features = ["rc"]`, but this crate uses `#[derive(Serialize, Deserialize)]`. It compiles in-workspace only because `typetag`/`iceberg` happen to activate `serde/derive` through feature unification — a downstream consumer depending on just this crate would hit `use of undeclared crate serde_derive`. Since `publish = true`, I'd declare it explicitly: `serde = { workspace = true, features = ["derive"] }`. ########## crates/storage/object_store/src/lib.rs: ########## @@ -0,0 +1,364 @@ +// 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. + +//! `object_store`-based storage implementation for Apache Iceberg. +//! +//! This crate provides [`ObjectStoreStorage`] and [`ObjectStoreStorageFactory`], +//! which implement the [`Storage`] and +//! [`StorageFactory`] traits from the `iceberg` crate +//! using the [`object_store`](https://docs.rs/object_store) crate as the backend. +//! +//! Currently only S3 storage is supported (via the `object_store-s3` feature flag, +//! enabled by default). + +#[cfg(feature = "object_store-s3")] +mod s3; + +use std::ops::Range; +use std::sync::Arc; + +use async_trait::async_trait; +use bytes::Bytes; +use dashmap::DashMap; +use futures::stream::BoxStream; +use futures::{StreamExt, TryStreamExt}; +#[cfg(feature = "object_store-s3")] +use iceberg::io::S3Config; +use iceberg::io::{ + FileMetadata, FileRead, FileWrite, InputFile, OutputFile, Storage, StorageConfig, + StorageFactory, +}; +use iceberg::{Error, ErrorKind, Result}; +use object_store::path::Path as ObjectStorePath; +use object_store::{ObjectStore, ObjectStoreExt, PutPayload, WriteMultipart}; +#[cfg(feature = "object_store-s3")] +use s3::{build_s3_store, parse_s3_url}; +use serde::{Deserialize, Serialize}; + +/// Convert an `object_store::Error` into an `iceberg::Error`. +fn from_object_store_error(e: object_store::Error) -> Error { + Error::new(ErrorKind::Unexpected, "Failure in doing io operation").with_source(e) +} + +/// `object_store`-based storage factory. +/// +/// Use this factory with `FileIOBuilder::new(factory)` to create FileIO instances +/// backed by the `object_store` crate. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum ObjectStoreStorageFactory { + /// S3 storage factory. + #[cfg(feature = "object_store-s3")] + S3, +} + +#[typetag::serde(name = "ObjectStoreStorageFactory")] +impl StorageFactory for ObjectStoreStorageFactory { + #[allow(unused_variables)] + fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>> { + match self { + #[cfg(feature = "object_store-s3")] + ObjectStoreStorageFactory::S3 => { + let s3_config = S3Config::try_from(config)?; + Ok(Arc::new(ObjectStoreStorage::S3 { + config: Arc::new(s3_config), + store_cache: Arc::new(DashMap::new()), + })) + } + } + } +} + +/// `object_store`-based storage implementation. +/// +/// Stores are cached per bucket to avoid rebuilding the client on every operation. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum ObjectStoreStorage { + /// S3 storage variant. + #[cfg(feature = "object_store-s3")] + S3 { + /// Parsed S3 configuration from iceberg core. + config: Arc<S3Config>, + /// Per-bucket store cache. + #[serde(skip, default)] + store_cache: Arc<DashMap<String, Arc<dyn ObjectStore>>>, + }, +} + +impl ObjectStoreStorage { + /// Get or create a cached store and extract the relative `ObjectStorePath`. + fn get_store_and_path(&self, path: &str) -> Result<(Arc<dyn ObjectStore>, ObjectStorePath)> { + match self { + #[cfg(feature = "object_store-s3")] + ObjectStoreStorage::S3 { + config, + store_cache, + } => { + let (_scheme, bucket, relative) = parse_s3_url(path)?; + + let store = store_cache + .entry(bucket.to_string()) + .or_try_insert_with(|| build_s3_store(config, bucket))? + .value() + .clone(); + + Ok((store, ObjectStorePath::from(relative))) + } + } + } +} + +#[typetag::serde(name = "ObjectStoreStorage")] +#[async_trait] +impl Storage for ObjectStoreStorage { + async fn exists(&self, path: &str) -> Result<bool> { + let (store, object_path) = self.get_store_and_path(path)?; + match store.head(&object_path).await { + Ok(_) => Ok(true), + Err(object_store::Error::NotFound { .. }) => Ok(false), + Err(e) => Err(from_object_store_error(e)), + } + } + + async fn metadata(&self, path: &str) -> Result<FileMetadata> { + let (store, object_path) = self.get_store_and_path(path)?; + let meta = store + .head(&object_path) + .await + .map_err(from_object_store_error)?; + Ok(FileMetadata { + size: meta.size as u64, + }) + } + + async fn read(&self, path: &str) -> Result<Bytes> { + let (store, object_path) = self.get_store_and_path(path)?; + let result = store + .get(&object_path) + .await + .map_err(from_object_store_error)?; + result.bytes().await.map_err(from_object_store_error) + } + + async fn reader(&self, path: &str) -> Result<Box<dyn FileRead>> { + let (store, object_path) = self.get_store_and_path(path)?; + Ok(Box::new(ObjectStoreReader { + store, + path: object_path, + })) + } + + async fn write(&self, path: &str, bs: Bytes) -> Result<()> { + let (store, object_path) = self.get_store_and_path(path)?; + store + .put(&object_path, PutPayload::from_bytes(bs)) + .await + .map_err(from_object_store_error)?; + Ok(()) + } + + async fn writer(&self, path: &str) -> Result<Box<dyn FileWrite>> { + let (store, object_path) = self.get_store_and_path(path)?; + let upload = store + .put_multipart(&object_path) + .await + .map_err(from_object_store_error)?; + let writer = WriteMultipart::new(upload); + Ok(Box::new(ObjectStoreWriter { + writer: Some(writer), + })) + } + + async fn delete(&self, path: &str) -> Result<()> { + let (store, object_path) = self.get_store_and_path(path)?; + store + .delete(&object_path) + .await + .map_err(from_object_store_error)?; + Ok(()) + } + + async fn delete_prefix(&self, path: &str) -> Result<()> { + let (store, object_path) = self.get_store_and_path(path)?; + let prefix = if object_path.as_ref().ends_with('/') { Review Comment: `ObjectStorePath::from` always strips trailing slashes, so `ends_with('/')` is always false and the `else` branch just re-appends-then-strips — this whole `if`/`else` collapses to `let prefix = object_path;`. It's only correct today because `store.list` matches on path-segment boundaries anyway. ########## crates/storage/object_store/src/s3.rs: ########## @@ -0,0 +1,156 @@ +// 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::sync::Arc; + +use iceberg::io::S3Config; +use iceberg::{Error, ErrorKind, Result}; +use object_store::ObjectStore; +use object_store::aws::AmazonS3Builder; +use url::Url; + +/// Parse an absolute S3 URL into (scheme, bucket, relative_path). +/// +/// Accepts `s3://` and `s3a://` `s3n://` schemes. +pub(crate) fn parse_s3_url(path: &str) -> Result<(&str, &str, &str)> { + let url = Url::parse(path).map_err(|e| { + Error::new(ErrorKind::DataInvalid, format!("Invalid URL: {path}")).with_source(e) + })?; + + let scheme = &path[..url.scheme().len()]; + match scheme { + "s3" | "s3a" | "s3n" => {} + _ => { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("Unsupported S3 scheme: {scheme} in url: {path}"), + )); + } + } + + let bucket_str = url.host_str().ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid s3 url: {path}, missing bucket"), + ) + })?; + + if bucket_str.is_empty() { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("Invalid s3 url: {path}, missing bucket"), + )); + }; + + let prefix_len = scheme.len() + "://".len() + bucket_str.len() + "/".len(); + let relative = if path.len() > prefix_len { + &path[prefix_len..] + } else { + "" + }; + + let bucket_start = scheme.len() + "://".len(); + let bucket = &path[bucket_start..bucket_start + bucket_str.len()]; + + Ok((scheme, bucket, relative)) +} + +/// Build an `AmazonS3` store from iceberg's `S3Config` for a given bucket. +pub(crate) fn build_s3_store(config: &S3Config, bucket: &str) -> Result<Arc<dyn ObjectStore>> { + let mut builder = AmazonS3Builder::new().with_bucket_name(bucket); + + if let Some(ref endpoint) = config.endpoint { + builder = builder.with_endpoint(endpoint); + if endpoint.starts_with("http://";) { + builder = builder.with_allow_http(true); + } + } + if let Some(ref access_key_id) = config.access_key_id { + builder = builder.with_access_key_id(access_key_id); + } + if let Some(ref secret_access_key) = config.secret_access_key { + builder = builder.with_secret_access_key(secret_access_key); + } + if let Some(ref session_token) = config.session_token { + builder = builder.with_token(session_token); + } + if let Some(ref region) = config.region { + builder = builder.with_region(region); + } + if config.enable_virtual_host_style { Review Comment: We only call `with_virtual_hosted_style_request(true)` when the flag is set and rely on the builder defaulting to path-style otherwise. That's accidentally correct today because object_store 0.13 defaults to path-style, but it means `s3.path-style-access=true` never explicitly sets `false` — if that default ever shifts, MinIO and other path-style-only stores break silently. I'd set it unconditionally: `builder.with_virtual_hosted_style_request(config.enable_virtual_host_style)`. ########## crates/storage/object_store/src/lib.rs: ########## @@ -0,0 +1,364 @@ +// 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. + +//! `object_store`-based storage implementation for Apache Iceberg. +//! +//! This crate provides [`ObjectStoreStorage`] and [`ObjectStoreStorageFactory`], +//! which implement the [`Storage`] and +//! [`StorageFactory`] traits from the `iceberg` crate +//! using the [`object_store`](https://docs.rs/object_store) crate as the backend. +//! +//! Currently only S3 storage is supported (via the `object_store-s3` feature flag, +//! enabled by default). + +#[cfg(feature = "object_store-s3")] +mod s3; + +use std::ops::Range; +use std::sync::Arc; + +use async_trait::async_trait; +use bytes::Bytes; +use dashmap::DashMap; +use futures::stream::BoxStream; +use futures::{StreamExt, TryStreamExt}; +#[cfg(feature = "object_store-s3")] +use iceberg::io::S3Config; +use iceberg::io::{ + FileMetadata, FileRead, FileWrite, InputFile, OutputFile, Storage, StorageConfig, + StorageFactory, +}; +use iceberg::{Error, ErrorKind, Result}; +use object_store::path::Path as ObjectStorePath; +use object_store::{ObjectStore, ObjectStoreExt, PutPayload, WriteMultipart}; +#[cfg(feature = "object_store-s3")] +use s3::{build_s3_store, parse_s3_url}; +use serde::{Deserialize, Serialize}; + +/// Convert an `object_store::Error` into an `iceberg::Error`. +fn from_object_store_error(e: object_store::Error) -> Error { + Error::new(ErrorKind::Unexpected, "Failure in doing io operation").with_source(e) +} + +/// `object_store`-based storage factory. +/// +/// Use this factory with `FileIOBuilder::new(factory)` to create FileIO instances +/// backed by the `object_store` crate. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum ObjectStoreStorageFactory { + /// S3 storage factory. + #[cfg(feature = "object_store-s3")] + S3, +} + +#[typetag::serde(name = "ObjectStoreStorageFactory")] +impl StorageFactory for ObjectStoreStorageFactory { + #[allow(unused_variables)] + fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>> { + match self { + #[cfg(feature = "object_store-s3")] + ObjectStoreStorageFactory::S3 => { + let s3_config = S3Config::try_from(config)?; + Ok(Arc::new(ObjectStoreStorage::S3 { + config: Arc::new(s3_config), + store_cache: Arc::new(DashMap::new()), + })) + } + } + } +} + +/// `object_store`-based storage implementation. +/// +/// Stores are cached per bucket to avoid rebuilding the client on every operation. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum ObjectStoreStorage { + /// S3 storage variant. + #[cfg(feature = "object_store-s3")] + S3 { + /// Parsed S3 configuration from iceberg core. + config: Arc<S3Config>, + /// Per-bucket store cache. + #[serde(skip, default)] + store_cache: Arc<DashMap<String, Arc<dyn ObjectStore>>>, + }, +} + +impl ObjectStoreStorage { + /// Get or create a cached store and extract the relative `ObjectStorePath`. + fn get_store_and_path(&self, path: &str) -> Result<(Arc<dyn ObjectStore>, ObjectStorePath)> { + match self { + #[cfg(feature = "object_store-s3")] + ObjectStoreStorage::S3 { + config, + store_cache, + } => { + let (_scheme, bucket, relative) = parse_s3_url(path)?; + + let store = store_cache + .entry(bucket.to_string()) + .or_try_insert_with(|| build_s3_store(config, bucket))? + .value() + .clone(); + + Ok((store, ObjectStorePath::from(relative))) + } + } + } +} + +#[typetag::serde(name = "ObjectStoreStorage")] +#[async_trait] +impl Storage for ObjectStoreStorage { + async fn exists(&self, path: &str) -> Result<bool> { + let (store, object_path) = self.get_store_and_path(path)?; + match store.head(&object_path).await { + Ok(_) => Ok(true), + Err(object_store::Error::NotFound { .. }) => Ok(false), + Err(e) => Err(from_object_store_error(e)), + } + } + + async fn metadata(&self, path: &str) -> Result<FileMetadata> { + let (store, object_path) = self.get_store_and_path(path)?; + let meta = store + .head(&object_path) + .await + .map_err(from_object_store_error)?; + Ok(FileMetadata { + size: meta.size as u64, + }) + } + + async fn read(&self, path: &str) -> Result<Bytes> { + let (store, object_path) = self.get_store_and_path(path)?; + let result = store + .get(&object_path) + .await + .map_err(from_object_store_error)?; + result.bytes().await.map_err(from_object_store_error) + } + + async fn reader(&self, path: &str) -> Result<Box<dyn FileRead>> { + let (store, object_path) = self.get_store_and_path(path)?; + Ok(Box::new(ObjectStoreReader { + store, + path: object_path, + })) + } + + async fn write(&self, path: &str, bs: Bytes) -> Result<()> { + let (store, object_path) = self.get_store_and_path(path)?; + store + .put(&object_path, PutPayload::from_bytes(bs)) + .await + .map_err(from_object_store_error)?; + Ok(()) + } + + async fn writer(&self, path: &str) -> Result<Box<dyn FileWrite>> { + let (store, object_path) = self.get_store_and_path(path)?; + let upload = store + .put_multipart(&object_path) + .await + .map_err(from_object_store_error)?; + let writer = WriteMultipart::new(upload); + Ok(Box::new(ObjectStoreWriter { + writer: Some(writer), + })) + } + + async fn delete(&self, path: &str) -> Result<()> { + let (store, object_path) = self.get_store_and_path(path)?; + store + .delete(&object_path) + .await + .map_err(from_object_store_error)?; + Ok(()) + } + + async fn delete_prefix(&self, path: &str) -> Result<()> { + let (store, object_path) = self.get_store_and_path(path)?; + let prefix = if object_path.as_ref().ends_with('/') { + object_path + } else { + ObjectStorePath::from(format!("{}/", object_path.as_ref())) + }; + + let mut list_stream = store.list(Some(&prefix)); + while let Some(entry) = list_stream.next().await { Review Comment: This deletes objects one at a time inside the list loop, so a prefix with 10,000 files is 10,000 sequential round-trips where S3's `DeleteObjects` batches up to 1,000 per request. It's also not atomic — a failure mid-loop leaves a partially-deleted prefix. I'd pipe the list stream of paths into `store.delete_stream(...)`, which the opendal sibling effectively does via `delete_with(...).recursive(true)`. Same fix applies to `delete_stream` below. ########## crates/storage/object_store/src/lib.rs: ########## @@ -0,0 +1,364 @@ +// 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. + +//! `object_store`-based storage implementation for Apache Iceberg. +//! +//! This crate provides [`ObjectStoreStorage`] and [`ObjectStoreStorageFactory`], +//! which implement the [`Storage`] and +//! [`StorageFactory`] traits from the `iceberg` crate +//! using the [`object_store`](https://docs.rs/object_store) crate as the backend. +//! +//! Currently only S3 storage is supported (via the `object_store-s3` feature flag, +//! enabled by default). + +#[cfg(feature = "object_store-s3")] +mod s3; + +use std::ops::Range; +use std::sync::Arc; + +use async_trait::async_trait; +use bytes::Bytes; +use dashmap::DashMap; +use futures::stream::BoxStream; +use futures::{StreamExt, TryStreamExt}; +#[cfg(feature = "object_store-s3")] +use iceberg::io::S3Config; +use iceberg::io::{ + FileMetadata, FileRead, FileWrite, InputFile, OutputFile, Storage, StorageConfig, + StorageFactory, +}; +use iceberg::{Error, ErrorKind, Result}; +use object_store::path::Path as ObjectStorePath; +use object_store::{ObjectStore, ObjectStoreExt, PutPayload, WriteMultipart}; +#[cfg(feature = "object_store-s3")] +use s3::{build_s3_store, parse_s3_url}; +use serde::{Deserialize, Serialize}; + +/// Convert an `object_store::Error` into an `iceberg::Error`. +fn from_object_store_error(e: object_store::Error) -> Error { + Error::new(ErrorKind::Unexpected, "Failure in doing io operation").with_source(e) +} + +/// `object_store`-based storage factory. +/// +/// Use this factory with `FileIOBuilder::new(factory)` to create FileIO instances +/// backed by the `object_store` crate. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum ObjectStoreStorageFactory { + /// S3 storage factory. + #[cfg(feature = "object_store-s3")] + S3, +} + +#[typetag::serde(name = "ObjectStoreStorageFactory")] +impl StorageFactory for ObjectStoreStorageFactory { + #[allow(unused_variables)] + fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>> { + match self { + #[cfg(feature = "object_store-s3")] + ObjectStoreStorageFactory::S3 => { + let s3_config = S3Config::try_from(config)?; + Ok(Arc::new(ObjectStoreStorage::S3 { + config: Arc::new(s3_config), + store_cache: Arc::new(DashMap::new()), + })) + } + } + } +} + +/// `object_store`-based storage implementation. +/// +/// Stores are cached per bucket to avoid rebuilding the client on every operation. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum ObjectStoreStorage { + /// S3 storage variant. + #[cfg(feature = "object_store-s3")] + S3 { + /// Parsed S3 configuration from iceberg core. + config: Arc<S3Config>, + /// Per-bucket store cache. + #[serde(skip, default)] + store_cache: Arc<DashMap<String, Arc<dyn ObjectStore>>>, + }, +} + +impl ObjectStoreStorage { + /// Get or create a cached store and extract the relative `ObjectStorePath`. + fn get_store_and_path(&self, path: &str) -> Result<(Arc<dyn ObjectStore>, ObjectStorePath)> { + match self { + #[cfg(feature = "object_store-s3")] + ObjectStoreStorage::S3 { + config, + store_cache, + } => { + let (_scheme, bucket, relative) = parse_s3_url(path)?; + + let store = store_cache + .entry(bucket.to_string()) + .or_try_insert_with(|| build_s3_store(config, bucket))? + .value() + .clone(); + + Ok((store, ObjectStorePath::from(relative))) + } + } + } +} + +#[typetag::serde(name = "ObjectStoreStorage")] +#[async_trait] +impl Storage for ObjectStoreStorage { + async fn exists(&self, path: &str) -> Result<bool> { + let (store, object_path) = self.get_store_and_path(path)?; + match store.head(&object_path).await { + Ok(_) => Ok(true), + Err(object_store::Error::NotFound { .. }) => Ok(false), + Err(e) => Err(from_object_store_error(e)), + } + } + + async fn metadata(&self, path: &str) -> Result<FileMetadata> { + let (store, object_path) = self.get_store_and_path(path)?; + let meta = store + .head(&object_path) + .await + .map_err(from_object_store_error)?; + Ok(FileMetadata { + size: meta.size as u64, + }) + } + + async fn read(&self, path: &str) -> Result<Bytes> { + let (store, object_path) = self.get_store_and_path(path)?; + let result = store + .get(&object_path) + .await + .map_err(from_object_store_error)?; + result.bytes().await.map_err(from_object_store_error) + } + + async fn reader(&self, path: &str) -> Result<Box<dyn FileRead>> { + let (store, object_path) = self.get_store_and_path(path)?; + Ok(Box::new(ObjectStoreReader { + store, + path: object_path, + })) + } + + async fn write(&self, path: &str, bs: Bytes) -> Result<()> { + let (store, object_path) = self.get_store_and_path(path)?; + store + .put(&object_path, PutPayload::from_bytes(bs)) + .await + .map_err(from_object_store_error)?; + Ok(()) + } + + async fn writer(&self, path: &str) -> Result<Box<dyn FileWrite>> { + let (store, object_path) = self.get_store_and_path(path)?; + let upload = store + .put_multipart(&object_path) + .await + .map_err(from_object_store_error)?; + let writer = WriteMultipart::new(upload); + Ok(Box::new(ObjectStoreWriter { + writer: Some(writer), + })) + } + + async fn delete(&self, path: &str) -> Result<()> { + let (store, object_path) = self.get_store_and_path(path)?; + store + .delete(&object_path) + .await + .map_err(from_object_store_error)?; + Ok(()) + } + + async fn delete_prefix(&self, path: &str) -> Result<()> { + let (store, object_path) = self.get_store_and_path(path)?; + let prefix = if object_path.as_ref().ends_with('/') { + object_path + } else { + ObjectStorePath::from(format!("{}/", object_path.as_ref())) + }; + + let mut list_stream = store.list(Some(&prefix)); + while let Some(entry) = list_stream.next().await { + let entry = entry.map_err(from_object_store_error)?; + store + .delete(&entry.location) + .await + .map_err(from_object_store_error)?; + } + Ok(()) + } + + async fn delete_stream(&self, paths: BoxStream<'static, String>) -> Result<()> { + paths + .map(Ok) + .try_for_each_concurrent(16, |path| async move { Review Comment: Same batching point as `delete_prefix` — this issues one `DeleteObject` per path capped at 16 in flight, where `DeleteObjects` takes 1,000 per request. I'd route this through `store.delete_stream(paths)` too; if we keep the concurrent form, pull the `16` out into a named const. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
