Feature/s3tables

.github/copilot-instructions.md

@@ -116,7 +116,7 @@ cargo run --example file_uploader
 cargo run --example file_downloader
 cargo run --example object_prompt
 ```
-- Examples require network access to MinIO (defaults to play.min.io)
+- Examples require network access to MinIO (defaults to localhost:9000)
 - Will fail if network is unavailable
 - Located in `examples/` directory (10 examples)
 
@@ -194,7 +194,7 @@ use minio::s3::{MinioClient, MinioClientBuilder};
 use minio::s3::creds::StaticProvider;
 use minio::s3::http::BaseUrl;
 
-let base_url = "play.min.io".parse::<BaseUrl>()?;
+let base_url = "http://localhost:9000".parse::<BaseUrl>()?;
 let provider = StaticProvider::new("access_key", "secret_key", None);
 let client = MinioClientBuilder::new(base_url)
     .provider(Some(provider))
@@ -278,7 +278,7 @@ async fn my_test(ctx: TestContext, bucket_name: String) {
 **Solution**: These warnings are acceptable and don't need to be fixed for PRs
 
 ### Test Environment Variables
-When running tests locally without CI=true, tests use play.min.io by default. To use local MinIO:
+When running tests locally without CI=true, tests use localhost:9000 by default. To use a different MinIO server:
 ```bash
 export CI=true
 export SERVER_ENDPOINT=localhost:9000

Cargo.toml

@@ -10,6 +10,9 @@ readme = "README.md"
 keywords = ["object-storage", "minio", "s3"]
 categories = ["api-bindings", "web-programming::http-client"]
 
+[package.metadata.docs.rs]
+features = ["datafusion", "puffin-compression"]
+
 [features]
 default = ["default-tls", "default-crypto", "http2"]
 default-tls = ["reqwest/default-tls"]
@@ -22,9 +25,13 @@ ring = ["dep:ring"]
 # Gracefully falls back to HTTP/1.1 when the server doesn't support it.
 http2 = ["reqwest/http2"]
 localhost = []
+# Puffin compression support for Iceberg table compression
+puffin-compression = ["dep:zstd", "dep:lz4_flex"]
+# DataFusion integration for query pushdown support
+datafusion = ["dep:datafusion", "dep:arrow", "dep:parquet", "dep:object_store", "dep:tokio"]
 
 [workspace.dependencies]
-uuid = "1.18"
+uuid = "1.19"
 futures-util = "0.3"
 futures-io = "0.3"
 reqwest = { version = "0.12", default-features = false }
@@ -49,7 +56,7 @@ async-stream = "0.3"
 async-trait = "0.1"
 base64 = "0.22"
 chrono = { workspace = true, features = ["serde"] }
-crc = "3.4"
+crc-fast = "1.8"
 dashmap = "6.1.0"
 env_logger = "0.11"
 hmac = { version = "0.12", optional = true }
@@ -58,18 +65,30 @@ lazy_static = "1.5"
 log = { workspace = true }
 md5 = "0.8"
 multimap = "0.10"
