diff --git a/Cargo.lock b/Cargo.lock index b6db2308c10..d172886ccc2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1457,6 +1457,30 @@ dependencies = [ "url", ] +[[package]] +name = "aws-sdk-s3tables" +version = "1.54.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e0ec266873694efc365debded01f44e27a0de3946a3ac15d24c489759e5ddf8" +dependencies = [ + "aws-credential-types", + "aws-runtime", + "aws-smithy-async", + "aws-smithy-http", + "aws-smithy-json", + "aws-smithy-observability", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-types", + "aws-types", + "bytes", + "fastrand", + "http 0.2.12", + "http 1.3.1", + "regex-lite", + "tracing", +] + [[package]] name = "aws-sdk-sso" version = "1.97.0" @@ -1882,7 +1906,7 @@ dependencies = [ "bitflags 2.10.0", "cexpr", "clang-sys", - "itertools 0.10.5", + "itertools 0.11.0", "proc-macro2", "quote", "regex", @@ -1900,7 +1924,7 @@ dependencies = [ "bitflags 2.10.0", "cexpr", "clang-sys", - "itertools 0.10.5", + "itertools 0.11.0", "proc-macro2", "quote", "regex", @@ -1918,7 +1942,7 @@ dependencies = [ "bitflags 2.10.0", "cexpr", "clang-sys", - "itertools 0.10.5", + "itertools 0.11.0", "log", "prettyplease", "proc-macro2", @@ -2600,7 +2624,7 @@ version = "3.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fde0e0ec90c9dfb3b4b1a0891a7dcd0e2bffde2f7efed5fe7c9bb00e5bfb915e" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.59.0", ] [[package]] @@ -5342,6 +5366,7 @@ dependencies = [ "iceberg", "iceberg-catalog-glue", "iceberg-catalog-rest", + "iceberg-catalog-s3tables", "iceberg-datafusion", "iceberg-storage-opendal", "log", @@ -6542,7 +6567,7 @@ dependencies = [ "js-sys", "log", "wasm-bindgen", - "windows-core 0.61.2", + "windows-core 0.57.0", ] [[package]] @@ -6644,6 +6669,19 @@ dependencies = [ "uuid", ] +[[package]] +name = "iceberg-catalog-s3tables" +version = "0.9.0" +source = "git+https://github.com/feldera/iceberg-rust.git?rev=28c4800f198686f306bd80d11d03f232670feec3#28c4800f198686f306bd80d11d03f232670feec3" +dependencies = [ + "anyhow", + "async-trait", + "aws-config", + "aws-sdk-s3tables", + "iceberg", + "iceberg-storage-opendal", +] + [[package]] name = "iceberg-datafusion" version = "0.9.0" @@ -7258,7 +7296,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "07033963ba89ebaf1584d767badaa2e8fcec21aedea6b8c0346d487d49c28667" dependencies = [ "cfg-if", - "windows-targets 0.48.5", + "windows-targets 0.53.5", ] [[package]] @@ -9552,7 +9590,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" dependencies = [ "anyhow", - "itertools 0.14.0", + "itertools 0.11.0", "proc-macro2", "quote", "syn 2.0.117", @@ -9565,7 +9603,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b" dependencies = [ "anyhow", - "itertools 0.14.0", + "itertools 0.11.0", "proc-macro2", "quote", "syn 2.0.117", @@ -13946,7 +13984,7 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf221c93e13a30d793f7645a0e7762c55d169dbb0a49671918a2319d289b10bb" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.59.0", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index f986117d4a8..3b433eb54a7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -166,6 +166,7 @@ home = "=0.5.9" iceberg = { git = "https://github.com/feldera/iceberg-rust.git", rev = "28c4800f198686f306bd80d11d03f232670feec3" } iceberg-catalog-glue = { git = "https://github.com/feldera/iceberg-rust.git", rev = "28c4800f198686f306bd80d11d03f232670feec3" } iceberg-catalog-rest = { git = "https://github.com/feldera/iceberg-rust.git", rev = "28c4800f198686f306bd80d11d03f232670feec3" } +iceberg-catalog-s3tables = { git = "https://github.com/feldera/iceberg-rust.git", rev = "28c4800f198686f306bd80d11d03f232670feec3" } iceberg-datafusion = { git = "https://github.com/feldera/iceberg-rust.git", rev = "28c4800f198686f306bd80d11d03f232670feec3" } iceberg-storage-opendal = { git = "https://github.com/feldera/iceberg-rust.git", rev = "28c4800f198686f306bd80d11d03f232670feec3", features = ["opendal-s3", "opendal-gcs", "opendal-fs", "opendal-memory"] } impl-trait-for-tuples = "0.2" diff --git a/crates/adapters/Cargo.toml b/crates/adapters/Cargo.toml index f5c8080bf3a..576af7b0a4a 100644 --- a/crates/adapters/Cargo.toml +++ b/crates/adapters/Cargo.toml @@ -66,6 +66,7 @@ feldera-enterprise = [] iceberg-tests-fs = [] iceberg-tests-glue = [] iceberg-tests-rest = [] +iceberg-tests-s3tables = [] fips = ["rustls/fips"] bench-mode = [] with-postgres-cdc = ["etl", "etl-config", "etl-postgres"] diff --git a/crates/adapters/src/test.rs b/crates/adapters/src/test.rs index 149590d4b95..93d28f6b47b 100644 --- a/crates/adapters/src/test.rs +++ b/crates/adapters/src/test.rs @@ -49,7 +49,8 @@ mod datagen; any( feature = "iceberg-tests-fs", feature = "iceberg-tests-glue", - feature = "iceberg-tests-rest" + feature = "iceberg-tests-rest", + feature = "iceberg-tests-s3tables" ) ))] mod iceberg; @@ -58,8 +59,9 @@ use crate::catalog::InputCollectionHandle; use crate::format::get_input_format; use crate::transport::input_transport_config_to_endpoint; pub use data::{ - DatabricksPeople, DeltaTestStruct, EmbeddedStruct, IcebergTestStruct, KeyStruct, TestStruct, - TestStruct2, generate_test_batch, generate_test_batches, generate_test_batches_with_weights, + DatabricksPeople, DeltaTestStruct, EmbeddedStruct, IcebergTestStruct, KeyStruct, + S3TablesTestStruct, TestStruct, TestStruct2, generate_test_batch, generate_test_batches, + generate_test_batches_with_weights, }; use dbsp::circuit::{CircuitConfig, NodeId}; use dbsp::utils::Tup2; diff --git a/crates/adapters/src/test/data.rs b/crates/adapters/src/test/data.rs index 28c36bf881b..86c7c9789cd 100644 --- a/crates/adapters/src/test/data.rs +++ b/crates/adapters/src/test/data.rs @@ -954,6 +954,55 @@ deserialize_table_record!(IcebergTestStruct["IcebergTestStruct", Variant, 12] { (varbin, "varbin", false, ByteArray, |_| None) }); +/// Records in the Amazon S3 Tables integration-test table (`dev.test_table`). +/// +/// Mirrors the table created by the `s3tables` test-setup script: +/// `id BIGINT NOT NULL`, `name STRING`, `created_at TIMESTAMP`. +#[derive( + Debug, + Default, + PartialEq, + Eq, + PartialOrd, + Ord, + serde::Serialize, + Clone, + Hash, + SizeOf, + rkyv::Archive, + rkyv::Serialize, + rkyv::Deserialize, + IsNone, +)] +#[archive_attr(derive(Ord, Eq, PartialEq, PartialOrd))] +pub struct S3TablesTestStruct { + pub id: i64, + pub name: Option, + pub created_at: Option, +} + +impl S3TablesTestStruct { + pub fn schema() -> Vec { + vec![ + Field::new("id".into(), ColumnType::bigint(false)), + Field::new("name".into(), ColumnType::varchar(true)), + Field::new("created_at".into(), ColumnType::timestamp(true)), + ] + } +} + +serialize_table_record!(S3TablesTestStruct[3]{ + id["id"]: i64, + name["name"]: Option, + created_at["created_at"]: Option +}); + +deserialize_table_record!(S3TablesTestStruct["S3TablesTestStruct", Variant, 3] { + (id, "id", false, i64, |_| None), + (name, "name", true, Option, |_| Some(None)), + (created_at, "created_at", true, Option, |_| Some(None)) +}); + /// Struct will all types supported by the DeltaLake connector. #[derive( Debug, diff --git a/crates/adapters/src/test/iceberg.rs b/crates/adapters/src/test/iceberg.rs index 358ee59760e..990056c55c5 100644 --- a/crates/adapters/src/test/iceberg.rs +++ b/crates/adapters/src/test/iceberg.rs @@ -6,7 +6,9 @@ use crate::{ }; use crossbeam::channel::Receiver; use dbsp::DBData; -use feldera_sqllib::{ByteArray, F32, F64, Variant}; +use feldera_sqllib::Variant; +#[cfg(feature = "iceberg-tests-fs")] +use feldera_sqllib::{ByteArray, F32, F64}; use feldera_types::{ program_schema::Field, serde_with_context::{DeserializeWithContext, SerializeWithContext, SqlSerdeConfig}, @@ -21,7 +23,15 @@ use tracing_subscriber::{EnvFilter, layer::SubscriberExt, util::SubscriberInitEx #[cfg(feature = "iceberg-tests-fs")] use std::io::Write; -use super::{IcebergTestStruct, test_circuit}; +#[cfg(any( + feature = "iceberg-tests-fs", + feature = "iceberg-tests-glue", + feature = "iceberg-tests-rest" +))] +use super::IcebergTestStruct; +#[cfg(feature = "iceberg-tests-s3tables")] +use super::S3TablesTestStruct; +use super::test_circuit; fn init_logging() { let _ = tracing_subscriber::registry() @@ -327,6 +337,42 @@ fn iceberg_glue_s3_input_test() { assert_eq!(zset.len(), 2000000); } +#[test] +#[cfg(feature = "iceberg-tests-s3tables")] +fn iceberg_s3tables_input_test() { + use dbsp::trace::BatchReader; + + // Reads `dev.test_table` (schema `id BIGINT NOT NULL, name STRING, + // created_at TIMESTAMP`, 100 rows) from an Amazon S3 Tables bucket. + // + // Credentials and region resolve from the ambient AWS provider chain + // (environment variables, shared config file, or SSO profile), so no keys + // are embedded in the connector config. The resolved identity must be + // authorized for `s3tables:GetTable` (to locate the table metadata) and + // `s3tables:GetTableData` (the FileIO reads the metadata and data files). + // Run with AWS credentials configured, e.g. `AWS_PROFILE=` or + // `AWS_ACCESS_KEY_ID`/`AWS_SECRET_ACCESS_KEY`(/`AWS_SESSION_TOKEN`) exported. + let mut json_file = iceberg_snapshot_to_json::( + &S3TablesTestStruct::schema(), + &[ + ("catalog_type".to_string(), "s3tables".to_string()), + ( + "s3tables.table-bucket-arn".to_string(), + "arn:aws:s3tables:us-west-1:737834633458:bucket/iceberg-test".to_string(), + ), + ("table_name".to_string(), "dev.test_table".to_string()), + ("s3tables.region".to_string(), "us-west-1".to_string()), + ("s3.region".to_string(), "us-west-1".to_string()), + ] + .into_iter() + .collect::>(), + ); + + let zset = file_to_zset::(json_file.as_file_mut()); + + assert_eq!(zset.len(), 100); +} + #[test] #[cfg(feature = "iceberg-tests-rest")] fn iceberg_rest_s3_input_test() { diff --git a/crates/feldera-types/src/transport/iceberg.rs b/crates/feldera-types/src/transport/iceberg.rs index 59d9923900d..da84a6527a0 100644 --- a/crates/feldera-types/src/transport/iceberg.rs +++ b/crates/feldera-types/src/transport/iceberg.rs @@ -44,6 +44,8 @@ pub enum IcebergCatalogType { Rest, #[serde(rename = "glue")] Glue, + #[serde(rename = "s3tables")] + S3Tables, } /// AWS Glue catalog config. @@ -136,6 +138,44 @@ pub struct RestCatalogConfig { pub resource: Option, } +/// Amazon S3 Tables catalog config. +#[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema)] +pub struct S3TablesCatalogConfig { + /// ARN of the S3 table bucket that contains the table. + /// + /// Note that this is the ARN of the table *bucket*, not of an individual table, + /// e.g., `"arn:aws:s3tables:us-east-2:123456789012:bucket/my-bucket"`. + #[serde(rename = "s3tables.table-bucket-arn")] + pub table_bucket_arn: Option, + + /// Custom endpoint URL for the S3 Tables service. + /// + /// Primarily used to target a local or mock S3 Tables implementation for testing. + /// When omitted, the default regional endpoint is used. + #[serde(rename = "s3tables.endpoint")] + pub endpoint: Option, + + /// Access key id used to access the S3 Tables catalog. + #[serde(rename = "s3tables.access-key-id")] + pub access_key_id: Option, + + /// Secret access key used to access the S3 Tables catalog. + #[serde(rename = "s3tables.secret-access-key")] + pub secret_access_key: Option, + + /// Static session token used to access the S3 Tables catalog. + #[serde(rename = "s3tables.session-token")] + pub session_token: Option, + + /// Profile used to access the S3 Tables catalog. + #[serde(rename = "s3tables.profile-name")] + pub profile_name: Option, + + /// Region of the S3 Tables catalog. + #[serde(rename = "s3tables.region")] + pub region: Option, +} + /// Iceberg input connector configuration. #[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema)] pub struct IcebergReaderConfig { @@ -219,7 +259,8 @@ pub struct IcebergReaderConfig { /// Specifies the catalog type used to access the Iceberg table. /// - /// Supported options include "rest" and "glue". This property is mutually exclusive with `metadata_location`. + /// Supported options include "rest", "glue", and "s3tables". This property is mutually + /// exclusive with `metadata_location`. pub catalog_type: Option, #[serde(flatten)] @@ -228,6 +269,9 @@ pub struct IcebergReaderConfig { #[serde(flatten)] pub rest_catalog_config: RestCatalogConfig, + #[serde(flatten)] + pub s3tables_catalog_config: S3TablesCatalogConfig, + /// Storage options for configuring backend object store. /// /// See the [list of available options in PyIceberg documentation](https://py.iceberg.apache.org/configuration/#fileio). @@ -241,6 +285,7 @@ impl IcebergReaderConfig { self.validate_table_name()?; self.validate_glue_catalog_config()?; self.validate_rest_catalog_config()?; + self.validate_s3tables_catalog_config()?; Ok(()) } @@ -293,6 +338,40 @@ impl IcebergReaderConfig { Ok(()) } + /// Reject S3 Tables catalog config when 'catalog_type' isn't set to 's3tables'. + pub fn validate_s3tables_catalog_config(&self) -> Result<(), String> { + if self.catalog_type == Some(IcebergCatalogType::S3Tables) { + if self.s3tables_catalog_config.table_bucket_arn.is_none() { + return Err(r#"missing S3 table bucket ARN; set the 's3tables.table-bucket-arn' property to the ARN of the S3 table bucket (e.g., 'arn:aws:s3tables:us-east-2:123456789012:bucket/my-bucket') when using "catalog_type" = "s3tables""#.to_string()); + } + } else { + ensure_s3tables_property_not_set( + &self.s3tables_catalog_config.table_bucket_arn, + "table-bucket-arn", + )?; + ensure_s3tables_property_not_set(&self.s3tables_catalog_config.endpoint, "endpoint")?; + ensure_s3tables_property_not_set( + &self.s3tables_catalog_config.access_key_id, + "access-key-id", + )?; + ensure_s3tables_property_not_set( + &self.s3tables_catalog_config.secret_access_key, + "secret-access-key", + )?; + ensure_s3tables_property_not_set( + &self.s3tables_catalog_config.session_token, + "session-token", + )?; + ensure_s3tables_property_not_set( + &self.s3tables_catalog_config.profile_name, + "profile-name", + )?; + ensure_s3tables_property_not_set(&self.s3tables_catalog_config.region, "region")?; + } + + Ok(()) + } + /// Table name must be configured iff 'catalog_type' is set. pub fn validate_table_name(&self) -> Result<(), String> { if self.catalog_type.is_none() && self.table_name.is_some() { @@ -336,6 +415,16 @@ fn ensure_rest_property_not_set(property: &Option, name: &str) -> Result<( } } +fn ensure_s3tables_property_not_set(property: &Option, name: &str) -> Result<(), String> { + if property.is_some() { + Err(format!( + r#"unexpected 's3tables.{name}' property—S3 Tables catalog configuration properties are only valid when "catalog_type" = "s3tables""# + )) + } else { + Ok(()) + } +} + impl IcebergReaderConfig { /// `true` if the configuration requires taking an initial snapshot of the table. pub fn snapshot(&self) -> bool { @@ -354,3 +443,84 @@ impl IcebergReaderConfig { ) } } + +#[cfg(test)] +mod test { + use super::*; + use serde_json::json; + + fn config(value: serde_json::Value) -> IcebergReaderConfig { + serde_json::from_value(value).unwrap() + } + + #[test] + fn s3tables_config_deserializes() { + let config = config(json!({ + "mode": "snapshot", + "catalog_type": "s3tables", + "table_name": "namespace.table", + "s3tables.table-bucket-arn": "arn:aws:s3tables:us-east-2:123456789012:bucket/my-bucket", + "s3tables.region": "us-east-2", + "s3tables.access-key-id": "key", + "s3tables.secret-access-key": "secret", + "s3tables.session-token": "token", + "s3tables.profile-name": "profile", + "s3tables.endpoint": "http://localhost:4566", + })); + + // Each prefixed key must land on its struct field, not in the catch-all + // `fileio_config` map (which is what happens on a rename mismatch). + let s3tables = &config.s3tables_catalog_config; + assert_eq!( + s3tables.table_bucket_arn.as_deref(), + Some("arn:aws:s3tables:us-east-2:123456789012:bucket/my-bucket") + ); + assert_eq!(s3tables.region.as_deref(), Some("us-east-2")); + assert_eq!(s3tables.access_key_id.as_deref(), Some("key")); + assert_eq!(s3tables.secret_access_key.as_deref(), Some("secret")); + assert_eq!(s3tables.session_token.as_deref(), Some("token")); + assert_eq!(s3tables.profile_name.as_deref(), Some("profile")); + assert_eq!(s3tables.endpoint.as_deref(), Some("http://localhost:4566")); + assert!(config.fileio_config.is_empty()); + + config.validate_catalog_config().unwrap(); + } + + #[test] + fn s3tables_requires_table_bucket_arn() { + let err = config(json!({ + "mode": "snapshot", + "catalog_type": "s3tables", + "table_name": "namespace.table", + })) + .validate_catalog_config() + .unwrap_err(); + assert!(err.contains("s3tables.table-bucket-arn"), "{err}"); + } + + #[test] + fn s3tables_props_rejected_for_other_catalog() { + let err = config(json!({ + "mode": "snapshot", + "catalog_type": "glue", + "table_name": "namespace.table", + "glue.warehouse": "s3://warehouse/", + "s3tables.region": "us-east-2", + })) + .validate_catalog_config() + .unwrap_err(); + assert!(err.contains("s3tables.region"), "{err}"); + } + + #[test] + fn s3tables_props_rejected_without_catalog() { + let err = config(json!({ + "mode": "snapshot", + "metadata_location": "s3://warehouse/metadata.json", + "s3tables.table-bucket-arn": "arn:aws:s3tables:us-east-2:123456789012:bucket/my-bucket", + })) + .validate_catalog_config() + .unwrap_err(); + assert!(err.contains("s3tables.table-bucket-arn"), "{err}"); + } +} diff --git a/crates/iceberg/Cargo.toml b/crates/iceberg/Cargo.toml index 59f35f31f74..deda2cf94ce 100644 --- a/crates/iceberg/Cargo.toml +++ b/crates/iceberg/Cargo.toml @@ -22,6 +22,7 @@ iceberg = { workspace = true } iceberg-datafusion = { workspace = true } iceberg-catalog-glue = { workspace = true } iceberg-catalog-rest = { workspace = true } +iceberg-catalog-s3tables = { workspace = true } iceberg-storage-opendal = { workspace = true } url = { workspace = true } chrono = { workspace = true } diff --git a/crates/iceberg/src/input.rs b/crates/iceberg/src/input.rs index 5d386661900..fbfab988835 100644 --- a/crates/iceberg/src/input.rs +++ b/crates/iceberg/src/input.rs @@ -37,6 +37,10 @@ use iceberg_catalog_glue::{ use iceberg_catalog_rest::{ RestCatalogBuilder, REST_CATALOG_PROP_URI, REST_CATALOG_PROP_WAREHOUSE, }; +use iceberg_catalog_s3tables::{ + S3TablesCatalogBuilder, S3TABLES_CATALOG_PROP_ENDPOINT_URL, + S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN, +}; use iceberg_datafusion::IcebergStaticTableProvider; use iceberg_storage_opendal::OpenDalResolvingStorageFactory; use log::{debug, info, trace}; @@ -56,6 +60,16 @@ fn storage_factory() -> Arc { Arc::new(OpenDalResolvingStorageFactory::new()) } +// `iceberg-catalog-s3tables` reads AWS credentials for its S3 Tables API client +// from these property keys but does not re-export them (they live in the crate's +// private `utils` module). Mirror them here. See +// crates/catalog/s3tables/src/utils.rs in the iceberg-rust fork. +const S3TABLES_PROP_ACCESS_KEY_ID: &str = "aws_access_key_id"; +const S3TABLES_PROP_SECRET_ACCESS_KEY: &str = "aws_secret_access_key"; +const S3TABLES_PROP_SESSION_TOKEN: &str = "aws_session_token"; +const S3TABLES_PROP_PROFILE_NAME: &str = "profile_name"; +const S3TABLES_PROP_REGION_NAME: &str = "region_name"; + enum SnapshotDescr { /// Open the latest snapshot (default) Latest, @@ -477,6 +491,7 @@ impl IcebergInputEndpointInner { None => self.open_table_no_catalog().await, Some(IcebergCatalogType::Glue) => self.open_table_glue().await, Some(IcebergCatalogType::Rest) => self.open_table_rest().await, + Some(IcebergCatalogType::S3Tables) => self.open_table_s3tables().await, } // // TODO: Validate that table schema matches relation schema @@ -689,6 +704,103 @@ impl IcebergInputEndpointInner { }) } + async fn open_table_s3tables(&self) -> Result { + let mut props = self.config.fileio_config.clone(); + + // Safe due to checks in 'validate_catalog_config'. + props.insert( + S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN.to_string(), + self.config + .s3tables_catalog_config + .table_bucket_arn + .as_ref() + .unwrap() + .clone(), + ); + + if let Some(endpoint) = self.config.s3tables_catalog_config.endpoint.as_ref() { + props.insert( + S3TABLES_CATALOG_PROP_ENDPOINT_URL.to_string(), + endpoint.clone(), + ); + } + + // Credentials for the S3 Tables API client. These use a different key + // namespace than the `s3.*` `fileio_config` keys that authenticate the + // FileIO used to read the table's data files, so both can coexist in the + // same property map (each consumer ignores the other's keys). + self.config + .s3tables_catalog_config + .access_key_id + .as_ref() + .map(|aws_access_key_id| { + props.insert( + S3TABLES_PROP_ACCESS_KEY_ID.to_string(), + aws_access_key_id.clone(), + ) + }); + + self.config + .s3tables_catalog_config + .secret_access_key + .as_ref() + .map(|aws_secret_access_key| { + props.insert( + S3TABLES_PROP_SECRET_ACCESS_KEY.to_string(), + aws_secret_access_key.clone(), + ) + }); + + self.config + .s3tables_catalog_config + .session_token + .as_ref() + .map(|session_token| { + props.insert( + S3TABLES_PROP_SESSION_TOKEN.to_string(), + session_token.clone(), + ) + }); + + self.config + .s3tables_catalog_config + .profile_name + .as_ref() + .map(|profile_name| { + props.insert(S3TABLES_PROP_PROFILE_NAME.to_string(), profile_name.clone()) + }); + + self.config + .s3tables_catalog_config + .region + .as_ref() + .map(|region_name| { + props.insert(S3TABLES_PROP_REGION_NAME.to_string(), region_name.clone()) + }); + + let catalog = S3TablesCatalogBuilder::default() + .with_storage_factory(storage_factory()) + .load("s3tables".to_string(), props) + .await + .map_err(|e| { + ControllerError::input_transport_error( + &self.endpoint_name, + true, + anyhow!("error creating S3 Tables catalog client: {e}"), + ) + })?; + + let table_ident = self.table_ident().unwrap()?; + + catalog.load_table(&table_ident).await.map_err(|e| { + ControllerError::input_transport_error( + &self.endpoint_name, + true, + anyhow!("error loading Iceberg table: {e}"), + ) + }) + } + /// Validate the filter expression specified in the 'snapshot_filter' parameter. fn validate_snapshot_filter(&self) -> Result<(), ControllerError> { if let Some(filter) = &self.config.snapshot_filter { diff --git a/crates/iceberg/src/test/README.md b/crates/iceberg/src/test/README.md index a89921ad1a9..6de4cef7142 100644 --- a/crates/iceberg/src/test/README.md +++ b/crates/iceberg/src/test/README.md @@ -7,6 +7,7 @@ and are feature gated using: * `iceberg-tests-fs` - enables tests using local file system * `iceberg-tests-glue` - enables tests using S3 and AWS Glue catalog * `iceberg-tests-rest` - enables tests using S3 and Rest catalog. +* `iceberg-tests-s3tables` - enables tests using an Amazon S3 Tables catalog. The `iceberg-rust` crate does not yes support writing Iceberg tables; therefore we use `pyiceberg` to create test tables for the Iceberg source connector. The @@ -46,6 +47,22 @@ implementation of the REST catalog from Databricks and bind it to the Glue catal * `cargo test --features="iceberg-tests-rest" iceberg_rest_s3_input_test` +## S3 Tables catalog test + +This test reads the table `dev.test_table` (schema `id BIGINT NOT NULL, name STRING, +created_at TIMESTAMP`, 100 rows) from an +[Amazon S3 Tables](https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables-buckets.html) +table bucket (`arn:aws:s3tables:us-west-1:737834633458:bucket/iceberg-test`). + +Credentials and region resolve from the ambient AWS provider chain, so the test embeds no +keys. Provide credentials for an identity authorized for `s3tables:GetTable` (to locate the +table metadata) and `s3tables:GetTableData` (to read the metadata and data files) on the +table (talk to leonid@feldera.com), for example via an SSO profile or exported access keys: + +* `export AWS_PROFILE=` (after `aws sso login`), or + `export AWS_ACCESS_KEY_ID=... AWS_SECRET_ACCESS_KEY=... AWS_SESSION_TOKEN=...` +* Run the following command in the `adapters` crate: `cargo test --features="iceberg-tests-s3tables" iceberg_s3tables_input_test` + # Running tests in CI Currently only Glue and FS-based tests run in CI. diff --git a/crates/pipeline-manager/src/api/main.rs b/crates/pipeline-manager/src/api/main.rs index 462a731792a..e7f26ed1509 100644 --- a/crates/pipeline-manager/src/api/main.rs +++ b/crates/pipeline-manager/src/api/main.rs @@ -416,6 +416,7 @@ It contains the following fields: feldera_types::transport::iceberg::IcebergCatalogType, feldera_types::transport::iceberg::RestCatalogConfig, feldera_types::transport::iceberg::GlueCatalogConfig, + feldera_types::transport::iceberg::S3TablesCatalogConfig, feldera_types::transport::postgres::PostgresReaderConfig, feldera_types::transport::postgres::PostgresCdcReaderConfig, feldera_types::transport::postgres::PostgresWriterConfig, diff --git a/docs.feldera.com/docs/connectors/sources/iceberg.md b/docs.feldera.com/docs/connectors/sources/iceberg.md index a69abad5854..b101c154db8 100644 --- a/docs.feldera.com/docs/connectors/sources/iceberg.md +++ b/docs.feldera.com/docs/connectors/sources/iceberg.md @@ -17,9 +17,9 @@ a Feldera pipeline. Currently, the connector supports batch reads, allowing user load a static snapshot of the table. However, it does not yet support ingesting incremental changes. Incremental ingestion capabilities are planned for future releases. -The connector is compatible with REST and AWS Glue catalogs and also supports direct -table reads without a catalog, provided the location of the metadata file. Supported -storage systems include S3, GCS, and local file systems. +The connector is compatible with REST, AWS Glue, and Amazon S3 Tables catalogs and also +supports direct table reads without a catalog, provided the location of the metadata file. +Supported storage systems include S3, GCS, and local file systems. The Iceberg input connector does not yet support [fault tolerance](/pipelines/fault-tolerance). @@ -34,7 +34,7 @@ The Iceberg input connector does not yet support [fault tolerance](/pipelines/fa | `datetime` | string |

