laskoviymishka commented on code in PR #3288: URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4148878844
########## crates/storage/common/tests/common/mod.rs: ########## @@ -0,0 +1,283 @@ +// 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. + +//! Shared test harness and helpers for storage integration suites. + +#![allow(dead_code)] + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use iceberg::io::{ + FileIO, FileIOBuilder, GCS_NO_AUTH, GCS_SERVICE_HOST, S3_ACCESS_KEY_ID, S3_ENDPOINT, + S3_PATH_STYLE_ACCESS, S3_REGION, S3_SECRET_ACCESS_KEY, +}; +use iceberg_storage_opendal::{OpenDalResolvingStorageFactory, OpenDalStorageFactory}; +use iceberg_test_utils::{ + get_gcs_endpoint, get_object_store_endpoint, normalize_test_name, set_up, +}; +use tempfile::TempDir; +use tokio::time::sleep; + +static FAKE_GCS_BUCKET: &str = "test-bucket"; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum StorageKind { + OpenDalS3, + OpenDalGcs, + OpenDalFs, + OpenDalMemory, + OpenDalResolving, + // TODO: Wire ObjectStoreStorage::S3 once PR #3165 is merged (https://github.com/apache/iceberg-rust/pull/3165) +} + +pub struct StorageHarness { + pub file_io: FileIO, + pub label: &'static str, + pub base_path: String, + pub _tempdirs: Vec<TempDir>, +} + +impl StorageKind { + pub const fn as_str(&self) -> &'static str { + match self { + Self::OpenDalS3 => "opendal_s3", + Self::OpenDalGcs => "opendal_gcs", + Self::OpenDalFs => "opendal_fs", + Self::OpenDalMemory => "opendal_memory", + Self::OpenDalResolving => "opendal_resolving", + } + } +} + +impl std::fmt::Display for StorageKind { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.as_str()) + } +} + +/// Fast probe to check if an endpoint service is listening before entering retry loops. +pub async fn is_endpoint_reachable(endpoint: &str) -> bool { + let Ok(client) = reqwest::Client::builder() + .timeout(Duration::from_millis(300)) + .build() + else { + return false; + }; + client.get(endpoint).send().await.is_ok() +} + +pub async fn load_storage(kind: StorageKind) -> Option<StorageHarness> { + set_up(); + match kind { + StorageKind::OpenDalS3 => load_opendal_s3().await, + StorageKind::OpenDalGcs => load_opendal_gcs().await, + StorageKind::OpenDalFs => load_opendal_fs().await, + StorageKind::OpenDalMemory => load_opendal_memory().await, + StorageKind::OpenDalResolving => load_opendal_resolving().await, + } +} + +async fn load_opendal_s3() -> Option<StorageHarness> { + let object_store_endpoint = get_object_store_endpoint(); + + if !is_endpoint_reachable(&object_store_endpoint).await { + eprintln!("Skipping S3 storage test: {object_store_endpoint} not reachable"); + return None; + } + + let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::S3 { + customized_credential_load: None, + })) + .with_props(vec![ + (S3_ENDPOINT, object_store_endpoint), + (S3_ACCESS_KEY_ID, "admin".to_string()), + (S3_SECRET_ACCESS_KEY, "password".to_string()), + (S3_REGION, "us-east-1".to_string()), + (S3_PATH_STYLE_ACCESS, "true".to_string()), + ]) + .build(); + + let mut retries = 0; + while retries < 15 { + if file_io.exists("s3://bucket1/").await.unwrap_or(false) { + return Some(StorageHarness { + file_io, + label: "opendal_s3", + base_path: "s3://bucket1/".to_string(), + _tempdirs: Vec::new(), + }); + } + sleep(Duration::from_millis(500)).await; + retries += 1; + } + + None Review Comment: `load_storage` returns `None` both when a backend is genuinely unreachable and when it's reachable but setup failed — the GCS bucket POST erroring (line 153), or this 15-attempt readiness loop exhausting — and every matrix test then does `return Ok(())`. The only trace is an `eprintln!` that cargo hides on a passing test. The practical outcome is that a broken docker-compose, a changed port, or an unhealthy container turns the whole S3/GCS/resolving suite green with zero assertions run, where the deleted tests called `get_file_io()` directly and failed hard. I'd split the two cases: only the "not configured / not reachable" path skips, and reachable-but-setup-failed returns an error (or panics) with which `kind` failed. On top of that I'd gate skips in CI behind something like `ICEBERG_REQUIRE_STORAGE=1` so a regression in the compose stack can't hide as a pass. Worth checking the GCS bucket POST status too — a 409 (already exists) is fine, a 5xx isn't. ########## crates/storage/common/tests/file_io_suite.rs: ########## @@ -0,0 +1,565 @@ +// 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. + +//! Shared FileIO integration tests parameterized over storage backends. + +mod common; + +use bytes::Bytes; +use common::{StorageHarness, StorageKind, load_storage, unique_path}; +use futures::StreamExt; +use iceberg::io::FileIO; +use rstest::rstest; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +fn roundtrip_file_io(file_io: &FileIO) -> FileIO { + let serialized = file_io.serialize_all().unwrap(); + FileIO::deserialize_all(&serialized).unwrap() +} + +// --------------------------------------------------------------------------- +// Shared Test Execution Bodies +// --------------------------------------------------------------------------- + +async fn run_exists(harness: StorageHarness) -> iceberg::Result<()> { Review Comment: These `run_*` helpers all return `iceberg::Result<()>` but `.unwrap()` every fallible call and end in `Ok(())`, so the `Result` is dead weight (the clippy `unnecessary_wraps` shape). I'd switch the `.unwrap()`s to `?` so the return type actually earns its place and a failure surfaces as a test error rather than a bare panic. ########## crates/storage/common/tests/resolving_suite.rs: ########## @@ -0,0 +1,370 @@ +// 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. + +//! OpenDAL resolving storage integration tests. + +mod common; + +use std::sync::Arc; + +use bytes::Bytes; +use common::{StorageKind, load_storage, unique_path}; +use iceberg::io::{FileIOBuilder, S3_ENDPOINT, S3_PATH_STYLE_ACCESS, S3_REGION}; +use iceberg_storage_opendal::{ + AwsCredential, CustomAwsCredentialLoader, OpenDalResolvingStorageFactory, ProvideCredential, +}; +use iceberg_test_utils::{get_object_store_endpoint, set_up}; +use reqsign_core::Context; +use rstest::rstest; + +fn temp_fs_path(name: &str) -> String { Review Comment: `temp_fs_path` uses a fixed directory and fixed filenames and only removes the file before use, so two concurrent runs (parallel CI jobs, or local alongside CI) collide, and nothing is cleaned up after. I'd use `tempfile::TempDir` here like the fs harness does. Same theme in the streaming-write and delete-prefix tests, which leave objects behind in the shared bucket — a per-run suffix or delete-on-exit keeps them hermetic. ########## crates/storage/common/tests/file_io_suite.rs: ########## @@ -0,0 +1,565 @@ +// 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. + +//! Shared FileIO integration tests parameterized over storage backends. + +mod common; + +use bytes::Bytes; +use common::{StorageHarness, StorageKind, load_storage, unique_path}; +use futures::StreamExt; +use iceberg::io::FileIO; +use rstest::rstest; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +fn roundtrip_file_io(file_io: &FileIO) -> FileIO { + let serialized = file_io.serialize_all().unwrap(); + FileIO::deserialize_all(&serialized).unwrap() +} + +// --------------------------------------------------------------------------- +// Shared Test Execution Bodies +// --------------------------------------------------------------------------- + +async fn run_exists(harness: StorageHarness) -> iceberg::Result<()> { + let non_existent = unique_path(&harness, "non_existent_file_that_does_not_exist"); + assert!(!harness.file_io.exists(&non_existent).await.unwrap()); + assert!(harness.file_io.exists(&harness.base_path).await.unwrap()); + Ok(()) +} + +async fn run_write(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_write"); + let _ = harness.file_io.delete(&path).await; + assert!(!harness.file_io.exists(&path).await.unwrap()); + + let output_file = harness.file_io.new_output(&path).unwrap(); + output_file.write("123".into()).await.unwrap(); + assert!(harness.file_io.exists(&path).await.unwrap()); + + let _ = harness.file_io.delete(&path).await; + Ok(()) +} + +async fn run_read(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_read"); + let _ = harness.file_io.delete(&path).await; + + let output_file = harness.file_io.new_output(&path).unwrap(); + output_file.write("test_input".into()).await.unwrap(); + + let input_file = harness.file_io.new_input(&path).unwrap(); + let buffer = input_file.read().await.unwrap(); + assert_eq!(buffer, "test_input".as_bytes()); + + let _ = harness.file_io.delete(&path).await; + Ok(()) +} + +async fn run_delete(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_delete"); + let _ = harness.file_io.delete(&path).await; + + harness + .file_io + .new_output(&path) + .unwrap() + .write("delete_me".into()) + .await + .unwrap(); + assert!(harness.file_io.exists(&path).await.unwrap()); + + harness.file_io.delete(&path).await.unwrap(); + assert!(!harness.file_io.exists(&path).await.unwrap()); + Ok(()) +} + +async fn run_delete_nonexistent(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_delete_nonexistent"); + harness.file_io.delete(&path).await.unwrap(); + Ok(()) +} + +async fn run_delete_stream(harness: StorageHarness) -> iceberg::Result<()> { + let base = unique_path(&harness, "test_file_io_delete_stream"); + let paths: Vec<String> = (0..5).map(|i| format!("{base}/file-{i}")).collect(); + for path in &paths { + let _ = harness.file_io.delete(path).await; + harness + .file_io + .new_output(path) + .unwrap() + .write("delete-me".into()) + .await + .unwrap(); + assert!(harness.file_io.exists(path).await.unwrap()); + } + let stream = futures::stream::iter(paths.clone()).boxed(); + harness.file_io.delete_stream(stream).await.unwrap(); + for path in &paths { + assert!(!harness.file_io.exists(path).await.unwrap()); + } + Ok(()) +} + +async fn run_delete_stream_empty(harness: StorageHarness) -> iceberg::Result<()> { + let stream = futures::stream::empty().boxed(); + harness.file_io.delete_stream(stream).await.unwrap(); + Ok(()) +} + +async fn run_metadata(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_metadata"); + let _ = harness.file_io.delete(&path).await; + let content = "metadata_test_content"; + harness + .file_io + .new_output(&path) + .unwrap() + .write(content.into()) + .await + .unwrap(); + let input_file = harness.file_io.new_input(&path).unwrap(); + let metadata = input_file.metadata().await.unwrap(); + assert_eq!(metadata.size, content.len() as u64); + let _ = harness.file_io.delete(&path).await; + Ok(()) +} + +async fn run_range_read(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_range_read"); + let _ = harness.file_io.delete(&path).await; + let content = b"0123456789abcdef"; + harness + .file_io + .new_output(&path) + .unwrap() + .write(Bytes::from_static(content)) + .await + .unwrap(); + let input_file = harness.file_io.new_input(&path).unwrap(); + let reader = input_file.reader().await.unwrap(); + let range_data = reader.read(4..10).await.unwrap(); + assert_eq!(range_data.as_ref(), &content[4..10]); + let _ = harness.file_io.delete(&path).await; + Ok(()) +} + +async fn run_zero_byte_file(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_zero_byte_file"); + let _ = harness.file_io.delete(&path).await; + + let output_file = harness.file_io.new_output(&path).unwrap(); + output_file.write(Bytes::new()).await.unwrap(); + + assert!(harness.file_io.exists(&path).await.unwrap()); + + let input_file = harness.file_io.new_input(&path).unwrap(); + let metadata = input_file.metadata().await.unwrap(); + assert_eq!(metadata.size, 0); + + let data = input_file.read().await.unwrap(); + assert_eq!(data, Bytes::new()); + + let _ = harness.file_io.delete(&path).await; + Ok(()) +} + +async fn run_delete_stream_mixed(harness: StorageHarness) -> iceberg::Result<()> { + let base = unique_path(&harness, "test_file_io_delete_stream_mixed"); + let existing_paths: Vec<String> = (0..3).map(|i| format!("{base}/exists-{i}")).collect(); + let nonexistent_paths: Vec<String> = (0..3).map(|i| format!("{base}/missing-{i}")).collect(); + + for path in &existing_paths { + let _ = harness.file_io.delete(path).await; + harness + .file_io + .new_output(path) + .unwrap() + .write("data".into()) + .await + .unwrap(); + assert!(harness.file_io.exists(path).await.unwrap()); + } + for path in &nonexistent_paths { + let _ = harness.file_io.delete(path).await; + assert!(!harness.file_io.exists(path).await.unwrap()); + } + + let mut all_paths = existing_paths.clone(); + all_paths.extend(nonexistent_paths); + + let stream = futures::stream::iter(all_paths).boxed(); + harness.file_io.delete_stream(stream).await.unwrap(); + + for path in &existing_paths { + assert!(!harness.file_io.exists(path).await.unwrap()); + } + Ok(()) +} + +async fn run_concurrent_writes(harness: StorageHarness) -> iceberg::Result<()> { + let base = unique_path(&harness, "test_file_io_concurrent_writes"); + let mut handles = Vec::new(); + + for i in 0..8 { + let file_io = harness.file_io.clone(); + let path = format!("{base}/concurrent-{i}"); + let payload = format!("payload-{i}"); + + handles.push(tokio::spawn(async move { + let output = file_io.new_output(&path).unwrap(); + output.write(payload.clone().into()).await.unwrap(); + + let input = file_io.new_input(&path).unwrap(); + let data = input.read().await.unwrap(); + assert_eq!(data, payload.as_bytes()); + + let _ = file_io.delete(&path).await; + })); + } + + for handle in handles { + handle.await.unwrap(); + } + + Ok(()) +} + +// --------------------------------------------------------------------------- +// Matrix Tests +// --------------------------------------------------------------------------- + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_exists(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_exists(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_write(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_write(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_read(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_read(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_delete(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_delete(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_delete_nonexistent(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_delete_nonexistent(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +// Note: fake-gcs-server emulator does not support batch delete (https://github.com/fsouza/fake-gcs-server/issues/1443) +#[tokio::test] +async fn test_file_io_delete_stream(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_delete_stream(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_delete_stream_empty(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_delete_stream_empty(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_delete_stream_mixed(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_delete_stream_mixed(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_delete_prefix(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + let prefix = unique_path(&harness, "test_file_io_delete_prefix"); + let paths: Vec<String> = (0..3).map(|i| format!("{prefix}/file-{i}")).collect(); + for path in &paths { + harness + .file_io + .new_output(path) + .unwrap() + .write("data".into()) + .await + .unwrap(); + assert!(harness.file_io.exists(path).await.unwrap()); + } + harness.file_io.delete_prefix(&prefix).await.unwrap(); + for path in &paths { + assert!(!harness.file_io.exists(path).await.unwrap()); + } + Ok(()) +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_delete_prefix_nonexistent(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + let prefix = unique_path(&harness, "test_file_io_delete_prefix_nonexistent"); + harness.file_io.delete_prefix(&prefix).await.unwrap(); + Ok(()) +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_metadata(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_metadata(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_metadata_nonexistent(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + let path = unique_path(&harness, "test_file_io_metadata_nonexistent"); + let input_file = harness.file_io.new_input(&path).unwrap(); + let result = input_file.metadata().await; + assert!(result.is_err()); + Ok(()) +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_range_read(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_range_read(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_range_read_out_of_bounds(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + let path = unique_path(&harness, "test_file_io_range_read_out_of_bounds"); + let _ = harness.file_io.delete(&path).await; + let content = b"0123456789"; + harness + .file_io + .new_output(&path) + .unwrap() + .write(Bytes::from_static(content)) + .await + .unwrap(); + + let input_file = harness.file_io.new_input(&path).unwrap(); + let reader = input_file.reader().await.unwrap(); + let result = reader.read(100..200).await; + assert!(result.is_err()); + + let _ = harness.file_io.delete(&path).await; + Ok(()) +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_zero_byte_file(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_zero_byte_file(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_concurrent_writes(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_concurrent_writes(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_streaming_write(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + let path = unique_path(&harness, "test_file_io_streaming_write"); + let output_file = harness.file_io.new_output(&path).unwrap(); + let mut writer = output_file.writer().await.unwrap(); + writer + .write(Bytes::from("streaming_content")) + .await + .unwrap(); + writer.close().await.unwrap(); + let input_file = harness.file_io.new_input(&path).unwrap(); + let buffer = input_file.read().await.unwrap(); + assert_eq!(buffer, Bytes::from("streaming_content")); + Ok(()) +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_streaming_write_double_close( + #[case] kind: StorageKind, +) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + let path = unique_path(&harness, "test_file_io_streaming_write_double_close"); + let output_file = harness.file_io.new_output(&path).unwrap(); + let mut writer = output_file.writer().await.unwrap(); + writer.write(Bytes::from("data")).await.unwrap(); + writer.close().await.unwrap(); + let result = writer.close().await; + assert!(result.is_err()); + Ok(()) +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] Review Comment: This matrix dropped GCS (the old `test_file_io_gcs_serialization_roundtrip`), and the resolving suite lost its roundtrip entirely — the old `test_mixed_scheme_write_and_read` ran through `roundtrip_file_io` and the new one doesn't. Both exercise distinct typetag Serialize/Deserialize paths that distributed-engine code relies on, so this is a real coverage drop, not a dedup. I'd add `#[case::opendal_gcs]` here and restore a roundtrip case for resolving. ########## crates/storage/common/src/lib.rs: ########## @@ -0,0 +1,21 @@ +// 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. + +//! Shared storage test suite for Apache Iceberg. Review Comment: `lib.rs` is doc-only, so nothing is actually shared — every suite lives in `tests/`, which cargo compiles per binary and no other crate can import, and the crate hard-depends on `iceberg-storage-opendal` with a backend-specific `StorageKind`. When #3165 lands, the object_store crate can't reuse any of this, so in practice #2211's "shared contract suite" goal isn't met; it's the opendal tests, moved. I'd expose the bodies from `src/` as generic `pub async fn run_*(file_io, base_path)` that each backend crate drives from its own `tests/`, which also drops the backend dev-dep off the library path. If it's meant to stay a single central matrix crate instead that's a fine call, but then I'd rename it and drop the "common" framing. ########## crates/storage/common/tests/credential_suite.rs: ########## @@ -0,0 +1,179 @@ +// 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. + +//! Custom AWS credential loader and FileIO builder property tests. + +mod common; + +use std::sync::Arc; + +use common::{StorageKind, load_storage}; +use iceberg::io::{ + FileIOBuilder, LocalFsStorageFactory, S3_ENDPOINT, S3_PATH_STYLE_ACCESS, S3_REGION, +}; +use iceberg_storage_opendal::{ + AwsCredential, CustomAwsCredentialLoader, OpenDalStorageFactory, ProvideCredential, +}; +use iceberg_test_utils::get_object_store_endpoint; +use reqsign_core::Context; +use rstest::rstest; + +/// Mock credential loader for testing custom AWS credential injection. +#[derive(Debug)] +struct MockCredentialLoader { + credential: Option<AwsCredential>, +} + +impl MockCredentialLoader { + fn new(credential: Option<AwsCredential>) -> Self { + Self { credential } + } + + fn new_object_store() -> Self { + Self::new(Some(AwsCredential { + access_key_id: "admin".to_string(), + secret_access_key: "password".to_string(), + session_token: None, + expires_in: None, + })) + } +} + +impl ProvideCredential for MockCredentialLoader { + type Credential = AwsCredential; + + async fn provide_credential( + &self, + _ctx: &Context, + ) -> reqsign_core::Result<Option<AwsCredential>> { + Ok(self.credential.clone()) + } +} + +#[test] +fn test_custom_aws_credential_loader_instantiation() { + let mock_loader = MockCredentialLoader::new_object_store(); + let custom_loader = CustomAwsCredentialLoader::new(mock_loader); + + let _builder = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::S3 { + customized_credential_load: Some(custom_loader), + })) + .with_props(vec![ + (S3_ENDPOINT, "http://localhost:9000".to_string()), + ("bucket", "test-bucket".to_string()), + (S3_REGION, "us-east-1".to_string()), + (S3_PATH_STYLE_ACCESS, "true".to_string()), + ]); +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[tokio::test] +async fn test_s3_with_custom_credential_loader_success( + #[case] kind: StorageKind, +) -> iceberg::Result<()> { + let Some(_harness) = load_storage(kind).await else { + return Ok(()); + }; + + let mock_loader = MockCredentialLoader::new_object_store(); + let custom_loader = CustomAwsCredentialLoader::new(mock_loader); + let object_store_endpoint = get_object_store_endpoint(); + + let file_io_with_custom_creds = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::S3 { + customized_credential_load: Some(custom_loader), + })) + .with_props(vec![ + (S3_ENDPOINT, object_store_endpoint), + (S3_REGION, "us-east-1".to_string()), + (S3_PATH_STYLE_ACCESS, "true".to_string()), + ]) + .build(); + + match file_io_with_custom_creds.exists("s3://bucket1/any").await { + Ok(_) => {} + Err(e) => panic!("Failed to check existence of bucket: {e}"), + } + + Ok(()) +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[tokio::test] +async fn test_s3_with_custom_credential_loader_failure( + #[case] kind: StorageKind, +) -> iceberg::Result<()> { + let Some(_harness) = load_storage(kind).await else { + return Ok(()); + }; + + let mock_loader = MockCredentialLoader::new(None); + let custom_loader = CustomAwsCredentialLoader::new(mock_loader); + let object_store_endpoint = get_object_store_endpoint(); + + let file_io_with_custom_creds = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::S3 { + customized_credential_load: Some(custom_loader), + })) + .with_props(vec![ + (S3_ENDPOINT, object_store_endpoint), + (S3_REGION, "us-east-1".to_string()), + (S3_PATH_STYLE_ACCESS, "true".to_string()), + ]) + .build(); + + match file_io_with_custom_creds.exists("s3://bucket1/any").await { + Ok(_) => panic!("Expected error, but got Ok"), + Err(e) => { + assert!( + e.to_string().contains("failed to load signing credential"), Review Comment: Asserting on the `"failed to load signing credential"` substring couples this to a reqsign-internal message that'll drift on upgrades. I'd assert on the error `kind()` instead, and use `let err = ...await.unwrap_err();` rather than the `match { Ok => panic!, .. }` shape. ########## crates/storage/common/tests/common/mod.rs: ########## @@ -0,0 +1,283 @@ +// 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. + +//! Shared test harness and helpers for storage integration suites. + +#![allow(dead_code)] + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use iceberg::io::{ + FileIO, FileIOBuilder, GCS_NO_AUTH, GCS_SERVICE_HOST, S3_ACCESS_KEY_ID, S3_ENDPOINT, + S3_PATH_STYLE_ACCESS, S3_REGION, S3_SECRET_ACCESS_KEY, +}; +use iceberg_storage_opendal::{OpenDalResolvingStorageFactory, OpenDalStorageFactory}; +use iceberg_test_utils::{ + get_gcs_endpoint, get_object_store_endpoint, normalize_test_name, set_up, +}; +use tempfile::TempDir; +use tokio::time::sleep; + +static FAKE_GCS_BUCKET: &str = "test-bucket"; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum StorageKind { + OpenDalS3, + OpenDalGcs, + OpenDalFs, + OpenDalMemory, + OpenDalResolving, + // TODO: Wire ObjectStoreStorage::S3 once PR #3165 is merged (https://github.com/apache/iceberg-rust/pull/3165) +} + +pub struct StorageHarness { + pub file_io: FileIO, + pub label: &'static str, + pub base_path: String, + pub _tempdirs: Vec<TempDir>, +} + +impl StorageKind { + pub const fn as_str(&self) -> &'static str { + match self { + Self::OpenDalS3 => "opendal_s3", + Self::OpenDalGcs => "opendal_gcs", + Self::OpenDalFs => "opendal_fs", + Self::OpenDalMemory => "opendal_memory", + Self::OpenDalResolving => "opendal_resolving", + } + } +} + +impl std::fmt::Display for StorageKind { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.as_str()) + } +} + +/// Fast probe to check if an endpoint service is listening before entering retry loops. +pub async fn is_endpoint_reachable(endpoint: &str) -> bool { + let Ok(client) = reqwest::Client::builder() + .timeout(Duration::from_millis(300)) + .build() + else { + return false; + }; + client.get(endpoint).send().await.is_ok() +} + +pub async fn load_storage(kind: StorageKind) -> Option<StorageHarness> { + set_up(); + match kind { + StorageKind::OpenDalS3 => load_opendal_s3().await, + StorageKind::OpenDalGcs => load_opendal_gcs().await, + StorageKind::OpenDalFs => load_opendal_fs().await, + StorageKind::OpenDalMemory => load_opendal_memory().await, + StorageKind::OpenDalResolving => load_opendal_resolving().await, + } +} + +async fn load_opendal_s3() -> Option<StorageHarness> { + let object_store_endpoint = get_object_store_endpoint(); + + if !is_endpoint_reachable(&object_store_endpoint).await { + eprintln!("Skipping S3 storage test: {object_store_endpoint} not reachable"); + return None; + } + + let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::S3 { + customized_credential_load: None, + })) + .with_props(vec![ + (S3_ENDPOINT, object_store_endpoint), + (S3_ACCESS_KEY_ID, "admin".to_string()), + (S3_SECRET_ACCESS_KEY, "password".to_string()), + (S3_REGION, "us-east-1".to_string()), + (S3_PATH_STYLE_ACCESS, "true".to_string()), + ]) + .build(); + + let mut retries = 0; + while retries < 15 { + if file_io.exists("s3://bucket1/").await.unwrap_or(false) { + return Some(StorageHarness { + file_io, + label: "opendal_s3", + base_path: "s3://bucket1/".to_string(), + _tempdirs: Vec::new(), + }); + } + sleep(Duration::from_millis(500)).await; + retries += 1; + } + + None +} + +async fn load_opendal_gcs() -> Option<StorageHarness> { + let gcs_endpoint = get_gcs_endpoint(); + + if !is_endpoint_reachable(&gcs_endpoint).await { + eprintln!("Skipping GCS storage test: {gcs_endpoint} not reachable"); + return None; + } + + let mut bucket_data = HashMap::new(); + bucket_data.insert("name", FAKE_GCS_BUCKET); + + let client = reqwest::Client::new(); + let endpoint = format!("{gcs_endpoint}/storage/v1/b"); + if client + .post(&endpoint) + .json(&bucket_data) + .send() + .await + .is_err() + { + return None; + } + + let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::Gcs)) + .with_props(vec![ + (GCS_SERVICE_HOST, gcs_endpoint), + (GCS_NO_AUTH, "true".to_string()), + ]) + .build(); + + let base_path = format!("gs://{FAKE_GCS_BUCKET}/"); + let mut retries = 0; + while retries < 15 { + if file_io.exists(&base_path).await.unwrap_or(false) { + return Some(StorageHarness { + file_io, + label: "opendal_gcs", + base_path, + _tempdirs: Vec::new(), + }); + } + sleep(Duration::from_millis(500)).await; + retries += 1; + } + + None +} + +async fn load_opendal_fs() -> Option<StorageHarness> { + let temp_dir = TempDir::new().ok()?; + let base_path = format!("file:{}/", temp_dir.path().display()); + let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::Fs)).build(); + + Some(StorageHarness { + file_io, + label: "opendal_fs", + base_path, + _tempdirs: vec![temp_dir], + }) +} + +async fn load_opendal_memory() -> Option<StorageHarness> { + let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::Memory)).build(); + + Some(StorageHarness { + file_io, + label: "opendal_memory", + base_path: "memory:/".to_string(), + _tempdirs: Vec::new(), + }) +} + +async fn load_opendal_resolving() -> Option<StorageHarness> { + let object_store_endpoint = get_object_store_endpoint(); + + if !is_endpoint_reachable(&object_store_endpoint).await { + eprintln!("Skipping Resolving storage test: {object_store_endpoint} not reachable"); + return None; + } + + let file_io = FileIOBuilder::new(Arc::new(OpenDalResolvingStorageFactory::new())) + .with_props(vec![ + (S3_ENDPOINT, object_store_endpoint), + (S3_ACCESS_KEY_ID, "admin".to_string()), + (S3_SECRET_ACCESS_KEY, "password".to_string()), + (S3_REGION, "us-east-1".to_string()), + (S3_PATH_STYLE_ACCESS, "true".to_string()), + ]) + .build(); + + let mut retries = 0; + while retries < 15 { + if file_io.exists("s3://bucket1/").await.unwrap_or(false) { + return Some(StorageHarness { + file_io, + label: "opendal_resolving", + base_path: "s3://bucket1/".to_string(), + _tempdirs: Vec::new(), + }); + } + sleep(Duration::from_millis(500)).await; + retries += 1; + } + + None +} + +pub fn unique_path(harness: &StorageHarness, test_name: &str) -> String { + format!("{}{}", harness.base_path, normalize_test_name(test_name)) +} + +#[cfg(test)] +mod endpoint_tests { Review Comment: `mod endpoint_tests` lives in `common/mod.rs`, which is `mod`-included by all three test binaries, so these probe tests compile and run three times over. One of them hits a real DNS name (`invalid-host-that-does-not-exist`, line 264), which makes it environment-dependent. I'd move them to a dedicated `tests/endpoint_probe.rs` and drop the DNS-based assertion. ########## crates/storage/common/tests/common/mod.rs: ########## @@ -0,0 +1,283 @@ +// 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. + +//! Shared test harness and helpers for storage integration suites. + +#![allow(dead_code)] + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use iceberg::io::{ + FileIO, FileIOBuilder, GCS_NO_AUTH, GCS_SERVICE_HOST, S3_ACCESS_KEY_ID, S3_ENDPOINT, + S3_PATH_STYLE_ACCESS, S3_REGION, S3_SECRET_ACCESS_KEY, +}; +use iceberg_storage_opendal::{OpenDalResolvingStorageFactory, OpenDalStorageFactory}; +use iceberg_test_utils::{ + get_gcs_endpoint, get_object_store_endpoint, normalize_test_name, set_up, +}; +use tempfile::TempDir; +use tokio::time::sleep; + +static FAKE_GCS_BUCKET: &str = "test-bucket"; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum StorageKind { + OpenDalS3, + OpenDalGcs, + OpenDalFs, + OpenDalMemory, + OpenDalResolving, + // TODO: Wire ObjectStoreStorage::S3 once PR #3165 is merged (https://github.com/apache/iceberg-rust/pull/3165) +} + +pub struct StorageHarness { + pub file_io: FileIO, + pub label: &'static str, + pub base_path: String, + pub _tempdirs: Vec<TempDir>, +} + +impl StorageKind { + pub const fn as_str(&self) -> &'static str { + match self { + Self::OpenDalS3 => "opendal_s3", + Self::OpenDalGcs => "opendal_gcs", + Self::OpenDalFs => "opendal_fs", + Self::OpenDalMemory => "opendal_memory", + Self::OpenDalResolving => "opendal_resolving", + } + } +} + +impl std::fmt::Display for StorageKind { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.as_str()) + } +} + +/// Fast probe to check if an endpoint service is listening before entering retry loops. +pub async fn is_endpoint_reachable(endpoint: &str) -> bool { + let Ok(client) = reqwest::Client::builder() + .timeout(Duration::from_millis(300)) Review Comment: 300ms is a magic timeout well under a cold CI runner's latency, so a slow-to-start backend flakes into a silent skip — which feeds straight into the false-green in the `load_storage` comment. I'd pull it into a named `const` with an env override, and factor the triplicated `while retries < 15` readiness loop (lines 117, 165, 224) into one `wait_until_ready` helper. `.send().await.is_ok()` also treats a 4xx/5xx as reachable, which is probably intended but worth a comment since it drives the skip decision. ########## crates/storage/common/tests/file_io_suite.rs: ########## @@ -0,0 +1,565 @@ +// 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. + +//! Shared FileIO integration tests parameterized over storage backends. + +mod common; + +use bytes::Bytes; +use common::{StorageHarness, StorageKind, load_storage, unique_path}; +use futures::StreamExt; +use iceberg::io::FileIO; +use rstest::rstest; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +fn roundtrip_file_io(file_io: &FileIO) -> FileIO { + let serialized = file_io.serialize_all().unwrap(); + FileIO::deserialize_all(&serialized).unwrap() +} + +// --------------------------------------------------------------------------- +// Shared Test Execution Bodies +// --------------------------------------------------------------------------- + +async fn run_exists(harness: StorageHarness) -> iceberg::Result<()> { + let non_existent = unique_path(&harness, "non_existent_file_that_does_not_exist"); + assert!(!harness.file_io.exists(&non_existent).await.unwrap()); + assert!(harness.file_io.exists(&harness.base_path).await.unwrap()); + Ok(()) +} + +async fn run_write(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_write"); + let _ = harness.file_io.delete(&path).await; + assert!(!harness.file_io.exists(&path).await.unwrap()); + + let output_file = harness.file_io.new_output(&path).unwrap(); + output_file.write("123".into()).await.unwrap(); + assert!(harness.file_io.exists(&path).await.unwrap()); + + let _ = harness.file_io.delete(&path).await; + Ok(()) +} + +async fn run_read(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_read"); + let _ = harness.file_io.delete(&path).await; + + let output_file = harness.file_io.new_output(&path).unwrap(); + output_file.write("test_input".into()).await.unwrap(); + + let input_file = harness.file_io.new_input(&path).unwrap(); + let buffer = input_file.read().await.unwrap(); + assert_eq!(buffer, "test_input".as_bytes()); + + let _ = harness.file_io.delete(&path).await; + Ok(()) +} + +async fn run_delete(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_delete"); + let _ = harness.file_io.delete(&path).await; + + harness + .file_io + .new_output(&path) + .unwrap() + .write("delete_me".into()) + .await + .unwrap(); + assert!(harness.file_io.exists(&path).await.unwrap()); + + harness.file_io.delete(&path).await.unwrap(); + assert!(!harness.file_io.exists(&path).await.unwrap()); + Ok(()) +} + +async fn run_delete_nonexistent(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_delete_nonexistent"); + harness.file_io.delete(&path).await.unwrap(); + Ok(()) +} + +async fn run_delete_stream(harness: StorageHarness) -> iceberg::Result<()> { + let base = unique_path(&harness, "test_file_io_delete_stream"); + let paths: Vec<String> = (0..5).map(|i| format!("{base}/file-{i}")).collect(); + for path in &paths { + let _ = harness.file_io.delete(path).await; + harness + .file_io + .new_output(path) + .unwrap() + .write("delete-me".into()) + .await + .unwrap(); + assert!(harness.file_io.exists(path).await.unwrap()); + } + let stream = futures::stream::iter(paths.clone()).boxed(); + harness.file_io.delete_stream(stream).await.unwrap(); + for path in &paths { + assert!(!harness.file_io.exists(path).await.unwrap()); + } + Ok(()) +} + +async fn run_delete_stream_empty(harness: StorageHarness) -> iceberg::Result<()> { + let stream = futures::stream::empty().boxed(); + harness.file_io.delete_stream(stream).await.unwrap(); + Ok(()) +} + +async fn run_metadata(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_metadata"); + let _ = harness.file_io.delete(&path).await; + let content = "metadata_test_content"; + harness + .file_io + .new_output(&path) + .unwrap() + .write(content.into()) + .await + .unwrap(); + let input_file = harness.file_io.new_input(&path).unwrap(); + let metadata = input_file.metadata().await.unwrap(); + assert_eq!(metadata.size, content.len() as u64); + let _ = harness.file_io.delete(&path).await; + Ok(()) +} + +async fn run_range_read(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_range_read"); + let _ = harness.file_io.delete(&path).await; + let content = b"0123456789abcdef"; + harness + .file_io + .new_output(&path) + .unwrap() + .write(Bytes::from_static(content)) + .await + .unwrap(); + let input_file = harness.file_io.new_input(&path).unwrap(); + let reader = input_file.reader().await.unwrap(); + let range_data = reader.read(4..10).await.unwrap(); + assert_eq!(range_data.as_ref(), &content[4..10]); + let _ = harness.file_io.delete(&path).await; + Ok(()) +} + +async fn run_zero_byte_file(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_zero_byte_file"); + let _ = harness.file_io.delete(&path).await; + + let output_file = harness.file_io.new_output(&path).unwrap(); + output_file.write(Bytes::new()).await.unwrap(); + + assert!(harness.file_io.exists(&path).await.unwrap()); + + let input_file = harness.file_io.new_input(&path).unwrap(); + let metadata = input_file.metadata().await.unwrap(); + assert_eq!(metadata.size, 0); + + let data = input_file.read().await.unwrap(); + assert_eq!(data, Bytes::new()); + + let _ = harness.file_io.delete(&path).await; + Ok(()) +} + +async fn run_delete_stream_mixed(harness: StorageHarness) -> iceberg::Result<()> { + let base = unique_path(&harness, "test_file_io_delete_stream_mixed"); + let existing_paths: Vec<String> = (0..3).map(|i| format!("{base}/exists-{i}")).collect(); + let nonexistent_paths: Vec<String> = (0..3).map(|i| format!("{base}/missing-{i}")).collect(); + + for path in &existing_paths { + let _ = harness.file_io.delete(path).await; + harness + .file_io + .new_output(path) + .unwrap() + .write("data".into()) + .await + .unwrap(); + assert!(harness.file_io.exists(path).await.unwrap()); + } + for path in &nonexistent_paths { + let _ = harness.file_io.delete(path).await; + assert!(!harness.file_io.exists(path).await.unwrap()); + } + + let mut all_paths = existing_paths.clone(); + all_paths.extend(nonexistent_paths); + + let stream = futures::stream::iter(all_paths).boxed(); + harness.file_io.delete_stream(stream).await.unwrap(); + + for path in &existing_paths { + assert!(!harness.file_io.exists(path).await.unwrap()); + } + Ok(()) +} + +async fn run_concurrent_writes(harness: StorageHarness) -> iceberg::Result<()> { + let base = unique_path(&harness, "test_file_io_concurrent_writes"); + let mut handles = Vec::new(); + + for i in 0..8 { + let file_io = harness.file_io.clone(); + let path = format!("{base}/concurrent-{i}"); + let payload = format!("payload-{i}"); + + handles.push(tokio::spawn(async move { + let output = file_io.new_output(&path).unwrap(); + output.write(payload.clone().into()).await.unwrap(); + + let input = file_io.new_input(&path).unwrap(); + let data = input.read().await.unwrap(); + assert_eq!(data, payload.as_bytes()); + + let _ = file_io.delete(&path).await; + })); + } + + for handle in handles { + handle.await.unwrap(); + } + + Ok(()) +} + +// --------------------------------------------------------------------------- +// Matrix Tests +// --------------------------------------------------------------------------- + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_exists(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_exists(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_write(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_write(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_read(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_read(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_delete(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_delete(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_delete_nonexistent(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_delete_nonexistent(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +// Note: fake-gcs-server emulator does not support batch delete (https://github.com/fsouza/fake-gcs-server/issues/1443) +#[tokio::test] +async fn test_file_io_delete_stream(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_delete_stream(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_delete_stream_empty(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_delete_stream_empty(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_delete_stream_mixed(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_delete_stream_mixed(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_delete_prefix(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + let prefix = unique_path(&harness, "test_file_io_delete_prefix"); + let paths: Vec<String> = (0..3).map(|i| format!("{prefix}/file-{i}")).collect(); + for path in &paths { + harness + .file_io + .new_output(path) + .unwrap() + .write("data".into()) + .await + .unwrap(); + assert!(harness.file_io.exists(path).await.unwrap()); + } + harness.file_io.delete_prefix(&prefix).await.unwrap(); + for path in &paths { + assert!(!harness.file_io.exists(path).await.unwrap()); + } + Ok(()) +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_delete_prefix_nonexistent(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + let prefix = unique_path(&harness, "test_file_io_delete_prefix_nonexistent"); + harness.file_io.delete_prefix(&prefix).await.unwrap(); + Ok(()) +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_metadata(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_metadata(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_metadata_nonexistent(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + let path = unique_path(&harness, "test_file_io_metadata_nonexistent"); + let input_file = harness.file_io.new_input(&path).unwrap(); + let result = input_file.metadata().await; + assert!(result.is_err()); + Ok(()) +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_range_read(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_range_read(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_range_read_out_of_bounds(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + let path = unique_path(&harness, "test_file_io_range_read_out_of_bounds"); + let _ = harness.file_io.delete(&path).await; + let content = b"0123456789"; + harness + .file_io + .new_output(&path) + .unwrap() + .write(Bytes::from_static(content)) + .await + .unwrap(); + + let input_file = harness.file_io.new_input(&path).unwrap(); + let reader = input_file.reader().await.unwrap(); + let result = reader.read(100..200).await; + assert!(result.is_err()); Review Comment: I'd not assert `is_err()` here across all four backends — the out-of-bounds read contract isn't specified, so this pins incidental OpenDAL behavior into a cross-backend assertion. S3/GCS can return 416 which OpenDAL may surface as an empty read rather than an error, fs/memory may clamp or return short, and the object_store backend coming in #3165 returns `InvalidRange`/`Generic`, which won't match. Either assert the weaker "error or empty read" here, or nail the contract down in the `FileRead` docs first and then assert it — as written this is likely to fail on at least one backend or lock in something we haven't agreed on. ########## crates/storage/common/tests/credential_suite.rs: ########## @@ -0,0 +1,179 @@ +// 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. + +//! Custom AWS credential loader and FileIO builder property tests. + +mod common; + +use std::sync::Arc; + +use common::{StorageKind, load_storage}; +use iceberg::io::{ + FileIOBuilder, LocalFsStorageFactory, S3_ENDPOINT, S3_PATH_STYLE_ACCESS, S3_REGION, +}; +use iceberg_storage_opendal::{ + AwsCredential, CustomAwsCredentialLoader, OpenDalStorageFactory, ProvideCredential, +}; +use iceberg_test_utils::get_object_store_endpoint; +use reqsign_core::Context; +use rstest::rstest; + +/// Mock credential loader for testing custom AWS credential injection. +#[derive(Debug)] +struct MockCredentialLoader { + credential: Option<AwsCredential>, +} + +impl MockCredentialLoader { + fn new(credential: Option<AwsCredential>) -> Self { + Self { credential } + } + + fn new_object_store() -> Self { + Self::new(Some(AwsCredential { + access_key_id: "admin".to_string(), + secret_access_key: "password".to_string(), + session_token: None, + expires_in: None, + })) + } +} + +impl ProvideCredential for MockCredentialLoader { + type Credential = AwsCredential; + + async fn provide_credential( + &self, + _ctx: &Context, + ) -> reqsign_core::Result<Option<AwsCredential>> { + Ok(self.credential.clone()) + } +} + +#[test] +fn test_custom_aws_credential_loader_instantiation() { Review Comment: This test asserts nothing — `_builder` is constructed and dropped — so it only checks that the code compiles. That plus the three `test_file_io_builder_*` tests below exercise `iceberg` core through `LocalFsStorageFactory`, not a storage backend, so they belong in `crates/iceberg` rather than this suite. The builder tests are also `async` with no `.await`; plain `#[test]` fits. ########## crates/storage/common/Cargo.toml: ########## @@ -0,0 +1,49 @@ +# 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 } +homepage = { workspace = true } +license = { workspace = true } +name = "iceberg-storage-common" +repository = { workspace = true } +rust-version = { workspace = true } +version = { workspace = true } + +categories = ["database"] +description = "Apache Iceberg Storage Common Test Suite" +keywords = ["iceberg", "storage"] +publish = false + +[dependencies] +iceberg = { workspace = true } Review Comment: `iceberg` is in both `[dependencies]` (here) and `[dev-dependencies]` (line 39); since this is a test-only crate the `[dependencies]` entry looks unused, so I'd drop it. And `reqsign-core = { version = "3.0.0" }` (line 42) is pinned literally instead of taken from the workspace like everything else — I'd move it to `{ workspace = true }`. ########## crates/storage/common/tests/file_io_suite.rs: ########## @@ -0,0 +1,565 @@ +// 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. + +//! Shared FileIO integration tests parameterized over storage backends. + +mod common; + +use bytes::Bytes; +use common::{StorageHarness, StorageKind, load_storage, unique_path}; +use futures::StreamExt; +use iceberg::io::FileIO; +use rstest::rstest; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +fn roundtrip_file_io(file_io: &FileIO) -> FileIO { + let serialized = file_io.serialize_all().unwrap(); + FileIO::deserialize_all(&serialized).unwrap() +} + +// --------------------------------------------------------------------------- +// Shared Test Execution Bodies +// --------------------------------------------------------------------------- + +async fn run_exists(harness: StorageHarness) -> iceberg::Result<()> { + let non_existent = unique_path(&harness, "non_existent_file_that_does_not_exist"); + assert!(!harness.file_io.exists(&non_existent).await.unwrap()); + assert!(harness.file_io.exists(&harness.base_path).await.unwrap()); + Ok(()) +} + +async fn run_write(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_write"); + let _ = harness.file_io.delete(&path).await; + assert!(!harness.file_io.exists(&path).await.unwrap()); + + let output_file = harness.file_io.new_output(&path).unwrap(); + output_file.write("123".into()).await.unwrap(); + assert!(harness.file_io.exists(&path).await.unwrap()); + + let _ = harness.file_io.delete(&path).await; + Ok(()) +} + +async fn run_read(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_read"); + let _ = harness.file_io.delete(&path).await; + + let output_file = harness.file_io.new_output(&path).unwrap(); + output_file.write("test_input".into()).await.unwrap(); + + let input_file = harness.file_io.new_input(&path).unwrap(); + let buffer = input_file.read().await.unwrap(); + assert_eq!(buffer, "test_input".as_bytes()); + + let _ = harness.file_io.delete(&path).await; + Ok(()) +} + +async fn run_delete(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_delete"); + let _ = harness.file_io.delete(&path).await; + + harness + .file_io + .new_output(&path) + .unwrap() + .write("delete_me".into()) + .await + .unwrap(); + assert!(harness.file_io.exists(&path).await.unwrap()); + + harness.file_io.delete(&path).await.unwrap(); + assert!(!harness.file_io.exists(&path).await.unwrap()); + Ok(()) +} + +async fn run_delete_nonexistent(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_delete_nonexistent"); + harness.file_io.delete(&path).await.unwrap(); + Ok(()) +} + +async fn run_delete_stream(harness: StorageHarness) -> iceberg::Result<()> { + let base = unique_path(&harness, "test_file_io_delete_stream"); + let paths: Vec<String> = (0..5).map(|i| format!("{base}/file-{i}")).collect(); + for path in &paths { + let _ = harness.file_io.delete(path).await; + harness + .file_io + .new_output(path) + .unwrap() + .write("delete-me".into()) + .await + .unwrap(); + assert!(harness.file_io.exists(path).await.unwrap()); + } + let stream = futures::stream::iter(paths.clone()).boxed(); + harness.file_io.delete_stream(stream).await.unwrap(); + for path in &paths { + assert!(!harness.file_io.exists(path).await.unwrap()); + } + Ok(()) +} + +async fn run_delete_stream_empty(harness: StorageHarness) -> iceberg::Result<()> { + let stream = futures::stream::empty().boxed(); + harness.file_io.delete_stream(stream).await.unwrap(); + Ok(()) +} + +async fn run_metadata(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_metadata"); + let _ = harness.file_io.delete(&path).await; + let content = "metadata_test_content"; + harness + .file_io + .new_output(&path) + .unwrap() + .write(content.into()) + .await + .unwrap(); + let input_file = harness.file_io.new_input(&path).unwrap(); + let metadata = input_file.metadata().await.unwrap(); + assert_eq!(metadata.size, content.len() as u64); + let _ = harness.file_io.delete(&path).await; + Ok(()) +} + +async fn run_range_read(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_range_read"); + let _ = harness.file_io.delete(&path).await; + let content = b"0123456789abcdef"; + harness + .file_io + .new_output(&path) + .unwrap() + .write(Bytes::from_static(content)) + .await + .unwrap(); + let input_file = harness.file_io.new_input(&path).unwrap(); + let reader = input_file.reader().await.unwrap(); + let range_data = reader.read(4..10).await.unwrap(); + assert_eq!(range_data.as_ref(), &content[4..10]); + let _ = harness.file_io.delete(&path).await; + Ok(()) +} + +async fn run_zero_byte_file(harness: StorageHarness) -> iceberg::Result<()> { + let path = unique_path(&harness, "test_file_io_zero_byte_file"); + let _ = harness.file_io.delete(&path).await; + + let output_file = harness.file_io.new_output(&path).unwrap(); + output_file.write(Bytes::new()).await.unwrap(); + + assert!(harness.file_io.exists(&path).await.unwrap()); + + let input_file = harness.file_io.new_input(&path).unwrap(); + let metadata = input_file.metadata().await.unwrap(); + assert_eq!(metadata.size, 0); + + let data = input_file.read().await.unwrap(); + assert_eq!(data, Bytes::new()); + + let _ = harness.file_io.delete(&path).await; + Ok(()) +} + +async fn run_delete_stream_mixed(harness: StorageHarness) -> iceberg::Result<()> { + let base = unique_path(&harness, "test_file_io_delete_stream_mixed"); + let existing_paths: Vec<String> = (0..3).map(|i| format!("{base}/exists-{i}")).collect(); + let nonexistent_paths: Vec<String> = (0..3).map(|i| format!("{base}/missing-{i}")).collect(); + + for path in &existing_paths { + let _ = harness.file_io.delete(path).await; + harness + .file_io + .new_output(path) + .unwrap() + .write("data".into()) + .await + .unwrap(); + assert!(harness.file_io.exists(path).await.unwrap()); + } + for path in &nonexistent_paths { + let _ = harness.file_io.delete(path).await; + assert!(!harness.file_io.exists(path).await.unwrap()); + } + + let mut all_paths = existing_paths.clone(); + all_paths.extend(nonexistent_paths); + + let stream = futures::stream::iter(all_paths).boxed(); + harness.file_io.delete_stream(stream).await.unwrap(); + + for path in &existing_paths { + assert!(!harness.file_io.exists(path).await.unwrap()); + } + Ok(()) +} + +async fn run_concurrent_writes(harness: StorageHarness) -> iceberg::Result<()> { + let base = unique_path(&harness, "test_file_io_concurrent_writes"); + let mut handles = Vec::new(); + + for i in 0..8 { + let file_io = harness.file_io.clone(); + let path = format!("{base}/concurrent-{i}"); + let payload = format!("payload-{i}"); + + handles.push(tokio::spawn(async move { + let output = file_io.new_output(&path).unwrap(); + output.write(payload.clone().into()).await.unwrap(); + + let input = file_io.new_input(&path).unwrap(); + let data = input.read().await.unwrap(); + assert_eq!(data, payload.as_bytes()); + + let _ = file_io.delete(&path).await; + })); + } + + for handle in handles { + handle.await.unwrap(); + } + + Ok(()) +} + +// --------------------------------------------------------------------------- +// Matrix Tests +// --------------------------------------------------------------------------- + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_exists(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_exists(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_write(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_write(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_read(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_read(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_delete(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_delete(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_delete_nonexistent(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_delete_nonexistent(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +// Note: fake-gcs-server emulator does not support batch delete (https://github.com/fsouza/fake-gcs-server/issues/1443) +#[tokio::test] +async fn test_file_io_delete_stream(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_delete_stream(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_delete_stream_empty(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_delete_stream_empty(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_delete_stream_mixed(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_delete_stream_mixed(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_delete_prefix(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + let prefix = unique_path(&harness, "test_file_io_delete_prefix"); + let paths: Vec<String> = (0..3).map(|i| format!("{prefix}/file-{i}")).collect(); + for path in &paths { + harness + .file_io + .new_output(path) + .unwrap() + .write("data".into()) + .await + .unwrap(); + assert!(harness.file_io.exists(path).await.unwrap()); + } + harness.file_io.delete_prefix(&prefix).await.unwrap(); + for path in &paths { + assert!(!harness.file_io.exists(path).await.unwrap()); + } + Ok(()) +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_delete_prefix_nonexistent(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + let prefix = unique_path(&harness, "test_file_io_delete_prefix_nonexistent"); + harness.file_io.delete_prefix(&prefix).await.unwrap(); + Ok(()) +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_metadata(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + run_metadata(harness).await +} + +#[rstest] +#[case::opendal_s3(StorageKind::OpenDalS3)] +#[case::opendal_gcs(StorageKind::OpenDalGcs)] +#[case::opendal_fs(StorageKind::OpenDalFs)] +#[case::opendal_memory(StorageKind::OpenDalMemory)] +#[tokio::test] +async fn test_file_io_metadata_nonexistent(#[case] kind: StorageKind) -> iceberg::Result<()> { + let Some(harness) = load_storage(kind).await else { + return Ok(()); + }; + let path = unique_path(&harness, "test_file_io_metadata_nonexistent"); + let input_file = harness.file_io.new_input(&path).unwrap(); + let result = input_file.metadata().await; + assert!(result.is_err()); Review Comment: `metadata` on a missing path has a meaningful `NotFound` contract, but `is_err()` passes on any error. I'd assert `ErrorKind::NotFound` here. Same for the double-close test (line 535) — if we're asserting it errors, assert the kind rather than just that it's an error. -- 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]