+once_cell = "1.21"
 percent-encoding = "2.3"
 url = "2.5"
 regex = "1.12"
 ring = { version = "0.17", optional = true, default-features = false, features = ["alloc"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
+sha1 = "0.10"
 sha2 = { version = "0.10", optional = true }
 urlencoding = "2.1"
 xmltree = "0.12"
 http = { workspace = true }
 thiserror = "2.0"
 typed-builder = "0.23"
+# DataFusion integration (optional, for query pushdown)
+datafusion = { version = "51.0", optional = true }
+arrow = { version = "57.1", optional = true }
+parquet = { version = "57.1", features = ["snap"], optional = true }
+object_store = { version = "0.12", optional = true }
+tokio = { workspace = true, optional = true, features = ["rt-multi-thread"] }
+# Puffin compression (optional, for Iceberg table compression)
+zstd = { version = "0.13", optional = true }
+lz4_flex = { version = "0.11", optional = true }
+plotters = "0.3.7"
 
 [dev-dependencies]
 minio-common = { path = "./common" }
@@ -80,6 +99,17 @@ clap = { version = "4.5", features = ["derive"] }
 rand = { workspace = true, features = ["small_rng"] }
 quickcheck = "1.0"
 criterion = "0.8"
+# DataFusion benchmark dependencies (also available as optional feature)
+object_store = { version = "0.12", features = ["aws"] }
+futures = "0.3"
+# Iceberg-rust for proper manifest file creation in benchmarks
+iceberg = { version = "0.7", features = ["storage-s3"] }
+iceberg-catalog-rest = "0.7"
+# Arrow/Parquet versions matching iceberg-rust 0.7 (v55.1)
+# Use package aliasing to avoid conflicts with datafusion's arrow/parquet
+arrow-array-55 = { version = "55.1", package = "arrow-array" }
+arrow-schema-55 = { version = "55.1", package = "arrow-schema" }
+parquet-55 = { version = "55.1", package = "parquet", features = ["async"] }
 
 [lib]
 name = "minio"
@@ -100,7 +130,51 @@ name = "append_object"
 [[example]]
 name = "load_balancing_with_hooks"
 
+[[example]]
+name = "tables_quickstart"
+path = "examples/s3tables/tables_quickstart.rs"
+
+[[example]]
+name = "tables_stress_throughput_saturation"
+path = "examples/s3tables/tables_stress_throughput_saturation.rs"
+
+[[example]]
+name = "tables_stress_sustained_load"
+path = "examples/s3tables/tables_stress_sustained_load.rs"
+
+[[example]]
+name = "tables_stress_state_chaos"
+path = "examples/s3tables/tables_stress_state_chaos.rs"
+
+[[example]]
+name = "sdk_comparison_benchmark"
+path = "examples/datafusion/sdk_comparison_benchmark.rs"
+required-features = ["datafusion"]
+
+[[example]]
+name = "s3tables_pushdown_benchmark"
+path = "examples/datafusion/s3tables_pushdown_benchmark.rs"
+required-features = ["datafusion"]
+
+[[example]]
+name = "simd_benchmark"
+path = "examples/s3tables/simd_benchmark.rs"
+
+[[example]]
+name = "simd_benchmark_full"
+path = "examples/s3tables/simd_benchmark_full.rs"
+required-features = ["datafusion"]
+
+[[example]]
+name = "storage_pushdown_benchmark"
+path = "examples/s3tables/storage_pushdown_benchmark.rs"
+
 [[bench]]
 name = "s3-api"
 path = "benches/s3/api_benchmarks.rs"
 harness = false
+
+[[bench]]
+name = "bench_checksums"
+path = "benches/s3/bench_checksums.rs"
+harness = false

README.md

@@ -26,8 +26,8 @@ use minio::s3::response::BucketExistsResponse;
 
 #[tokio::main]
 async fn main() {
-    let base_url = "play.min.io".parse::<BaseUrl>().unwrap();
-    let static_provider = StaticProvider::new("Q3AM3UQ867SPQQA43P2F", "zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG", None);
+    let base_url = "http://localhost:9000".parse::<BaseUrl>().unwrap();
+    let static_provider = StaticProvider::new("minioadmin", "minioadmin", None);
     let client = MinioClient::new(base_url, Some(static_provider), None, None).unwrap();
 
     let exists: BucketExistsResponse = client

benches/s3/bench_bucket_lifecycle.rs

@@ -43,7 +43,7 @@ pub(crate) fn bench_get_bucket_lifecycle(criterion: &mut Criterion) {
             let ctx = Ctx2::new().await;
             let config = create_bucket_lifecycle_config_examples();
             ctx.client
-                .put_bucket_lifecycle(&ctx.bucket)
+                .put_bucket_lifecycle(ctx.bucket.clone())
                 .life_cycle_config(config)
                 .build()
                 .send()

benches/s3/bench_bucket_notification.rs

@@ -45,7 +45,7 @@ pub(crate) fn bench_get_bucket_notification(criterion: &mut Criterion) {
             let ctx = Ctx2::new().await;
             let config = create_bucket_notification_config_example();
             ctx.client
-                .put_bucket_notification(&ctx.bucket)
+                .put_bucket_notification(ctx.bucket.clone())
                 .notification_config(config)
                 .build()
                 .send()

benches/s3/bench_bucket_replication.rs

@@ -34,7 +34,7 @@ pub(crate) fn bench_put_bucket_replication(criterion: &mut Criterion) {
 
             let _resp: PutBucketVersioningResponse = ctx
                 .client
-                .put_bucket_versioning(&ctx.bucket)
+                .put_bucket_versioning(ctx.bucket.clone())
                 .versioning_status(VersioningStatus::Enabled)
                 .build()
                 .send()
@@ -53,8 +53,7 @@ pub(crate) fn bench_put_bucket_replication(criterion: &mut Criterion) {
             ctx
         },
         |ctx| {
-            let config =
-                create_bucket_replication_config_example(ctx.aux_bucket.clone().unwrap().as_str());
+            let config = create_bucket_replication_config_example(&ctx.aux_bucket.clone().unwrap());
             PutBucketReplication::builder()
                 .client(ctx.client.clone())
                 .bucket(ctx.bucket.clone())
@@ -74,7 +73,7 @@ pub(crate) fn bench_get_bucket_replication(criterion: &mut Criterion) {
 
             let _resp: PutBucketVersioningResponse = ctx
                 .client
-                .put_bucket_versioning(&ctx.bucket)
+                .put_bucket_versioning(ctx.bucket.clone())
                 .versioning_status(VersioningStatus::Enabled)
                 .build()
                 .send()
@@ -111,7 +110,7 @@ pub(crate) fn bench_delete_bucket_replication(criterion: &mut Criterion) {
 
             let _resp: PutBucketVersioningResponse = ctx
                 .client
-                .put_bucket_versioning(&ctx.bucket)
+                .put_bucket_versioning(ctx.bucket.clone())
                 .versioning_status(VersioningStatus::Enabled)
                 .build()
                 .send()

benches/s3/bench_bucket_tagging.rs

@@ -47,7 +47,7 @@ pub(crate) async fn bench_get_bucket_tagging(criterion: &mut Criterion) {
         || async {
             let ctx = Ctx2::new().await;
             ctx.client
-                .put_bucket_tagging(&ctx.bucket)
+                .put_bucket_tagging(ctx.bucket.clone())
                 .tags(create_tags_example())
                 .build()
                 .send()

benches/s3/bench_checksums.rs

@@ -0,0 +1,53 @@
+// MinIO Rust Library for Amazon S3 Compatible Cloud Storage
+// Copyright 2025 MinIO, Inc.
+//
+// Licensed 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 criterion::{Criterion, Throughput, criterion_group, criterion_main};
+use minio::s3::utils::{
+    crc32_checksum, crc32c, crc64nvme_checksum, md5sum_hash, sha1_hash, sha256_checksum,
+};
+
+fn bench_checksums(c: &mut Criterion) {
+    let sizes = vec![
+        ("1KB", 1024),
+        ("10KB", 10 * 1024),
+        ("100KB", 100 * 1024),
+        ("1MB", 1024 * 1024),
+        ("10MB", 10 * 1024 * 1024),
+    ];
+
+    for (name, size) in sizes {
+        let data = vec![0u8; size];
+
+        let mut group = c.benchmark_group(format!("checksum_{}", name));
+        group.throughput(Throughput::Bytes(size as u64));
+
+        group.bench_function("CRC32", |b| b.iter(|| crc32_checksum(&data)));
+
+        group.bench_function("CRC32C", |b| b.iter(|| crc32c(&data)));
+
+        group.bench_function("CRC64NVME", |b| b.iter(|| crc64nvme_checksum(&data)));
+
+        group.bench_function("MD5", |b| b.iter(|| md5sum_hash(&data)));
+
+        group.bench_function("SHA1", |b| b.iter(|| sha1_hash(&data)));
+
+        group.bench_function("SHA256", |b| b.iter(|| sha256_checksum(&data)));
+
+        group.finish();
+    }
+}
+
+criterion_group!(benches, bench_checksums);
+criterion_main!(benches);

benches/s3/bench_object_append.rs

@@ -44,7 +44,7 @@ pub(crate) async fn bench_object_append(criterion: &mut Criterion) {
                     tokio::runtime::Runtime::new().map_err(|e| Error::DriveIo(e.into()))?;
                 runtime.block_on(
                     ctx.client
-                        .stat_object(&ctx.bucket, &ctx.object)
+                        .stat_object(ctx.bucket.clone(), ctx.object.clone())
                         .build()
                         .send(),
                 )

benches/s3/bench_object_copy.rs

@@ -25,7 +25,6 @@ pub(crate) fn bench_object_copy_internal(criterion: &mut Criterion) {
         criterion,
         || async { Ctx2::new_with_object(false).await },
         |ctx| {
-            let object_name_src = &ctx.object;
             let object_name_dst = rand_object_name();
             CopyObjectInternal::builder()
                 .client(ctx.client.clone())
@@ -34,7 +33,7 @@ pub(crate) fn bench_object_copy_internal(criterion: &mut Criterion) {
                 .source(
                     CopySource::builder()
                         .bucket(ctx.bucket.clone())
-                        .object(object_name_src)
+                        .object(ctx.object.clone())
                         .build(),
                 )
                 .build()

benches/s3/bench_object_legal_hold.rs

@@ -47,7 +47,7 @@ pub(crate) async fn bench_get_object_legal_hold(criterion: &mut Criterion) {
         || async {
             let ctx = Ctx2::new_with_object(true).await;
             ctx.client
-                .get_object_legal_hold(&ctx.bucket, &ctx.object)
+                .get_object_legal_hold(ctx.bucket.clone(), ctx.object.clone())
                 .build()
                 .send()
                 .await

benches/s3/bench_object_put.rs

@@ -29,7 +29,7 @@ pub(crate) fn bench_object_put(criterion: &mut Criterion) {
         criterion,
         || async { Ctx2::new().await },
         |ctx| {
-            let object_name: String = rand_object_name();
+            let object_name = rand_object_name();
             let size = 1024 * 1024_u64; // 1MB
             let object_content = ObjectContent::new_from_stream(RandSrc::new(size), Some(size));
 

benches/s3/bench_object_tagging.rs

@@ -51,7 +51,7 @@ pub(crate) async fn bench_get_object_tagging(criterion: &mut Criterion) {
 
             let _resp: PutObjectTaggingResponse = ctx
                 .client
-                .put_object_tagging(&ctx.bucket, &ctx.object)
+                .put_object_tagging(ctx.bucket.clone(), ctx.object.clone())
                 .tags(create_tags_example())
                 .build()
                 .send()