feldera · ryzhyk · Jul 9, 2026 · Jul 9, 2026 · Jul 9, 2026 · mihaibudiu
diff --git a/Cargo.lock b/Cargo.lock
diff --git 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
@@ -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
@@ -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
@@ -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<String>,
+    pub created_at: Option<Timestamp>,
+}
+
+impl S3TablesTestStruct {
+    pub fn schema() -> Vec<Field> {
+        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<String>,
+    created_at["created_at"]: Option<Timestamp>
+});
+
+deserialize_table_record!(S3TablesTestStruct["S3TablesTestStruct", Variant, 3] {
+    (id, "id", false, i64, |_| None),
+    (name, "name", true, Option<String>, |_| Some(None)),
+    (created_at, "created_at", true, Option<Timestamp>, |_| 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
@@ -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=<profile>` or
+    // `AWS_ACCESS_KEY_ID`/`AWS_SECRET_ACCESS_KEY`(/`AWS_SESSION_TOKEN`) exported.
+    let mut json_file = iceberg_snapshot_to_json::<S3TablesTestStruct>(
+        &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::<HashMap<_, _>>(),
+    );
+
+    let zset = file_to_zset::<S3TablesTestStruct>(json_file.as_file_mut());
+
+    assert_eq!(zset.len(), 100);
+}
+
 #[test]
 #[cfg(feature = "iceberg-tests-rest")]
 fn iceberg_rest_s3_input_test() {