You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "Sruhvx-jpg (via GitHub)" <gi...@apache.org> on 2026/09/28 13:22:17 UTC
[PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Sruhvx-jpg opened a new pull request, #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288
## Which issue does this PR close?
- Closes #2211
- Supersedes and revives #2253
## What changes are included in this PR?
- **New Crate `iceberg-storage-common`**: Defines a shared parameterized integration test suite (`file_io_suite`, `resolving_suite`, `credential_suite`) to enforce `FileIO` contracts across storage implementations.
- **De-duplicate OpenDAL Tests**: Removes duplicate integration test files from `crates/storage/opendal` while keeping HuggingFace tests isolated to prevent CI secret leak risks.
- **Fast Probe Health Check**: Adds a 300ms TCP/HTTP endpoint probe before retry loops so tests fail-fast / skip cleanly (0ms) when Docker emulators are offline.
- **Extra Contract Assertions**:
- `test_file_io_zero_byte_file` (0-byte write/read/exists)
- `test_file_io_range_read_out_of_bounds` (enforce error contract on range OOB)
- `test_file_io_delete_stream_mixed` (batch delete with mixed existing/non-existent keys)
- `test_file_io_concurrent_writes` (concurrent multi-task writers on shared FileIO)
- `test_file_io_serialization_roundtrip` (FileIO serde roundtrip)
- **Forward-Compatibility**: Added `TODO` tracking #3165 (approved `object_store` S3 backend) to plug into the `StorageKind` matrix upon merge.
## Are these changes tested?
Tested across S3 (RustFS/MinIO), GCS emulator, local FS, and in-memory storage:
- `cargo test -p iceberg-storage-common` (92 tests passed, 0 failed)
- `cargo test -p iceberg-storage-opendal` (all unit & integration tests pass)
- `cargo clippy` & `cargo fmt` clean.
## Generative AI Disclosure
This contribution was created with the assistance of an AI coding assistant (Pi / Anthropic / Gemini). All generated code has been manually audited, reviewed, compiled, and verified by the author in accordance with the Apache Software Foundation Generative Tooling Guidance.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub) via issues" <is...@iceberg.apache.org>.
Sruhvx-jpg commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4431052200
##########
crates/storage/common/tests/credential_suite.rs:
##########
@@ -0,0 +1,125 @@
+// 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, 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())
+ }
+}
+
+#[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();
+
+ let result = file_io_with_custom_creds.exists("s3://bucket1/any").await;
+ assert!(result.is_err());
Review Comment:
We originally had `.contains("failed to load signing credential")`, but @laskoviymishka flagged that matching internal error strings is fragile:
> "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 on asserting specific error kinds:
> "`Unexpected` is OpenDAL's current mapping, not a `FileIO` contract [...] I'd assert just `is_err()` here (same for the double-close at `:536` and the credential case in `credential_suite.rs:126`), and if we want a specific kind, spec it in the trait docs first."
Also, yes, we are making the `backends/` module refactor as you suggested!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub)" <gi...@apache.org>.
Sruhvx-jpg commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4158300280
##########
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 added a TODO, wdyt?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub) via issues" <is...@iceberg.apache.org>.
Sruhvx-jpg commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4431033486
##########
crates/storage/common/tests/common/mod.rs:
##########
@@ -0,0 +1,193 @@
+// 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 iceberg::io::{
+ 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};
+#[allow(unused_imports)]
+pub use iceberg_test_utils::storage::{
+ StorageHarness, handle_unreachable_endpoint, is_endpoint_reachable, roundtrip_file_io,
+ unique_path, wait_until_ready,
+};
+use iceberg_test_utils::{get_gcs_endpoint, get_object_store_endpoint, set_up};
+use tempfile::TempDir;
+
+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)
+}
+
+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())
+ }
+}
+
+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> {
Review Comment:
ummm, what could be its benefits tho?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub)" <gi...@apache.org>.
Sruhvx-jpg commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4152934498
##########
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:
This is intended as a single central test matrix for all storage implementations, following the precedent from #2131 for catalogs. OpenDAL is currently the only merged storage crate under crates/storage/ in main, but the TODO in StorageKind is already in place and I will personally follow up to wire up the object_store backend as soon as #3165 merges. Perhaps I should have stacked this PR directly on top of #3165 instead, wdyt?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub)" <gi...@apache.org>.
Sruhvx-jpg commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4153146001
##########
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:
Great idea!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub)" <gi...@apache.org>.
Sruhvx-jpg commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4152934498
##########
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:
This is intended as a single central test matrix for all storage implementations, following the precedent from #2131 for catalogs. OpenDAL is currently the only merged backend in main, but the TODO in StorageKind is already in place to plug in the object_store backend from #3165 as soon as it lands.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub)" <gi...@apache.org>.
Sruhvx-jpg commented on PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#issuecomment-6478745872
@laskoviymishka Hey laskov, I had asked my clanker to performa research to identify more test we could that we incorporate on top of these, It has identified them, alot of em actually!
So i was thinking that maybe I can address them as a stacked pr which would be ideal, but pretty sure I can't as for that I would have to atleast a committer with write accesss. I do see that prolly not possible. So should I just submit a individual new PR, what do u say?
I will review the research and stage the changes, once u tell me what to do, I will execute ur orders :smile:
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub)" <gi...@apache.org>.
Sruhvx-jpg commented on PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#issuecomment-5927613554
Pushed a commit addressing the review feedback.
For the items raised:
- split the unreachable skip path from setup/readiness failures (with readiness timeout and GCS bucket creation failure now failing hard with backend context), and gated skips behind `ICEBERG_REQUIRE_STORAGE=1`;
- restored the dropped serialization roundtrips for both GCS in `file_io_suite` and resolving storage in `resolving_suite`;
- loosened the out-of-bounds range-read assertion to accept empty reads alongside errors, adding a `TODO` comment to tighten it once the `FileRead` spec is standardized;
- left a reply on the inline thread regarding the shared suite architecture and #3165 stacking — there is an explicit `TODO` in place to wire up the `object_store` S3 backend once #3165 merges, which I will follow up and implement myself.
The rest — switching `.unwrap()` to `?` in the helpers, `TempDir` and delete-on-exit for hermetic resolving paths, moving probe tests to a dedicated `tests/endpoint_probe.rs` without the DNS query, asserting `ErrorKind::Unexpected` (as noted inline), dropping the no-op builder tests from `credential_suite`, and the `Cargo.toml` dependency cleanup — are all resolved.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub) via issues" <is...@iceberg.apache.org>.
Sruhvx-jpg commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4431032197
##########
crates/storage/common/tests/credential_suite.rs:
##########
@@ -0,0 +1,125 @@
+// 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, 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())
+ }
+}
+
+#[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();
+
+ let result = file_io_with_custom_creds.exists("s3://bucket1/any").await;
+ assert!(result.is_err());
Review Comment:
yeahh!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub)" <gi...@apache.org>.
Sruhvx-jpg commented on PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#issuecomment-6009787269
I understand ur concern, let's do one thing till s3 is not wired I will expose the loaders too 🙂
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub)" <gi...@apache.org>.
Sruhvx-jpg commented on PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#issuecomment-6517439863
Pushed a commit addressing the review feedback:
- **Moved shared test utilities to `crates/test_utils/src/storage/`**: Relocated `StorageHarness` and `endpoint_probe` into `iceberg_test_utils::storage`, keeping them generic over `&FileIO` without backend dependencies (@CTTY).
- **Inlined contract test assertions**: Removed `src/file_io.rs` and the dangling `run_all_file_io_contract_tests` runner, embedding test assertions directly into the `#[rstest]` cases in `tests/file_io_suite.rs` (@CTTY).
- **Endpoint probe proxy resilience**: Added `.no_proxy()` to the probe `reqwest::Client::builder()` to prevent proxy false-positives on localhost endpoints (@laskoviymishka).
- **Cargo.toml hygiene**: Pointed `reqsign-core` to `{ workspace = true, optional = true }` in `crates/storage/opendal/Cargo.toml`, and scoped `tokio` test features (`io-util`, `macros`, `net`) to `[dev-dependencies]` in `crates/storage/common/Cargo.toml` (@laskoviymishka).
- **Error propagation**: Swapped `TempDir::new().unwrap()` calls for `?` in `tests/resolving_suite.rs` (@laskoviymishka).
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4164784806
##########
crates/storage/common/tests/common/mod.rs:
##########
@@ -0,0 +1,276 @@
+// 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: Option<Box<TempDir>>,
+}
+
+fn handle_unreachable_endpoint(kind: &'static str, endpoint: &str) -> Option<StorageHarness> {
+ if std::env::var("ICEBERG_REQUIRE_STORAGE")
+ .map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
+ .unwrap_or(false)
+ {
+ panic!(
+ "storage backed '{kind}' is required by ICEBERG_REQUIRE_STORAGE, but endpoint '{endpoint}' is unreachable"
+ );
+ }
+ eprintln!("Skipping {kind} storage test: {endpoint} not reachable");
+ None
+}
+
+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())
+ }
+}
+
+const DEFAULT_PROBE_TIMEOUT_MS: u64 = 1000;
+
+fn get_probe_timeout() -> Duration {
+ let ms = std::env::var("ICEBERG_PROBE_TIMEOUT_MS")
+ .ok()
+ .and_then(|v| v.parse().ok())
+ .unwrap_or(DEFAULT_PROBE_TIMEOUT_MS);
+ Duration::from_millis(ms)
+}
+
+/// Fast probe to check if an endpoint service is listening before entering retry loops.
+///
+/// Note: Any HTTP response from `.send().await.is_ok()` (including 4xx/5xx) is treated
+/// as reachable, as it proves the underlying server is up, listening on the port,
+/// and actively responding to HTTP requests.
+pub async fn is_endpoint_reachable(endpoint: &str) -> bool {
+ let Ok(client) = reqwest::Client::builder()
+ .timeout(get_probe_timeout())
+ .build()
+ else {
+ return false;
+ };
+ client.get(endpoint).send().await.is_ok()
+}
+
+async fn wait_until_ready(file_io: &FileIO, check_path: &str, kind: &'static str, endpoint: &str) {
+ let mut retries = 0;
+ while retries < 15 {
+ if file_io.exists(check_path).await.unwrap_or(false) {
+ return;
+ }
+ sleep(Duration::from_millis(500)).await;
+ retries += 1;
+ }
+
+ panic!(
+ "Storage backend '{kind}' was reachable at '{endpoint}', but failed readiness check on '{check_path}' after 15 retries"
+ );
+}
+
+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 {
+ return handle_unreachable_endpoint("opendal_s3", &object_store_endpoint);
+ }
+
+ let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::S3 {
+ customized_credential_load: None,
+ }))
+ .with_props(vec![
+ (S3_ENDPOINT, object_store_endpoint.clone()),
+ (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();
+
+ wait_until_ready(
+ &file_io,
+ "s3://bucket1/",
+ "opendal_s3",
+ &object_store_endpoint,
+ )
+ .await;
+
+ Some(StorageHarness {
+ file_io,
+ label: "opendal_s3",
+ base_path: "s3://bucket1/".to_string(),
+ _tempdirs: None,
+ })
+}
+
+async fn load_opendal_gcs() -> Option<StorageHarness> {
+ let gcs_endpoint = get_gcs_endpoint();
+
+ if !is_endpoint_reachable(&gcs_endpoint).await {
+ return handle_unreachable_endpoint("opendal_gcs", &gcs_endpoint);
+ }
+
+ 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");
+ let response = client
+ .post(&endpoint)
+ .json(&bucket_data)
+ .send()
+ .await
+ .unwrap_or_else(|e| {
+ panic!("Failed to send GCS Bucket creation request to '{endpoint}': {e}")
+ });
+
+ let status = response.status();
+ if !status.is_success() && status != reqwest::StatusCode::CONFLICT {
+ panic!("failed to create GCS test bucket '{FAKE_GCS_BUCKET}': HTTP status {status}");
+ }
+ let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::Gcs))
+ .with_props(vec![
+ (GCS_SERVICE_HOST, gcs_endpoint.clone()),
+ (GCS_NO_AUTH, "true".to_string()),
+ ])
+ .build();
+
+ let base_path = format!("gs://{FAKE_GCS_BUCKET}/");
+
+ wait_until_ready(&file_io, &base_path, "opendal_gcs", &gcs_endpoint).await;
+
+ Some(StorageHarness {
+ file_io,
+ label: "opendal_gcs",
+ base_path,
+ _tempdirs: None,
+ })
+}
+
+async fn load_opendal_fs() -> Option<StorageHarness> {
+ let temp_dir = TempDir::new().ok()?;
Review Comment:
`.ok()?` here turns a tempdir failure into `None`, which every caller treats as "skip" and returns `Ok(())` — but fs has no endpoint, so `ICEBERG_REQUIRE_STORAGE` can't catch it. A full or read-only `$TMPDIR` would green the whole fs matrix silently. I'd make the local backends infallible — `TempDir::new().expect("create temp dir for fs harness")` — so a setup failure is loud.
##########
crates/storage/common/tests/file_io_suite.rs:
##########
@@ -0,0 +1,566 @@
+// 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::ErrorKind;
+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?);
+ assert!(harness.file_io.exists(&harness.base_path).await?);
+ 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?);
+
+ let output_file = harness.file_io.new_output(&path)?;
+ output_file.write("123".into()).await?;
+ assert!(harness.file_io.exists(&path).await?);
+
+ 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)?;
+ output_file.write("test_input".into()).await?;
+
+ let input_file = harness.file_io.new_input(&path)?;
+ let buffer = input_file.read().await?;
+ 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)?
+ .write("delete_me".into())
+ .await?;
+ assert!(harness.file_io.exists(&path).await?);
+
+ harness.file_io.delete(&path).await?;
+ assert!(!harness.file_io.exists(&path).await?);
+ 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?;
+ 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)?
+ .write("delete-me".into())
+ .await?;
+ assert!(harness.file_io.exists(path).await?);
+ }
+ let stream = futures::stream::iter(paths.clone()).boxed();
+ harness.file_io.delete_stream(stream).await?;
+ for path in &paths {
+ assert!(!harness.file_io.exists(path).await?);
+ }
+ Ok(())
+}
+
+async fn run_delete_stream_empty(harness: StorageHarness) -> iceberg::Result<()> {
+ let stream = futures::stream::empty().boxed();
+ harness.file_io.delete_stream(stream).await?;
+ 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)?
+ .write(content.into())
+ .await?;
+ let input_file = harness.file_io.new_input(&path)?;
+ let metadata = input_file.metadata().await?;
+ 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)?
+ .write(Bytes::from_static(content))
+ .await?;
+ let input_file = harness.file_io.new_input(&path)?;
+ let reader = input_file.reader().await?;
+ let range_data = reader.read(4..10).await?;
+ 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)?;
+ output_file.write(Bytes::new()).await?;
+
+ assert!(harness.file_io.exists(&path).await?);
+
+ let input_file = harness.file_io.new_input(&path)?;
+ let metadata = input_file.metadata().await?;
+ assert_eq!(metadata.size, 0);
+
+ let data = input_file.read().await?;
+ 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)?
+ .write("data".into())
+ .await?;
+ assert!(harness.file_io.exists(path).await?);
+ }
+ for path in &nonexistent_paths {
+ let _ = harness.file_io.delete(path).await;
+ assert!(!harness.file_io.exists(path).await?);
+ }
+
+ 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?;
+
+ for path in &existing_paths {
+ assert!(!harness.file_io.exists(path).await?);
+ }
+ 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)?;
+ output.write(payload.clone().into()).await?;
+
+ let input = file_io.new_input(&path)?;
+ let data = input.read().await?;
+ assert_eq!(data, payload.as_bytes());
+
+ let _ = file_io.delete(&path).await;
+ Ok::<(), iceberg::Error>(())
+ }));
+ }
+
+ 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)?
+ .write("data".into())
+ .await?;
+ assert!(harness.file_io.exists(path).await?);
+ }
+ harness.file_io.delete_prefix(&prefix).await?;
+ for path in &paths {
+ assert!(!harness.file_io.exists(path).await?);
+ }
+ 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?;
+ 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)?;
+ let Err(err) = input_file.metadata().await else {
+ panic!("expected metadata on nonexistent file to return error");
+ };
+ assert_eq!(err.kind(), ErrorKind::Unexpected);
Review Comment:
`Unexpected` is OpenDAL's current mapping, not a `FileIO` contract — for metadata on a missing file `NotFound` is the semantically right kind, and the trait docs don't promise either. The object_store backend from #3165 returning `NotFound` would fail this, and a later fix to the OpenDAL mapping would too — both false failures. I'd assert just `is_err()` here (same for the double-close at `:536` and the credential case in `credential_suite.rs:126`), and if we want a specific kind, spec it in the trait docs first.
##########
crates/storage/common/tests/common/mod.rs:
##########
@@ -0,0 +1,276 @@
+// 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: Option<Box<TempDir>>,
+}
+
+fn handle_unreachable_endpoint(kind: &'static str, endpoint: &str) -> Option<StorageHarness> {
+ if std::env::var("ICEBERG_REQUIRE_STORAGE")
+ .map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
+ .unwrap_or(false)
+ {
+ panic!(
+ "storage backed '{kind}' is required by ICEBERG_REQUIRE_STORAGE, but endpoint '{endpoint}' is unreachable"
+ );
+ }
+ eprintln!("Skipping {kind} storage test: {endpoint} not reachable");
Review Comment:
The reachable-but-broken cases panic now, which handles the main half of last round's concern. But the skip is still the default and nothing sets `ICEBERG_REQUIRE_STORAGE` — it's not in the workflows or the Makefile. So if `docker-up` fails or the endpoint moves, every S3/GCS/resolving case lands here, returns `Ok(())`, and nextest reports it green: that's the "must pass → may silently vanish" drop from round 1, just narrowed to the unreachable path.
I'd set `ICEBERG_REQUIRE_STORAGE=1` in the CI step right after `docker-up`, or invert the default so unreachable fails unless something like `ICEBERG_SKIP_STORAGE=1` is set — inverting is safer for a fork or a local run that forgets the flag.
While we're here, the probe is a single 1s GET with the retry loop only running after it succeeds, so a slow container reads as absent (a false skip by default, a false failure under the gate); a short probe retry would smooth that.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub)" <gi...@apache.org>.
Sruhvx-jpg commented on PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#issuecomment-5950941509
>binary, so https://github.com/apache/iceberg-rust/pull/3165 can't actually reuse them. I wouldn't redesign it here, but we should probably pick one direction: expose the shared suite from src/, or rename the crate and drop the shared framing.
@laskoviymishka ok exposing shared suite - sounds the right thing to do for "now", wdyt?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "CTTY (via GitHub) via issues" <is...@iceberg.apache.org>.
CTTY commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4429798084
##########
crates/storage/common/tests/file_io_suite.rs:
##########
@@ -0,0 +1,494 @@
+// 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::{StorageKind, load_storage};
+use futures::StreamExt;
+use iceberg_test_utils::storage::roundtrip_file_io;
+use rstest::rstest;
+
+#[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(());
+ };
Review Comment:
what if `load_storage` failed? would it just skip the tests and report green?
##########
crates/storage/common/tests/credential_suite.rs:
##########
@@ -0,0 +1,125 @@
+// 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, 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())
+ }
+}
+
+#[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();
+
+ let result = file_io_with_custom_creds.exists("s3://bucket1/any").await;
+ assert!(result.is_err());
Review Comment:
I think we should validate the error message here
##########
crates/storage/common/tests/common/mod.rs:
##########
@@ -0,0 +1,193 @@
+// 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 iceberg::io::{
+ 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};
+#[allow(unused_imports)]
+pub use iceberg_test_utils::storage::{
+ StorageHarness, handle_unreachable_endpoint, is_endpoint_reachable, roundtrip_file_io,
+ unique_path, wait_until_ready,
+};
+use iceberg_test_utils::{get_gcs_endpoint, get_object_store_endpoint, set_up};
+use tempfile::TempDir;
+
+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)
+}
+
+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())
+ }
+}
+
+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> {
Review Comment:
We should have a new module `backends` and move these helpers to `backends/opendal.rs`
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub) via issues" <is...@iceberg.apache.org>.
Sruhvx-jpg commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4431045664
##########
crates/storage/common/tests/common/mod.rs:
##########
@@ -0,0 +1,193 @@
+// 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 iceberg::io::{
+ 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};
+#[allow(unused_imports)]
+pub use iceberg_test_utils::storage::{
+ StorageHarness, handle_unreachable_endpoint, is_endpoint_reachable, roundtrip_file_io,
+ unique_path, wait_until_ready,
+};
+use iceberg_test_utils::{get_gcs_endpoint, get_object_store_endpoint, set_up};
+use tempfile::TempDir;
+
+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)
+}
+
+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())
+ }
+}
+
+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> {
Review Comment:
ohhhh, yeah taking into account the fact that more loader will be made then yes. Thanks for the suggestion!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub)" <gi...@apache.org>.
Sruhvx-jpg commented on PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#issuecomment-6014131348
Pushed a commit addressing the remaining review hygiene items. This drops the unused `anyhow` dev-dependency and explicitly declares the required `tokio` features (`io-util`, `macros`, `net`, `rt-multi-thread`, `time`) in `crates/storage/common/Cargo.toml`. It also removes the duplicated `roundtrip_file_io` helper in `resolving_suite.rs` in favor of the shared import with `?` error propagation throughout, and scopes `ICEBERG_REQUIRE_STORAGE=1` directly to the `test: docker-up` recipe in the `Makefile`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4204862044
##########
crates/storage/common/Cargo.toml:
##########
@@ -0,0 +1,53 @@
+# 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]
+bytes = { workspace = true }
+futures = { workspace = true }
+iceberg = { workspace = true }
+iceberg_test_utils = { path = "../../test_utils", features = ["tests"] }
+reqwest = { workspace = true }
+tempfile = { workspace = true }
+tokio = { workspace = true, features = [
Review Comment:
`src/` only needs `rt` and `time` here (`tokio::spawn` plus `sleep`); `io-util`, `macros`, and `net` are only pulled in by the test binaries, so under `[dependencies]` they get turned on for every consumer of the crate. I'd move those three down:
```toml
[dependencies]
tokio = { workspace = true, features = ["rt", "time"] }
[dev-dependencies]
tokio = { workspace = true, features = ["io-util", "macros", "net", "rt-multi-thread"] }
```
that's the last of the two Cargo.toml nits I left open.
##########
Cargo.toml:
##########
@@ -115,6 +115,7 @@ pyo3 = "0.29"
quote = "1"
rand = "0.9.3"
regex = "1.11.3"
+reqsign-core = "3.0.0"
Review Comment:
now that `reqsign-core` is a workspace dep, the opendal crate still pins it as a literal `"3.0.0"` (`crates/storage/opendal/Cargo.toml:58`) — point that at `reqsign-core = { workspace = true, optional = true }` so the version lives in one place.
##########
crates/storage/common/src/endpoint_probe.rs:
##########
@@ -0,0 +1,97 @@
+// 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.
+
+//! Fast endpoint probe and readiness utilities for integration test emulators.
+
+use std::time::Duration;
+
+use iceberg::io::FileIO;
+use tokio::time::sleep;
+
+use crate::harness::StorageHarness;
+
+const DEFAULT_PROBE_TIMEOUT_MS: u64 = 1000;
+const DEFAULT_PROBE_RETRIES: usize = 3;
+const PROBE_RETRY_INTERVAL_MS: u64 = 200;
+
+fn get_probe_timeout() -> Duration {
+ let ms = std::env::var("ICEBERG_PROBE_TIMEOUT_MS")
+ .ok()
+ .and_then(|v| v.parse().ok())
+ .unwrap_or(DEFAULT_PROBE_TIMEOUT_MS);
+ Duration::from_millis(ms)
+}
+
+/// Fast probe to check if an endpoint service is listening before entering retry loops.
+///
+/// Note: Any HTTP response from `.send().await.is_ok()` (including 4xx/5xx) is treated
+/// as reachable, as it proves the underlying server is up, listening on the port,
+/// and actively responding to HTTP requests.
+pub async fn is_endpoint_reachable(endpoint: &str) -> bool {
+ let Ok(client) = reqwest::Client::builder()
Review Comment:
`reqwest::Client::builder()` honors `HTTP_PROXY`/`ALL_PROXY`, so on a runner with a proxy a GET to `localhost:9000` can come back as a proxy 502 — and since any HTTP response counts as reachable, the probe returns `true` for a down endpoint and the `ICEBERG_REQUIRE_STORAGE` hard-fail never fires (the failure just moves to `wait_until_ready` or a test body). one `.no_proxy()` on the builder closes it.
##########
crates/storage/common/tests/resolving_suite.rs:
##########
@@ -0,0 +1,344 @@
+// 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_common::roundtrip_file_io;
+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;
+use tempfile::TempDir;
+
+#[rstest]
+#[case::opendal_resolving(StorageKind::OpenDalResolving)]
+#[tokio::test]
+async fn test_mixed_scheme_write_and_read(#[case] kind: StorageKind) -> iceberg::Result<()> {
+ let Some(harness) = load_storage(kind).await else {
+ return Ok(());
+ };
+
+ let s3_path = unique_path(&harness, "test_mixed_scheme_write_and_read");
+ let temp_dir = TempDir::new().unwrap();
Review Comment:
the four `TempDir::new().unwrap()`s (45, 98, 141, 199) can go to `?` via `map_err` the way we did in `file_io.rs`, and `test_resolving_serialization_roundtrip` (325) is now a byte-for-byte copy of `run_file_io_serialization_roundtrip` — just call `run_file_io_serialization_roundtrip(&harness).await`. non-blocking, same cleanup we noted last round.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub)" <gi...@apache.org>.
Sruhvx-jpg commented on PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#issuecomment-5966538974
changes coming thru!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub) via issues" <is...@iceberg.apache.org>.
Sruhvx-jpg commented on PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#issuecomment-6541091588
We originally had .contains("failed to load signing credential"), but @laskoviymishka flagged that matching internal error strings is fragile:
> 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 on asserting specific error kinds:
> Unexpected is OpenDAL's current mapping, not a FileIO contract [...] I'd assert just is_err() here (same for the double-close at :536 and the credential case in credential_suite.rs:126), and if we want a specific kind, spec it in the trait docs first.
Also, yes, am making the backends/ module refactor as you suggested!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub) via issues" <is...@iceberg.apache.org>.
Sruhvx-jpg commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4431033486
##########
crates/storage/common/tests/common/mod.rs:
##########
@@ -0,0 +1,193 @@
+// 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 iceberg::io::{
+ 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};
+#[allow(unused_imports)]
+pub use iceberg_test_utils::storage::{
+ StorageHarness, handle_unreachable_endpoint, is_endpoint_reachable, roundtrip_file_io,
+ unique_path, wait_until_ready,
+};
+use iceberg_test_utils::{get_gcs_endpoint, get_object_store_endpoint, set_up};
+use tempfile::TempDir;
+
+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)
+}
+
+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())
+ }
+}
+
+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> {
Review Comment:
ummm, what could be its benefits tho?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4187831265
##########
crates/storage/common/tests/resolving_suite.rs:
##########
@@ -0,0 +1,418 @@
+// 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::{FileIO, 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;
+use tempfile::TempDir;
+
+fn roundtrip_file_io(file_io: &FileIO) -> FileIO {
Review Comment:
This re-implements the `roundtrip_file_io` we already export from `src/file_io.rs` — I'd import the shared one and delete the local copy. Same story for the `.unwrap()` chain through these tests: they already return `iceberg::Result<()>`, so `?` is the cleanup we did in `file_io.rs` and it's worth finishing here too.
##########
crates/storage/common/Cargo.toml:
##########
@@ -0,0 +1,48 @@
+# 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]
+bytes = { workspace = true }
+futures = { workspace = true }
+iceberg = { workspace = true }
+iceberg_test_utils = { path = "../../test_utils", features = ["tests"] }
+reqwest = { workspace = true }
+tempfile = { workspace = true }
+tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
+
+[dev-dependencies]
+anyhow = { workspace = true }
Review Comment:
Nothing in the crate references `anyhow`, and this same PR drops it from the opendal crate — cargo-machete will flag it and turn CI red, so I'd pull it from the dev-deps.
##########
crates/storage/common/Cargo.toml:
##########
@@ -0,0 +1,48 @@
+# 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]
+bytes = { workspace = true }
+futures = { workspace = true }
+iceberg = { workspace = true }
+iceberg_test_utils = { path = "../../test_utils", features = ["tests"] }
+reqwest = { workspace = true }
+tempfile = { workspace = true }
+tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
Review Comment:
I'd spell out the tokio features this crate actually uses — `time` for the probe's `sleep`, and `net` + `io-util` for the `TcpListener`/`AsyncWrite` in the probe tests. Right now it only builds because workspace feature unification pulls those in; `cargo check -p iceberg-storage-common` on its own would fail.
##########
Makefile:
##########
@@ -17,6 +17,8 @@
.EXPORT_ALL_VARIABLES:
+ICEBERG_REQUIRE_STORAGE ?= 1
Review Comment:
Not blocking, but `?= 1` with `.EXPORT_ALL_VARIABLES` exports to every target, so `make build && cargo test` or a bare `make nextest` hard-fail for anyone without docker up — only `make test` runs docker-up. I'd set it on the test recipe instead (`ICEBERG_REQUIRE_STORAGE=1 $(MAKE) nextest`) so the default flow still works.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub)" <gi...@apache.org>.
Sruhvx-jpg commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4153164204
##########
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:
Good catch on checking the error kind. Note that iceberg::ErrorKind currently does not have a general NotFound variant (only catalog-level TableNotFound/NamespaceNotFound exist), and from_opendal_error in
iceberg-storage-opendal maps all OpenDAL errors to ErrorKind::Unexpected. I've updated both metadata_nonexistent and streaming_write_double_close to assert ErrorKind::Unexpected instead of generic is_err().
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub)" <gi...@apache.org>.
Sruhvx-jpg commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4153164204
##########
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:
Good catch on checking the error kind. Note that iceberg::ErrorKind currently does not have a general NotFound variant (only catalog-level TableNotFound/NamespaceNotFound exist), and from_opendal_error in iceberg-storage-opendal maps all OpenDAL errors to ErrorKind::Unexpected. I've updated both metadata_nonexistent and streaming_write_double_close to assert ErrorKind::Unexpected instead of generic is_err().
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub)" <gi...@apache.org>.
Sruhvx-jpg commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4152934498
##########
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:
This is intended as a single central test matrix for all storage implementations, following the precedent from #2131 for catalogs. OpenDAL is currently the only merged storage crate under crates/storage/ in main, but the TODO in StorageKind is already in place and I will personally follow up to wire up the object_store backend as soon as #3165 merges. Perhaps I should have stacked this PR directly on top of #3165 instead?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4418190895
##########
crates/storage/common/src/file_io.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.
+
+//! Generic `FileIO` contract tests executable across all storage backends.
+
+use bytes::Bytes;
+use futures::StreamExt;
+use iceberg::io::FileIO;
+
+use crate::harness::StorageHarness;
+
+/// Helper to serialize and deserialize a `FileIO` instance.
+pub fn roundtrip_file_io(file_io: &FileIO) -> FileIO {
Review Comment:
I found this folder structure confusing. how about moving common test utils to crates/test_utils?
`crates/test_utils/src/storage/...` specifically
Also do we really need these helpers to be functions? Previously I was thinking of just putting them directly to file_io_suite.rs
##########
crates/storage/common/src/file_io.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.
+
+//! Generic `FileIO` contract tests executable across all storage backends.
+
+use bytes::Bytes;
+use futures::StreamExt;
+use iceberg::io::FileIO;
+
+use crate::harness::StorageHarness;
+
+/// Helper to serialize and deserialize a `FileIO` instance.
+pub fn roundtrip_file_io(file_io: &FileIO) -> FileIO {
+ let serialized = file_io
+ .serialize_all()
+ .expect("FileIO serialization must succeed");
+ FileIO::deserialize_all(&serialized).expect("FileIO deserialization must succeed")
+}
+
+/// Tests that `exists` returns true for base directory and false for nonexistent path.
+pub async fn run_file_io_exists(harness: &StorageHarness) -> iceberg::Result<()> {
+ let non_existent = harness.unique_path("non_existent_file_that_does_not_exist");
+ assert!(!harness.file_io.exists(&non_existent).await?);
+ assert!(harness.file_io.exists(&harness.base_path).await?);
+ Ok(())
+}
+
+/// Tests basic write and existence validation.
+pub async fn run_file_io_write(harness: &StorageHarness) -> iceberg::Result<()> {
+ let path = harness.unique_path("test_file_io_write");
+ let _ = harness.file_io.delete(&path).await;
+ assert!(!harness.file_io.exists(&path).await?);
+
+ let output_file = harness.file_io.new_output(&path)?;
+ output_file.write("123".into()).await?;
+ assert!(harness.file_io.exists(&path).await?);
+
+ let _ = harness.file_io.delete(&path).await;
+ Ok(())
+}
+
+/// Tests write followed by reading back contents.
+pub async fn run_file_io_read(harness: &StorageHarness) -> iceberg::Result<()> {
+ let path = harness.unique_path("test_file_io_read");
+ let _ = harness.file_io.delete(&path).await;
+
+ let output_file = harness.file_io.new_output(&path)?;
+ output_file.write("test_input".into()).await?;
+
+ let input_file = harness.file_io.new_input(&path)?;
+ let buffer = input_file.read().await?;
+ assert_eq!(buffer, "test_input".as_bytes());
+
+ let _ = harness.file_io.delete(&path).await;
+ Ok(())
+}
+
+/// Tests writing and deleting a file.
+pub async fn run_file_io_delete(harness: &StorageHarness) -> iceberg::Result<()> {
+ let path = harness.unique_path("test_file_io_delete");
+ let _ = harness.file_io.delete(&path).await;
+
+ harness
+ .file_io
+ .new_output(&path)?
+ .write("delete_me".into())
+ .await?;
+ assert!(harness.file_io.exists(&path).await?);
+
+ harness.file_io.delete(&path).await?;
+ assert!(!harness.file_io.exists(&path).await?);
+ Ok(())
+}
+
+/// Tests deleting a nonexistent file (must succeed without error).
+pub async fn run_file_io_delete_nonexistent(harness: &StorageHarness) -> iceberg::Result<()> {
+ let path = harness.unique_path("test_file_io_delete_nonexistent");
+ harness.file_io.delete(&path).await?;
+ Ok(())
+}
+
+/// Tests bulk batch deletion via stream.
+pub async fn run_file_io_delete_stream(harness: &StorageHarness) -> iceberg::Result<()> {
+ let base = harness.unique_path("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)?
+ .write("delete-me".into())
+ .await?;
+ assert!(harness.file_io.exists(path).await?);
+ }
+ let stream = futures::stream::iter(paths.clone()).boxed();
+ harness.file_io.delete_stream(stream).await?;
+ for path in &paths {
+ assert!(!harness.file_io.exists(path).await?);
+ }
+ Ok(())
+}
+
+/// Tests empty delete stream (must succeed).
+pub async fn run_file_io_delete_stream_empty(harness: &StorageHarness) -> iceberg::Result<()> {
+ let stream = futures::stream::empty().boxed();
+ harness.file_io.delete_stream(stream).await?;
+ Ok(())
+}
+
+/// Tests bulk delete stream containing mixed existing and nonexistent paths.
+pub async fn run_file_io_delete_stream_mixed(harness: &StorageHarness) -> iceberg::Result<()> {
+ let base = harness.unique_path("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)?
+ .write("data".into())
+ .await?;
+ assert!(harness.file_io.exists(path).await?);
+ }
+ for path in &nonexistent_paths {
+ let _ = harness.file_io.delete(path).await;
+ assert!(!harness.file_io.exists(path).await?);
+ }
+
+ 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?;
+
+ for path in &existing_paths {
+ assert!(!harness.file_io.exists(path).await?);
+ }
+ Ok(())
+}
+
+/// Tests prefix deletion.
+pub async fn run_file_io_delete_prefix(harness: &StorageHarness) -> iceberg::Result<()> {
+ let prefix = harness.unique_path("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)?
+ .write("data".into())
+ .await?;
+ assert!(harness.file_io.exists(path).await?);
+ }
+ harness.file_io.delete_prefix(&prefix).await?;
+ for path in &paths {
+ assert!(!harness.file_io.exists(path).await?);
+ }
+ Ok(())
+}
+
+/// Tests deleting a nonexistent prefix (must succeed).
+pub async fn run_file_io_delete_prefix_nonexistent(
+ harness: &StorageHarness,
+) -> iceberg::Result<()> {
+ let prefix = harness.unique_path("test_file_io_delete_prefix_nonexistent");
+ harness.file_io.delete_prefix(&prefix).await?;
+ Ok(())
+}
+
+/// Tests file metadata inspection.
+pub async fn run_file_io_metadata(harness: &StorageHarness) -> iceberg::Result<()> {
+ let path = harness.unique_path("test_file_io_metadata");
+ let _ = harness.file_io.delete(&path).await;
+ let content = "metadata_test_content";
+ harness
+ .file_io
+ .new_output(&path)?
+ .write(content.into())
+ .await?;
+ let input_file = harness.file_io.new_input(&path)?;
+ let metadata = input_file.metadata().await?;
+ assert_eq!(metadata.size, content.len() as u64);
+ let _ = harness.file_io.delete(&path).await;
+ Ok(())
+}
+
+/// Tests metadata on nonexistent file returns an error.
+pub async fn run_file_io_metadata_nonexistent(harness: &StorageHarness) -> iceberg::Result<()> {
+ let path = harness.unique_path("test_file_io_metadata_nonexistent");
+ let input_file = harness.file_io.new_input(&path)?;
+ assert!(input_file.metadata().await.is_err());
+ Ok(())
+}
+
+/// Tests byte range reading.
+pub async fn run_file_io_range_read(harness: &StorageHarness) -> iceberg::Result<()> {
+ let path = harness.unique_path("test_file_io_range_read");
+ let _ = harness.file_io.delete(&path).await;
+ let content = b"0123456789abcdef";
+ harness
+ .file_io
+ .new_output(&path)?
+ .write(Bytes::from_static(content))
+ .await?;
+ let input_file = harness.file_io.new_input(&path)?;
+ let reader = input_file.reader().await?;
+ let range_data = reader.read(4..10).await?;
+ assert_eq!(range_data.as_ref(), &content[4..10]);
+ let _ = harness.file_io.delete(&path).await;
+ Ok(())
+}
+
+/// Tests out-of-bounds byte range read behavior.
+pub async fn run_file_io_range_read_out_of_bounds(harness: &StorageHarness) -> iceberg::Result<()> {
+ let path = harness.unique_path("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)?
+ .write(Bytes::from_static(content))
+ .await?;
+
+ let input_file = harness.file_io.new_input(&path)?;
+ let reader = input_file.reader().await?;
+ let result = reader.read(100..200).await;
+
+ // TODO: Standardize out-of-bounds range-read behavior in `FileRead` trait docs.
+ // Cloud backends (S3/GCS via HTTP 416) or object_store may surface an error,
+ // while POSIX/memory backends may return an empty buffer (EOF semantics).
+ // Accept either an error or 0 bytes until the contract is formally specified.
+ if let Ok(bytes) = result {
+ assert!(
+ bytes.is_empty(),
+ "expected empty read for out-of-bounds range, got {} bytes",
+ bytes.len()
+ );
+ }
+
+ let _ = harness.file_io.delete(&path).await;
+ Ok(())
+}
+
+/// Tests 0-byte file write, existence, metadata, and read.
+pub async fn run_file_io_zero_byte_file(harness: &StorageHarness) -> iceberg::Result<()> {
+ let path = harness.unique_path("test_file_io_zero_byte_file");
+ let _ = harness.file_io.delete(&path).await;
+
+ let output_file = harness.file_io.new_output(&path)?;
+ output_file.write(Bytes::new()).await?;
+
+ assert!(harness.file_io.exists(&path).await?);
+
+ let input_file = harness.file_io.new_input(&path)?;
+ let metadata = input_file.metadata().await?;
+ assert_eq!(metadata.size, 0);
+
+ let data = input_file.read().await?;
+ assert_eq!(data, Bytes::new());
+
+ let _ = harness.file_io.delete(&path).await;
+ Ok(())
+}
+
+/// Tests concurrent writes from multiple spawned tokio tasks.
+pub async fn run_file_io_concurrent_writes(harness: &StorageHarness) -> iceberg::Result<()> {
+ let base = harness.unique_path("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)?;
+ output.write(payload.clone().into()).await?;
+
+ let input = file_io.new_input(&path)?;
+ let data = input.read().await?;
+ assert_eq!(data, payload.as_bytes());
+
+ let _ = file_io.delete(&path).await;
+ Ok::<(), iceberg::Error>(())
+ }));
+ }
+
+ for handle in handles {
+ handle.await.expect("task join failed")?;
+ }
+
+ Ok(())
+}
+
+/// Tests streaming multipart writer.
+pub async fn run_file_io_streaming_write(harness: &StorageHarness) -> iceberg::Result<()> {
+ let path = harness.unique_path("test_file_io_streaming_write");
+ let output_file = harness.file_io.new_output(&path)?;
+ let mut writer = output_file.writer().await?;
+ writer.write(Bytes::from("streaming_content")).await?;
+ writer.close().await?;
+ let input_file = harness.file_io.new_input(&path)?;
+ let buffer = input_file.read().await?;
+ assert_eq!(buffer, Bytes::from("streaming_content"));
+ let _ = harness.file_io.delete(&path).await;
+ Ok(())
+}
+
+/// Tests double-closing a writer returns an error.
+pub async fn run_file_io_streaming_write_double_close(
+ harness: &StorageHarness,
+) -> iceberg::Result<()> {
+ let path = harness.unique_path("test_file_io_streaming_write_double_close");
+ let output_file = harness.file_io.new_output(&path)?;
+ let mut writer = output_file.writer().await?;
+ writer.write(Bytes::from("data")).await?;
+ writer.close().await?;
+ assert!(writer.close().await.is_err());
+ let _ = harness.file_io.delete(&path).await;
+ Ok(())
+}
+
+/// Tests serialization and deserialization roundtrip.
+pub async fn run_file_io_serialization_roundtrip(harness: &StorageHarness) -> iceberg::Result<()> {
+ let file_io = roundtrip_file_io(&harness.file_io);
+ let path = harness.unique_path("test_file_io_serialization_roundtrip");
+
+ let _ = file_io.delete(&path).await;
+ file_io
+ .new_output(&path)?
+ .write(Bytes::from_static(b"roundtrip"))
+ .await?;
+ assert_eq!(
+ file_io.new_input(&path)?.read().await?,
+ Bytes::from_static(b"roundtrip")
+ );
+ file_io.delete(&path).await?;
+ assert!(!file_io.exists(&path).await?);
+ Ok(())
+}
+
+/// Run all basic `FileIO` contract tests sequentially against a storage harness.
+pub async fn run_all_file_io_contract_tests(harness: &StorageHarness) -> iceberg::Result<()> {
Review Comment:
I don't see where we actually run these tests
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub) via issues" <is...@iceberg.apache.org>.
Sruhvx-jpg commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4431052200
##########
crates/storage/common/tests/credential_suite.rs:
##########
@@ -0,0 +1,125 @@
+// 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, 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())
+ }
+}
+
+#[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();
+
+ let result = file_io_with_custom_creds.exists("s3://bucket1/any").await;
+ assert!(result.is_err());
Review Comment:
We originally had `.contains("failed to load signing credential")`, but @laskoviymishka flagged that matching internal error strings is fragile:
> "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 on asserting specific error kinds:
> "`Unexpected` is OpenDAL's current mapping, not a `FileIO` contract [...] I'd assert just `is_err()` here (same for the double-close at `:536` and the credential case in `credential_suite.rs:126`), and if we want a specific kind, spec it in the trait docs first."
Also, yes, we are making the `backends/` module refactor as you suggested!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(storage)(test): Add shared integration test suite for iceberg-storage (Supersedes #2253) [iceberg-rust]
Posted by "Sruhvx-jpg (via GitHub) via issues" <is...@iceberg.apache.org>.
Sruhvx-jpg commented on code in PR #3288:
URL: https://github.com/apache/iceberg-rust/pull/3288#discussion_r4431032197
##########
crates/storage/common/tests/credential_suite.rs:
##########
@@ -0,0 +1,125 @@
+// 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, 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())
+ }
+}
+
+#[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();
+
+ let result = file_io_with_custom_creds.exists("s3://bucket1/any").await;
+ assert!(result.is_err());
Review Comment:
yeahh!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org