feat(storage): Implement ObjectStoreStorage::S3 (Supersedes #2257)

Cargo.toml

@@ -75,6 +75,7 @@ bytes = "1.11"
 cfg-if = "1"
 chrono = "0.4.41"
 crc32fast = "1"
+dashmap = "6.1"
 derive_builder = "0.20"
 expect-test = "1"
 fastnum = { version = "0.7", default-features = false, features = [
@@ -95,6 +96,7 @@ iceberg-catalog-rest = { version = "0.10.0", path = "./crates/catalog/rest" }
 iceberg-catalog-s3tables = { version = "0.10.0", path = "./crates/catalog/s3tables" }
 iceberg-catalog-sql = { version = "0.10.0", path = "./crates/catalog/sql" }
 iceberg-property-macro = { version = "0.10.0", path = "./crates/property-macro" }
+iceberg-storage-object_store = { version = "0.10.0", path = "./crates/storage/object_store" }
 iceberg-storage-opendal = { version = "0.10.0", path = "./crates/storage/opendal" }
 itertools = "0.13"
 linkedbytes = "0.1.8"
@@ -104,10 +106,12 @@ mockall = "0.13.1"
 mockito = "1"
 motore-macros = "0.4.3"
 murmur3 = "0.5.2"
+object_store = "0.13"
 once_cell = "1.20"
 opendal = "0.58"
 ordered-float = "4"
 parquet = "59.2"
+percent-encoding = "2.3"
 pilota = "0.11.10"
 pretty_assertions = "1.4"
 proc-macro2 = "1"
@@ -118,6 +122,10 @@ regex = "1.11.3"
 reqwest = { version = "0.12.12", default-features = false, features = ["json"] }
 roaring = { version = "0.11" }
 rstest = "0.26"
+rustls = "0.23.45"
+rustls-native-certs = "0.8"
+rustls-pki-types = "1.15"
+rustls-webpki = "0.103.15"
 serde = { version = "1.0.219", features = ["rc"] }
 serde_bytes = "0.11.17"
 serde_derive = "1.0.219"

crates/storage/object_store/Cargo.toml

@@ -0,0 +1,54 @@
+# 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 = false
+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 }
+percent-encoding = { workspace = true }
+serde = { workspace = true, features = ["derive"] }
+tokio = { workspace = true }
+tracing = { workspace = true }
+typetag = { workspace = true }
+url = { workspace = true }
+
+[dev-dependencies]
+iceberg_test_utils = { path = "../../test_utils", features = ["tests"] }
+serde_json = { workspace = true }
+tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
+
+[lints]
+workspace = true

crates/storage/object_store/LICENSE

@@ -0,0 +1 @@
+../../../LICENSE

crates/storage/object_store/NOTICE

@@ -0,0 +1 @@
+../../../NOTICE

crates/storage/object_store/public-api.txt

@@ -0,0 +1,44 @@
+pub mod iceberg_storage_object_store
+pub enum iceberg_storage_object_store::ObjectStoreStorage
+pub iceberg_storage_object_store::ObjectStoreStorage::S3(iceberg_storage_object_store::S3Storage)
+impl core::clone::Clone for iceberg_storage_object_store::ObjectStoreStorage
+pub fn iceberg_storage_object_store::ObjectStoreStorage::clone(&self) -> iceberg_storage_object_store::ObjectStoreStorage
+impl core::fmt::Debug for iceberg_storage_object_store::ObjectStoreStorage
+pub fn iceberg_storage_object_store::ObjectStoreStorage::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+impl iceberg::io::storage::Storage for iceberg_storage_object_store::ObjectStoreStorage
+pub fn iceberg_storage_object_store::ObjectStoreStorage::delete<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait
+pub fn iceberg_storage_object_store::ObjectStoreStorage::delete_prefix<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait
+pub fn iceberg_storage_object_store::ObjectStoreStorage::delete_stream<'life0, 'async_trait>(&'life0 self, paths: futures_core::stream::BoxStream<'static, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait
+pub fn iceberg_storage_object_store::ObjectStoreStorage::exists<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<bool>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait
+pub fn iceberg_storage_object_store::ObjectStoreStorage::metadata<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<iceberg::io::file_io::FileMetadata>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait
+pub fn iceberg_storage_object_store::ObjectStoreStorage::new_input(&self, path: &str) -> iceberg::error::Result<iceberg::io::file_io::InputFile>
+pub fn iceberg_storage_object_store::ObjectStoreStorage::new_output(&self, path: &str) -> iceberg::error::Result<iceberg::io::file_io::OutputFile>
+pub fn iceberg_storage_object_store::ObjectStoreStorage::read<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<bytes::bytes::Bytes>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait
+pub fn iceberg_storage_object_store::ObjectStoreStorage::reader<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg::io::file_io::FileRead>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait
+pub fn iceberg_storage_object_store::ObjectStoreStorage::write<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str, bs: bytes::bytes::Bytes) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait
+pub fn iceberg_storage_object_store::ObjectStoreStorage::writer<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg::io::file_io::FileWrite>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait
+impl serde_core::ser::Serialize for iceberg_storage_object_store::ObjectStoreStorage
+pub fn iceberg_storage_object_store::ObjectStoreStorage::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer
+impl<'de> serde_core::de::Deserialize<'de> for iceberg_storage_object_store::ObjectStoreStorage
+pub fn iceberg_storage_object_store::ObjectStoreStorage::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
+pub enum iceberg_storage_object_store::ObjectStoreStorageFactory
+pub iceberg_storage_object_store::ObjectStoreStorageFactory::S3
+impl core::clone::Clone for iceberg_storage_object_store::ObjectStoreStorageFactory
+pub fn iceberg_storage_object_store::ObjectStoreStorageFactory::clone(&self) -> iceberg_storage_object_store::ObjectStoreStorageFactory
+impl core::fmt::Debug for iceberg_storage_object_store::ObjectStoreStorageFactory
+pub fn iceberg_storage_object_store::ObjectStoreStorageFactory::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+impl iceberg::io::storage::StorageFactory for iceberg_storage_object_store::ObjectStoreStorageFactory
+pub fn iceberg_storage_object_store::ObjectStoreStorageFactory::build(&self, config: &iceberg::io::storage::config::StorageConfig) -> iceberg::error::Result<alloc::sync::Arc<dyn iceberg::io::storage::Storage>>
+impl serde_core::ser::Serialize for iceberg_storage_object_store::ObjectStoreStorageFactory
+pub fn iceberg_storage_object_store::ObjectStoreStorageFactory::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer
+impl<'de> serde_core::de::Deserialize<'de> for iceberg_storage_object_store::ObjectStoreStorageFactory
+pub fn iceberg_storage_object_store::ObjectStoreStorageFactory::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
+pub struct iceberg_storage_object_store::S3Storage
+impl core::clone::Clone for iceberg_storage_object_store::S3Storage
+pub fn iceberg_storage_object_store::S3Storage::clone(&self) -> iceberg_storage_object_store::S3Storage
+impl core::fmt::Debug for iceberg_storage_object_store::S3Storage
+pub fn iceberg_storage_object_store::S3Storage::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+impl serde_core::ser::Serialize for iceberg_storage_object_store::S3Storage
+pub fn iceberg_storage_object_store::S3Storage::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer
+impl<'de> serde_core::de::Deserialize<'de> for iceberg_storage_object_store::S3Storage
+pub fn iceberg_storage_object_store::S3Storage::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>