[SS-366] update iceberg-rust dep to v0.10.1

Cargo.toml

@@ -279,8 +279,8 @@ annotate-snippets = "0.12.15"
 anyhow = "1.0.102"
 array-concat = "0.5.5"
 arrayvec = "0.7.6"
-arrow = { version = "57", default-features = false }
-arrow-ipc = "57"
+arrow = { version = "58", default-features = false }
+arrow-ipc = "58"
 askama = { version = "0.12.1", default-features = false, features = ["config", "serde-json"] }
 assert_cmd = "2.2.2"
 async-compression = { version = "0.4.27", features = ["bzip2-sys", "gzip", "tokio", "xz", "zstd"] }
@@ -385,9 +385,9 @@ hyper-0-14 = { package = "hyper", version = "0.14", features = ["client", "tcp"]
 hyper-openssl = "0.10.2"
 hyper-util = "0.1.20"
 tower-service = "0.3.3"
-iceberg = "0.9.0"
-iceberg-catalog-rest = "0.9.0"
-iceberg-storage-opendal = { version = "0.9.0", default-features = false, features = ["opendal-s3", "opendal-gcs"] }
+iceberg = "0.10.1"
+iceberg-catalog-rest = "0.10.1"
+iceberg-storage-opendal = { version = "0.10.1", default-features = false, features = ["opendal-s3", "opendal-gcs"] }
 imbl = { version = "7.0.0", features = ["serde"] }
 include_dir = "0.7.4"
 indexmap = { version = "2.10.0", default-features = false, features = ["std"] }
@@ -435,7 +435,7 @@ opentelemetry_sdk = { version = "0.31.0", features = ["experimental_trace_batch_
 ordered-float = { version = "5.1.0", features = ["serde"] }
 os_info = "3.11.0"
 owo-colors = "4.3.0"
-parquet = { version = "57", default-features = false, features = ["arrow", "async", "brotli", "flate2", "flate2-zlib-rs", "lz4", "snap", "zstd"] }
+parquet = { version = "58", default-features = false, features = ["arrow", "async", "brotli", "flate2", "flate2-zlib-rs", "lz4", "snap", "zstd"] }
 paste = "1.0.15"
 pem = "3.0.6"
 phf = { version = "0.13.1", features = ["uncased"] }
@@ -474,6 +474,9 @@ rdkafka = { version = "0.29.0", features = ["cmake-build", "libz-static", "ssl-v
 rdkafka-sys = { version = "4.3.0", features = ["cmake-build", "libz-static", "ssl-vendored", "zstd"] }
 regex = "1.12.3"
 regex-syntax = "0.8.10"
+# Must stay compatible with the version `iceberg-storage-opendal` uses, otherwise
+# the `ProvideCredential` we implement is a different trait than the one it wants.
+reqsign-core = "3.0.0"
 reqwest = { version = "0.12.28", features = ["blocking", "charset", "cookies", "default-tls", "http2", "json", "native-tls-vendored", "stream"] }
 reqwest-middleware = { version = "0.4.2", features = ["json"] }
 reqwest-retry = "0.8.0"
@@ -663,10 +666,10 @@ tiberius = { git = "https://github.com/MaterializeInc/tiberius", rev="64ca594cc2
 async-compression = { git = "https://github.com/MaterializeInc/async-compression.git", rev = "fe7411eb6104a02a89e2c3a76ab326dd6594214d" }
 
 # Custom iceberg features for mz
-# All changes should go to the `mz_v0.9.0` branch.
-iceberg = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "dedd9231ee88ee979b648e14792878b40e74c20a" }
-iceberg-catalog-rest = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "dedd9231ee88ee979b648e14792878b40e74c20a" }
-iceberg-storage-opendal = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "dedd9231ee88ee979b648e14792878b40e74c20a" }
+# All changes should go to the `mz_v0.10.x` branch.
+iceberg = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "a51ff2c8dce700a1c158c372f8ce94906b5631b1" }
+iceberg-catalog-rest = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "a51ff2c8dce700a1c158c372f8ce94906b5631b1" }
+iceberg-storage-opendal = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "a51ff2c8dce700a1c158c372f8ce94906b5631b1" }
 
 # Custom duckdb crate to support mz needs
 # All changes should go to the `mz_changes` branch.

