SS-369 Support vended credentials for iceberg sinks

Cargo.toml

@@ -278,8 +278,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"] }
@@ -384,9 +384,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"] }
@@ -434,7 +434,7 @@ opentelemetry_sdk = { version = "0.32.1", 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 = "4.0.0"
 phf = { version = "0.13.1", features = ["uncased"] }
@@ -473,6 +473,10 @@ 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 semver-compatible with the `reqsign-core` that `iceberg-storage-opendal` links
+# against, otherwise our `ProvideCredential` impl targets a different trait than the one it
+# expects and the credential loader stops type-checking.
+reqsign-core = "3.3.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"
@@ -667,10 +671,13 @@ 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 `vended-creds` branch.
+# NOTE: The `[workspace.dependencies]` version requirement above must stay
+# semver-compatible with this revision's crate version, otherwise Cargo drops
+# these patches into `[[patch.unused]]` and silently builds against crates.io.
+iceberg = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "016a2a1c1a8342a721d13aa870334cd5459076fd" }
+iceberg-catalog-rest = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "016a2a1c1a8342a721d13aa870334cd5459076fd" }
+iceberg-storage-opendal = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "016a2a1c1a8342a721d13aa870334cd5459076fd" }
 
 # Custom duckdb crate to support mz needs
 # All changes should go to the `mz_changes` branch.

deny.toml

@@ -102,10 +102,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" },
+    # reqsign-aws (via iceberg) is ahead of opendal, which pulls 0.39.
+    { name = "quick-xml", version = "0.41.0" },
     # Used by sentry
     { name = "reqwest", version = "0.13.3" },
     # Conflicts between `bon` in apache-avro and `derive_builder` in iceberg