Optional timestamp for the snapshot in the ISO-8601/RFC-3339 format, e.g., "2024-12-09T16:09:53+00:00". When this option is set, the connector reads the version of the table as of the specified point in time (based on the server time recorded in the transaction log, not the event time encoded in the data).

Note: at most one of `version` and `datetime` options can be specified. When neither of the two options is specified, the latest committed version of the table is used.

| | `metadata_location` | string | Location of the table metadata JSON file. This property is used to access an Iceberg table directly, without a catalog. It is mutually exclusive with the `catalog_type` property.| | `table_name` | string | Specifies the Iceberg table name within the catalog in the `namespace.table` format. This option is applicable when an Iceberg catalog is configured using the `catalog_type` property.| -| `catalog_type` | enum | Type of the Iceberg catalog used to access the table. Supported options include `rest` and `glue`. This property is mutually exclusive with `metadata_location`.| +| `catalog_type` | enum | Type of the Iceberg catalog used to access the table. Supported options include `rest`, `glue`, and `s3tables`. This property is mutually exclusive with `metadata_location`.| @@ -77,6 +77,24 @@ The following properties are used when `catalog_type` is set to `glue` to config [*]: These fields are required when the `catalog_type` property is set to `glue`. +### S3 Tables catalog configuration + +The following properties are used when `catalog_type` is set to `s3tables` to configure +access to an [Amazon S3 Tables](https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables-buckets.html) +table bucket. + +| Property | Type | Description | +|---------------------------------|--------|---------------| +| `s3tables.table-bucket-arn`* | string | ARN of the S3 table bucket that contains the table. Note that this is the ARN of the table _bucket_, not of an individual table, e.g., `arn:aws:s3tables:us-east-2:123456789012:bucket/my-bucket`.| +| `s3tables.endpoint` | string | Custom endpoint URL for the S3 Tables service. Primarily used to target a local or mock S3 Tables implementation for testing.| +| `s3tables.access-key-id` | string | Access key id used to access the S3 Tables catalog.| +| `s3tables.secret-access-key` | string | Secret access key used to access the S3 Tables catalog.| +| `s3tables.session-token` | string | Static session token used to access the S3 Tables catalog. Required when using temporary credentials.| +| `s3tables.profile-name` | string | Profile used to access the S3 Tables catalog.| +| `s3tables.region` | string | Region of the S3 Tables catalog.| + +[*]: These fields are required when the `catalog_type` property is set to `s3tables`. + ### FileIO configuration Iceberg works with the concept of a FileIO which is a pluggable module for reading, writing, and deleting files. @@ -256,6 +274,44 @@ create table iceberg_table( ); ``` +### Read an Iceberg table from S3 through the S3 Tables catalog + +Create an Iceberg input connector to read a snapshot of a table stored in an +[Amazon S3 Tables](https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables-buckets.html) +table bucket. As with the AWS Glue catalog, the configuration specifies AWS credentials +separately for the S3 Tables catalog API (`s3tables.*`) and for reading the table's data +files from S3 (`s3.*`). These credentials can either be the same, when using a single IAM +identity for both, or different, when using separate IAM identities. + +```sql +create table iceberg_table( + id bigint, + name STRING, + b BOOLEAN, + ts TIMESTAMP, + dt DATE +) with ( + 'materialized' = 'true', + 'connectors' = '[{ + "transport": { + "name": "iceberg_input", + "config": { + "mode": "snapshot", + "catalog_type": "s3tables", + "table_name": "iceberg_test.test_table", + "s3tables.table-bucket-arn": "arn:aws:s3tables:us-east-2:123456789012:bucket/my-bucket", + "s3tables.access-key-id": "", + "s3tables.secret-access-key": "", + "s3tables.region": "us-east-2", + "s3.access-key-id": "", + "s3.secret-access-key": "", + "s3.region": "us-east-2" + } + } + }]' +); +``` + ### Read an Iceberg table from S3 through a REST catalog Create an Iceberg input connector to read a snapshot of a table stored in an S3 bucket diff --git a/openapi.json b/openapi.json index b29f8a7a19e..7e13c03fa31 100644 --- a/openapi.json +++ b/openapi.json @@ -9835,7 +9835,8 @@ "type": "string", "enum": [ "rest", - "glue" + "glue", + "s3tables" ] }, "IcebergIngestMode": { @@ -9855,6 +9856,9 @@ { "$ref": "#/components/schemas/RestCatalogConfig" }, + { + "$ref": "#/components/schemas/S3TablesCatalogConfig" + }, { "type": "object", "required": [ @@ -13291,6 +13295,47 @@ } } }, + "S3TablesCatalogConfig": { + "type": "object", + "description": "Amazon S3 Tables catalog config.", + "properties": { + "s3tables.access-key-id": { + "type": "string", + "description": "Access key id used to access the S3 Tables catalog.", + "nullable": true + }, + "s3tables.endpoint": { + "type": "string", + "description": "Custom endpoint URL for the S3 Tables service.\n\nPrimarily used to target a local or mock S3 Tables implementation for testing.\nWhen omitted, the default regional endpoint is used.", + "nullable": true + }, + "s3tables.profile-name": { + "type": "string", + "description": "Profile used to access the S3 Tables catalog.", + "nullable": true + }, + "s3tables.region": { + "type": "string", + "description": "Region of the S3 Tables catalog.", + "nullable": true + }, + "s3tables.secret-access-key": { + "type": "string", + "description": "Secret access key used to access the S3 Tables catalog.", + "nullable": true + }, + "s3tables.session-token": { + "type": "string", + "description": "Static session token used to access the S3 Tables catalog.", + "nullable": true + }, + "s3tables.table-bucket-arn": { + "type": "string", + "description": "ARN of the S3 table bucket that contains the table.\n\nNote that this is the ARN of the table *bucket*, not of an individual table,\ne.g., `\"arn:aws:s3tables:us-east-2:123456789012:bucket/my-bucket\"`.", + "nullable": true + } + } + }, "SampleStatistics": { "type": "object", "description": "One sample of time-series data.", diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/IcebergCatalogType.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/IcebergCatalogType.java index 96cd2fe4b01..42ee54df229 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/IcebergCatalogType.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/IcebergCatalogType.java @@ -2,6 +2,7 @@ import com.fasterxml.jackson.annotation.JsonProperty; public enum IcebergCatalogType { - @JsonProperty("rest") Rest, - @JsonProperty("glue") Glue, + @JsonProperty("rest") Rest, + @JsonProperty("glue") Glue, + @JsonProperty("s3tables") S3Tables, } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/IcebergReaderConfig.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/IcebergReaderConfig.java index 24ecc59e328..a32f5964a42 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/IcebergReaderConfig.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/connectors/config/IcebergReaderConfig.java @@ -66,6 +66,15 @@ public class IcebergReaderConfig implements IValidateConfig { @Nullable @JsonProperty("rest.audience") public String restAudience = null; @Nullable @JsonProperty("rest.resource") public String restResource = null; + // S3 Tables catalog fields (inlined from S3TablesCatalogConfig) + @Nullable @JsonProperty("s3tables.table-bucket-arn") public String s3tablesTableBucketArn = null; + @Nullable @JsonProperty("s3tables.endpoint") public String s3tablesEndpoint = null; + @Nullable @JsonProperty("s3tables.access-key-id") public String s3tablesAccessKeyId = null; + @Nullable @JsonProperty("s3tables.secret-access-key") public String s3tablesSecretAccessKey = null; + @Nullable @JsonProperty("s3tables.session-token") public String s3tablesSessionToken = null; + @Nullable @JsonProperty("s3tables.profile-name") public String s3tablesProfileName = null; + @Nullable @JsonProperty("s3tables.region") public String s3tablesRegion = null; + /** Absorbs flattened {@code fileio_config} keys so they are not rejected as unknown. */ @JsonAnySetter public void setFileioOption(String key, Object value) {} @@ -137,6 +146,22 @@ public boolean validate(ConfigReporter reporter) { ok &= checkRestPropAbsent(reporter, restAudience, "audience"); ok &= checkRestPropAbsent(reporter, restResource, "resource"); } + if (catalogType == IcebergCatalogType.S3Tables) { + if (s3tablesTableBucketArn == null) { + reporter.warn("Invalid configuration", + "missing S3 table bucket ARN—set \"s3tables.table-bucket-arn\" when using" + + " \"catalog_type\": \"s3tables\""); + ok = false; + } + } else { + ok &= checkS3tablesPropAbsent(reporter, s3tablesTableBucketArn, "table-bucket-arn"); + ok &= checkS3tablesPropAbsent(reporter, s3tablesEndpoint, "endpoint"); + ok &= checkS3tablesPropAbsent(reporter, s3tablesAccessKeyId, "access-key-id"); + ok &= checkS3tablesPropAbsent(reporter, s3tablesSecretAccessKey, "secret-access-key"); + ok &= checkS3tablesPropAbsent(reporter, s3tablesSessionToken, "session-token"); + ok &= checkS3tablesPropAbsent(reporter, s3tablesProfileName, "profile-name"); + ok &= checkS3tablesPropAbsent(reporter, s3tablesRegion, "region"); + } return ok; } @@ -157,4 +182,13 @@ private boolean checkRestPropAbsent(ConfigReporter reporter, Object value, Strin } return true; } + + private boolean checkS3tablesPropAbsent(ConfigReporter reporter, Object value, String name) { + if (value != null) { + reporter.warnPath("s3tables." + name, "Invalid configuration", + "\"s3tables." + name + "\" is only valid when \"catalog_type\": \"s3tables\""); + return false; + } + return true; + } }