deny.toml

@@ -104,10 +104,8 @@ skip = [
     { name = "socket2", version = "0.5.10" },
     # Used by azure_core
     { name = "quick-xml", version = "0.31.0" },
-    # Used by reqsign (via iceberg); opendal pulls quick-xml 0.38
-    { name = "quick-xml", version = "0.37.5" },
-    # Used by opendal via iceberg
-    { name = "quick-xml", version = "0.38.4" },
+    # Used by aws-sdk-*; opendal (via iceberg) pulls quick-xml 0.41
+    { name = "quick-xml", version = "0.39.4" },
     # Used by sentry
     { name = "reqwest", version = "0.13.3" },
     # Conflicts between `bon` in apache-avro and `derive_builder` in iceberg
@@ -126,7 +124,21 @@ skip = [
     { name = "phf_codegen", version = "0.12.1" },
     # duckdb
     { name = "fallible-iterator", version = "0.3.0" },
+    # duckdb 1.4 pins arrow 57. The workspace is on arrow 58, which iceberg requires.
+    { name = "arrow", version = "57.1.0" },
+    { name = "arrow-arith", version = "57.1.0" },
+    { name = "arrow-array", version = "57.2.0" },
+    { name = "arrow-buffer", version = "57.3.0" },
+    { name = "arrow-cast", version = "57.2.0" },
+    { name = "arrow-data", version = "57.2.0" },
+    { name = "arrow-ord", version = "57.2.0" },
+    { name = "arrow-row", version = "57.1.0" },
+    { name = "arrow-schema", version = "57.2.0" },
+    { name = "arrow-select", version = "57.2.0" },
+    { name = "arrow-string", version = "57.1.0" },
+    { name = "lz4_flex", version = "0.12.1" },
     # arrow
+    { name = "hashbrown", version = "0.15.3" },
     { name = "hashbrown", version = "0.16.1" },
     # Used by dynfmt; iceberg/typetag pulls in v0.4.
     { name = "erased-serde", version = "0.3.26" },
@@ -135,10 +147,17 @@ skip = [
     { name = "core-foundation", version = "0.10.1" },
     { name = "security-framework", version = "3.7.0" },
     { name = "openssl-probe", version = "0.2.1" },
-    # reqsign (via iceberg-storage-opendal / opendal) pins older deps
-    # than the workspace.
-    { name = "jsonwebtoken", version = "9.3.1" },
-    { name = "quick-xml", version = "0.37.5" },
+    # reqsign 3 (via iceberg-storage-opendal / opendal) is on the RustCrypto 0.11/0.12
+    # line while the rest of the workspace is still on 0.10.
+    { name = "base64", version = "0.22.1" },
+    { name = "block-buffer", version = "0.10.4" },
+    { name = "const-oid", version = "0.9.6" },
+    { name = "crypto-common", version = "0.1.6" },
+    { name = "digest", version = "0.10.7" },
+    { name = "hmac", version = "0.12.1" },
+    { name = "md-5", version = "0.10.6" },
+    { name = "sha1", version = "0.10.6" },
+    { name = "sha2", version = "0.10.9" },
     # aws-lc-rs (via jsonwebtoken 10) and ring pull different `untrusted`.
     { name = "untrusted", version = "0.7.1" },
     # Held back by lazy_static 1.4.0 (used by num-bigint-dig).
@@ -154,7 +173,6 @@ skip = [
     # Pulled by launchdarkly-server-sdk 3.x via launchdarkly-sdk-transport /
     # eventsource-client (proxy/timeout/rustls stack and the SDK's RNG path).
     # NB: tower 0.4.13 is already skipped above (mz-deploy).
-    { name = "rustls-native-certs", version = "0.7.3" },
     { name = "rand", version = "0.10.1" },
     { name = "rand_core", version = "0.10.1" },
     { name = "getrandom", version = "0.4.2" },
@@ -217,14 +235,20 @@ wrappers = [
     "launchdarkly-server-sdk-evaluation",
     "launchdarkly-sdk-transport",
     "native-tls",
-    "opendal",
+    "opendal-core",
+    "opendal-layer-retry",
+    "opendal-service-gcs",
+    "opendal-service-s3",
     "os_info",
     "postgres",
     "pprof",
     "prost-build",
     # TODO(guswynn): switch to tracing in rdkafka
     "rdkafka",
-    "reqsign",
+    "reqsign-aws-core",
+    "reqsign-aws-v4",
+    "reqsign-core",
+    "reqsign-google",
     "reqwest",
     "rustls",
     "sqlparser",

src/persist-types/src/parquet.rs

@@ -176,7 +176,7 @@ pub fn encode_arrays<W: Write + Send>(
         .set_compression(config.compression.into())
         .set_writer_version(WriterVersion::PARQUET_2_0)
         .set_data_page_size_limit(1024 * 1024)
-        .set_max_row_group_size(usize::MAX)
+        .set_max_row_group_row_count(None)
         .build();
     let mut writer = ArrowWriter::try_new(w, Arc::clone(&schema), Some(props))?;
 

src/persist/src/indexed/columnar/parquet.rs

@@ -118,7 +118,7 @@ pub fn encode_parquet_kvtd<W: Write + Send>(
         .set_compression(cfg.compression.into())
         .set_writer_version(WriterVersion::PARQUET_2_0)
         .set_data_page_size_limit(1024 * 1024)
-        .set_max_row_group_size(usize::MAX)
+        .set_max_row_group_row_count(None)
         .set_key_value_metadata(Some(vec![metadata]))
         .build();
 

src/storage-operators/src/s3_oneshot_sink/parquet.rs

@@ -287,7 +287,7 @@ impl ParquetFile {
         let props = WriterProperties::builder()
             // This refers to the number of rows per row-group, which we don't want the writer
             // to enforce since we will flush based on the byte-size of the active row group
-            .set_max_row_group_size(usize::MAX)
+            .set_max_row_group_row_count(None)
             // Max compatibility
             .set_writer_version(WriterVersion::PARQUET_1_0)
             .set_compression(Compression::SNAPPY)

src/storage-types/Cargo.toml

@@ -65,6 +65,7 @@ proptest-derive = { workspace = true, optional = true }
 prost.workspace = true
 rdkafka.workspace = true
 regex.workspace = true
+reqsign-core.workspace = true
 reqwest.workspace = true
 serde.workspace = true
 serde_json = { workspace = true, features = ["preserve_order"] }

src/storage-types/src/connections.rs

@@ -34,7 +34,7 @@ use iceberg_catalog_rest::{
     REST_CATALOG_PROP_URI, REST_CATALOG_PROP_WAREHOUSE, RequestAuthenticator, RestCatalogBuilder,
 };
 use iceberg_storage_opendal::{
-    AwsCredential, AwsCredentialLoad, CustomAwsCredentialLoader, OpenDalStorageFactory,
+    AwsCredential, CustomAwsCredentialLoader, OpenDalStorageFactory, ProvideCredential,
 };
 use itertools::Itertools;
 use mz_ccsr::tls::{Certificate, Identity};
@@ -99,6 +99,7 @@ const REST_CATALOG_PROP_CREDENTIAL: &str = "credential";
 /// We use this instead of OpenDAL's built-in assume role support because Materialize
 /// has a runtime-defined credential chain (ambient → jump role → user role with external ID)
 /// that can't be expressed via OpenDAL's static configuration properties.
+#[derive(Debug)]
 struct AwsSdkCredentialLoader {
     /// The underlying AWS SDK credentials provider. For assume role auth, this provider
     /// already handles the full chain: ambient creds -> jump role -> user role.
@@ -111,37 +112,56 @@ impl AwsSdkCredentialLoader {
     }
 }
 
-#[async_trait]
-impl AwsCredentialLoad for AwsSdkCredentialLoader {
-    async fn load_credential(
+impl ProvideCredential for AwsSdkCredentialLoader {
+    type Credential = AwsCredential;
+
+    async fn provide_credential(
         &self,
-        _client: reqwest::Client,
-    ) -> anyhow::Result<Option<AwsCredential>> {
-        let creds = self
-            .provider
-            .provide_credentials()
-            .await
-            .map_err(|e| {
-                warn!(
-                    error = %e.display_with_causes(),
-                    "failed to load AWS credentials for Iceberg FileIO from SDK provider"
-                );
-                e
-            })
-            .context(
+        _ctx: &reqsign_core::Context,
+    ) -> reqsign_core::Result<Option<AwsCredential>> {
+        let creds = self.provider.provide_credentials().await.map_err(|e| {
+            warn!(
+                error = %e.display_with_causes(),
+                "failed to load AWS credentials for Iceberg FileIO from SDK provider"
+            );
+            reqsign_core::Error::credential_invalid(
                 "failed to load AWS credentials from SDK provider for Iceberg FileIO \
                  (credential source may be temporarily unavailable)",
-            )?;
+            )
+            .with_source(e)
+        })?;
+
+        // `expires_in` is what tells reqsign to ask us for fresh credentials, so a failed
+        // conversion has to be an error. Reporting `None` would pin the signer to credentials
+        // that stop working once the underlying role session expires.
+        let expires_in = match creds.expiry() {
+            Some(expiry) => Some(system_time_to_timestamp(expiry)?),
+            None => None,
+        };
 
         Ok(Some(AwsCredential {
             access_key_id: creds.access_key_id().to_string(),
             secret_access_key: creds.secret_access_key().to_string(),
             session_token: creds.session_token().map(|s| s.to_string()),
-            expires_in: creds.expiry().map(|t| t.into()),
+            expires_in,
         }))
     }
 }
 
+/// Converts a `SystemTime` into reqsign's timestamp type, which has no `From` impl for it.
+fn system_time_to_timestamp(t: SystemTime) -> reqsign_core::Result<reqsign_core::time::Timestamp> {
+    let millis = t
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .map_err(|e| {
+            reqsign_core::Error::unexpected("timestamp precedes the Unix epoch").with_source(e)
+        })?
+        .as_millis();
+    let millis = i64::try_from(millis).map_err(|e| {
+        reqsign_core::Error::unexpected("timestamp too far in the future").with_source(e)
+    })?;
+    reqsign_core::time::Timestamp::from_millisecond(millis)
+}
+
 /// Signs each outgoing REST-catalog request with AWS SigV4.
 ///
 /// Holds a [`SharedCredentialsProvider`] (not static `Credentials`) so each
@@ -865,15 +885,14 @@ impl IcebergCatalogConnection<InlinedConnection> {
         // N.B. We're using the AWS credentials from the catalog connection for the storage layer
         //   even though the sink comes with its own (unused) AWS credentials for storage.
         let customized_credential_load = if matches!(aws_auth, AwsAuth::AssumeRole(_)) {
-            Some(CustomAwsCredentialLoader::new(Arc::new(
-                AwsSdkCredentialLoader::new(credentials_provider),
+            Some(CustomAwsCredentialLoader::new(AwsSdkCredentialLoader::new(
+                credentials_provider,
             )))
         } else {
             None
         };
 
         let storage_factory = Arc::new(OpenDalStorageFactory::S3 {
-            configured_scheme: "s3".to_string(),
             customized_credential_load,
         });
 
@@ -927,7 +946,6 @@ impl IcebergCatalogConnection<InlinedConnection> {
                 }
                 (
                     OpenDalStorageFactory::S3 {
-                        configured_scheme: "s3".to_string(),
                         // When used with MinIO, Polaris returns a config with:
                         //   s3.access-key-id, s3.secret-access-key, s3.endpoint, ...
                         // `iceberg-rust` forwards these props to `opendal`.