@@ -126,15 +124,41 @@ skip = [
     { name = "fallible-iterator", version = "0.3.0" },
     # arrow
     { name = "hashbrown", version = "0.16.1" },
+    { name = "hashbrown", version = "0.17.1" },
+    # duckdb pins arrow 57 while the workspace is on 58, so mz-testdrive links
+    # both majors. They never meet: duckdb's arrow stays inside duckdb's own API.
+    { 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" },
+    # opendal 0.57 and reqsign 3.x (both via iceberg) are on the next generation
+    # of the RustCrypto crates, while the rest of the workspace is still on the
+    # current one.
+    { name = "base64", version = "0.23.1" },
+    { name = "block-buffer", version = "0.12.1" },
+    { name = "const-oid", version = "0.10.2" },
+    { name = "crypto-common", version = "0.2.2" },
+    { name = "digest", version = "0.11.3" },
+    { name = "hmac", version = "0.13.0" },
+    { name = "md-5", version = "0.11.0" },
+    { name = "sha1", version = "0.11.0" },
+    { name = "sha2", version = "0.11.0" },
+    # Held back by mz-timely-util; iceberg and the persist crates use 0.13.
+    { name = "lz4_flex", version = "0.12.1" },
     # Used by dynfmt; iceberg/typetag pulls in v0.4.
     { name = "erased-serde", version = "0.3.26" },
     # gcp_auth → hyper-rustls → rustls-native-certs pulls newer versions
     # while native-tls still pulls older versions.
     { name = "core-foundation", version = "0.10.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" },
+    { name = "security-framework", version = "3.7.0" },
+    { name = "openssl-probe", version = "0.2.1" },
     # 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).
@@ -211,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",

misc/python/materialize/mzcompose/helpers/iceberg.py

@@ -98,21 +98,36 @@ def create_polaris_catalog(
     secret_key: str = "",
     endpoint: str = "http://minio:9000",
     region: str = "minio",
+    static_credentials: bool = True,
 ) -> None:
+    """Create a Polaris catalog backed by `bucket_name` in MinIO.
+
+    Catalog properties are returned to clients verbatim on `loadTable`, so the
+    `s3.access-key-id`/`s3.secret-access-key` written here become the credentials
+    every client uses. Pass `static_credentials=False` to leave them out, which
+    makes credential vending the only way a client can reach the bucket. Polaris
+    itself still reaches MinIO through the credentials in its environment.
+    """
+    properties = {
+        "default-base-location": f"s3://{bucket_name}/",
+        "s3.endpoint": endpoint,
+        "s3.path-style-access": "true",
+        "s3.region": region,
+    }
+    if static_credentials:
+        properties["s3.access-key-id"] = username
+        properties["s3.secret-access-key"] = secret_key
+
     catalog_payload = {
         "name": catalog_name,
         "type": "INTERNAL",
-        "properties": {
-            "default-base-location": f"s3://{bucket_name}/",
-            "s3.endpoint": endpoint,
-            "s3.path-style-access": "true",
-            "s3.access-key-id": username,
-            "s3.secret-access-key": secret_key,
-            "s3.region": region,
-        },
+        "properties": properties,
         "storageConfigInfo": {
             "storageType": "S3",
-            "allowedLocations": [f"s3://{bucket_name}/*"],
+            # Allowed locations are prefixes, not globs. Polaris rejects the
+            # catalog unless `default-base-location` sits within one of them, and
+            # a trailing `/*` is matched literally, so it contains nothing.
+            "allowedLocations": [f"s3://{bucket_name}/"],
             "endpoint": endpoint,
             "endpointInternal": endpoint,
             "pathStyleAccess": True,
@@ -124,6 +139,7 @@ def create_polaris_catalog(
         "curl",
         "-sS",
         "-i",
+        "--fail-with-body",
         "-X",
         "POST",
         "-H",
@@ -136,6 +152,30 @@ def create_polaris_catalog(
     )
 
 
+def assert_polaris_catalog_exists(
+    c: "Composition",
+    access_token: str,
+    catalog_name: str = "default_catalog",
+) -> None:
+    """Read the catalog back, failing here if it is missing.
+
+    A rejected catalog creation otherwise stays invisible until whatever first
+    uses the warehouse reports a 404, which points the investigation at the
+    consumer instead of at setup.
+    """
+    c.exec(
+        "polaris",
+        "curl",
+        "-sS",
+        "--fail-with-body",
+        "-X",
+        "GET",
+        "-H",
+        f"Authorization: Bearer {access_token}",
+        f"http://localhost:8181/api/management/v1/catalogs/{catalog_name}",
+    )
+
+
 def create_polaris_namespace(
     c: "Composition",
     access_token: str,
@@ -148,6 +188,7 @@ def create_polaris_namespace(
         "curl",
         "-sS",
         "-i",
+        "--fail-with-body",
         "-X",
         "POST",
         "-H",
@@ -160,13 +201,71 @@ def create_polaris_namespace(
     )
 
 
+def grant_catalog_role_privilege(
+    c: "Composition",
+    access_token: str,
+    privilege: str,
+    catalog_name: str = "default_catalog",
+    catalog_role: str = "catalog_admin",
+) -> None:
+    """Grant a catalog-level privilege to a catalog role."""
+    c.exec(
+        "polaris",
+        "curl",
+        "-sS",
+        "--fail-with-body",
+        "-X",
+        "PUT",
+        "-H",
+        f"Authorization: Bearer {access_token}",
+        "-H",
+        "Content-Type: application/json",
+        f"http://localhost:8181/api/management/v1/catalogs/{catalog_name}/catalog-roles/{catalog_role}/grants",
+        "-d",
+        json.dumps({"type": "catalog", "privilege": privilege}),
+    )
+
+
+def load_polaris_vended_credentials(
+    c: "Composition",
+    table: str,
+    namespace: str = "default_namespace",
+    catalog_name: str = "default_catalog",
+) -> dict[str, str]:
+    """Load a table through the REST catalog requesting credential vending, and
+    return the vended storage config (the `config` map, which contains the
+    temporary `s3.access-key-id`, `s3.secret-access-key`, and `s3.session-token`).
+
+    Requires the catalog to have been set up with `vended=True` so the principal
+    is authorized for `LOAD_TABLE_WITH_READ_DELEGATION`.
+    """
+    access_token = get_polaris_access_token(c)
+    resp = c.exec(
+        "polaris",
+        "curl",
+        "-sS",
+        "--fail-with-body",
+        "-X",
+        "GET",
+        "-H",
+        f"Authorization: Bearer {access_token}",
+        "-H",
+        "X-Iceberg-Access-Delegation: vended-credentials",
+        f"http://localhost:8181/api/catalog/v1/{catalog_name}/namespaces/{namespace}/tables/{table}",
+        capture=True,
+    )
+    return json.loads(resp.stdout)["config"]
+
+
 def setup_polaris_for_iceberg(
     c: "Composition",
     bucket_name: str = "test-bucket",
     minio_alias: str = "s3test",
     username: str = "tduser",
     catalog_name: str = "default_catalog",
     namespace: str = "default_namespace",
+    vended: bool = False,
+    static_credentials: bool = True,
 ) -> tuple[str, str]:
     """
     Set up Polaris catalog with MinIO for Iceberg sink usage.
@@ -176,6 +275,17 @@ def setup_polaris_for_iceberg(
     2. Creating a MinIO user with S3 permissions
     3. Starting Polaris with the user's credentials
     4. Creating a catalog and namespace in Polaris
+
+    With `vended=True`, also grant the catalog role the `TABLE_READ_DATA` and
+    `TABLE_WRITE_DATA` privileges. These authorize credential vending, so a
+    client sending the `X-Iceberg-Access-Delegation: vended-credentials` header
+    on `loadTable`/`commit` receives temporary, table-scoped MinIO STS
+    credentials instead of using its own static ones. Polaris mints them via
+    AssumeRole against MinIO using the credentials passed to it below.
+
+    With `static_credentials=False`, the catalog withholds the long-lived S3
+    credentials it would otherwise hand every client, so vending becomes the only
+    path to the bucket. Combine with `vended=True` to require vending.
     """
     from materialize.mzcompose.composition import Service
 
@@ -211,13 +321,22 @@ def setup_polaris_for_iceberg(
         bucket_name=bucket_name,
         username=username,
         secret_key=key,
+        static_credentials=static_credentials,
     )
 
+    assert_polaris_catalog_exists(c, access_token, catalog_name=catalog_name)
+
     create_polaris_namespace(
         c,
         access_token,
         namespace=namespace,
         catalog_name=catalog_name,
     )
 
+    if vended:
+        for privilege in ("TABLE_READ_DATA", "TABLE_WRITE_DATA"):
+            grant_catalog_role_privilege(
+                c, access_token, privilege, catalog_name=catalog_name
+            )
+
     return (username, key)

misc/python/materialize/mzcompose/services/polaris.py

@@ -18,7 +18,7 @@ def __init__(
         self,
         name: str = "polaris-bootstrap",
         image: str = "apache/polaris-admin-tool",
-        tag: str = "1.2.0-incubating",
+        tag: str = "1.7.0",
         environment: list[str] = [
             "POLARIS_BOOTSTRAP_CREDENTIALS=POLARIS,root,root",
             "POLARIS_PERSISTENCE_TYPE=relational-jdbc",
@@ -50,8 +50,9 @@ def __init__(
         self,
         name: str = "polaris",
         image: str = "apache/polaris",
-        # Fails with 1.1.0-incubating
-        tag: str = "1.2.0-incubating",
+        # Fails with 1.1.0-incubating. Releases from 1.4.0 on drop the
+        # `-incubating` suffix, so keep this tag in sync with the admin tool's.
+        tag: str = "1.7.0",
         # 8181: api port, 8182: management port
         ports: list[str | int] = [8181, 8182],
         environment: list[str] = [

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/sql-lexer/src/keywords.txt

@@ -144,6 +144,7 @@ Decoding
 Decorrelated
 Default
 Defaults
+Delegation
 Delete
 Delimited
 Delimiter