Skip to content

SS-369 Support vended credentials for iceberg sinks - #38181

Open
patrickwwbutler wants to merge 11 commits into
MaterializeInc:patrick/iceberg-unity-catalogfrom
patrickwwbutler:patrick/iceberg-vended
Open

SS-369 Support vended credentials for iceberg sinks#38181
patrickwwbutler wants to merge 11 commits into
MaterializeInc:patrick/iceberg-unity-catalogfrom
patrickwwbutler:patrick/iceberg-vended

Conversation

@patrickwwbutler

@patrickwwbutler patrickwwbutler commented Aug 12, 2026

Copy link
Copy Markdown
Contributor

Stacked on #38373 (sql: Add OAUTH2 SERVER URL option to Iceberg catalog connections). This PR's base is patrick/iceberg-unity-catalog, so the diff shown here is only the vended-credentials layer. Merge #38373 first. GitHub Stacks cannot include PRs from forks, so the two are linked by base branch rather than by a stack object.

Adds MZ-side support for vended creds by adding a syntax option called ACCESS DELEGATION to the ICEBERG CATALOG CONNECTION that controls the header requesting vended credentials. This also points our iceberg crates at a feature branch at https://github.com/MaterializeInc/iceberg-rust/tree/vended-creds, so we will need to merge that to our iceberg release branch (and likely update Cargo.toml here) before actually merging this.

@patrickwwbutler
patrickwwbutler marked this pull request as ready for review August 18, 2026 13:01
@patrickwwbutler
patrickwwbutler requested review from a team as code owners August 18, 2026 13:01
@patrickwwbutler patrickwwbutler changed the title Support vended credentials for iceberg sinks SS-369 Support vended credentials for iceberg sinks Aug 18, 2026

@ublubu ublubu left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It looks like you recreated a bunch of #38084

Was that intentional?

@def-

def- commented Aug 20, 2026

Copy link
Copy Markdown
Contributor

QA LLM Review

1. MEDIUM -- Vended storage credentials are captured once, so the sink stalls when they expire

src/storage/src/sink/iceberg.rs:1572

The write_data_files operator clones table.file_io() once, right after its
single catalog.load_table() at operator startup, and every Parquet data file
the sink ever writes goes through that clone. With ACCESS DELEGATION = 'vended-credentials' that FileIO holds the temporary
s3.access-key-id/s3.secret-access-key/s3.session-token the catalog minted
at that instant, and nothing refreshes them, so every sink using the new option
stalls and restarts once per vending window (commonly one hour, as little as 15
minutes for STS-minted sessions).

Details

Why they never refresh. The catalog's storage-credentials arrive as plain
s3.* properties (crates/storage/opendal/src/s3.rs: s3_config_parse) and are
baked into a static S3Config. For the REST path
customized_credential_load is None (src/storage-types/src/connections.rs,
the IcebergCatalogAuth::OAuth arm), so no ProvideCredential is installed and
reqsign's Signer caches a credential whose expires_in is None — which, as
this PR's own comment at connections.rs:141 notes, reqsign reads as "valid
forever". The AssumeRole/s3tables path avoids this precisely because it does
install AwsSdkCredentialLoader.

Only the data-file writes go stale. The commit operator is fine: in the
pinned iceberg-rust rev, Transaction::do_commit re-loads the table and
grafts the refreshed FileIO onto the committed table, so manifest writes pick
up freshly vended credentials on every commit. The writer operator has no
equivalent, and it is where the bulk of the write traffic goes.

Effect. Once the token expires, S3 returns ExpiredToken, the fallible
operator emits HealthStatusUpdate::halting, and the health operator issues a
SuspendAndRestart for the sink. The restart reloads the table and gets fresh
credentials, so it self-heals — but every sink using this option repeats the
stall/error/restart cycle indefinitely, with an error surfaced in
mz_sink_statuses each time and any uncommitted batch redone. The new
vended-credentials.td runs for ~10s, well inside any vending window, so it
cannot observe this.

Fix. Mirror the AssumeRole path: give the REST/vended case a
ProvideCredential impl that re-vends (a load_table against the catalog) and
reports the vended expiry as expires_in, so reqsign's is_valid() check
drives the refresh. A narrower alternative is to reload the table and rebuild
the writer's FileIO when a data-file write fails authentication, though that
still costs one failed batch per window.

@ublubu

ublubu commented Aug 20, 2026

Copy link
Copy Markdown
Contributor

I can think of two main approaches to the vended credentials refresh problem.

  1. Build the vended credentials refresh into the storage factory's credentials provider.
    For example, OpenDalStorageFactory::S3 supports a customized_credential_load. (GCS does not, so we'd have to add that capability. Maybe we can write a generic vended-credentials wrapper for storage factories.) Our credential loader would call the Catalog's load_table to refresh the vended credentials.

  2. (Re)Build the Iceberg writer inside the write_rows closure if the vended credentials will expire soon.

    let write_rows = async |rows: &OrdValBatch<_>,

1 feels wrong because the Storage Factory is for the Catalog, not the Table. But the vended credentials loader needs to know what table it's requesting credentials for. (And how does it work with prefix-specific credentials?)

2 feels wrong because our Sink code has to look inside the Table properties for the vended credentials (expiry). And creating a new writer throws out any state the old writer had--just to replace the FileIO credentials.

  • Resetting the writer state might be okay. Snapshots are treated as a sequence of inserts, so we don't need to remember rows to handle deletions (with_max_seen_rows). And snapshots should be the only batches large enough to exceed the 1hr expiry time for vended credentials.

Notes from digging through the code:

Currently, we get a FileIO instance one time in the life of the sink.
When we build the write_data_files operator, we load_table.

let table = catalog.load_table(&table_ident).await.with_context(|| {


The table's FileIO goes into the writer we package alongside each ready batch.

let data_rolling_writer = RollingFileWriterBuilder::new_with_default_file_size(
data_parquet_writer,
Arc::clone(&self.ctx.current_schema),
self.ctx.file_io.clone(),
self.ctx.location_generator.clone(),
self.ctx.file_name_generator.clone(),
);

in_flight_batches.push_back((batch_desc.clone(), batch_writer));


And we write the batch using the writer we built in the previous step.

async fn with_ready_batches<L: Layout, W, Write, Close>(
input_frontier: Antichain<Timestamp>,
input_batches: &mut VecDeque<Rc<OrdValBatch<L>>>,
output_frontier: Antichain<Timestamp>,
output_batches: &mut VecDeque<(BatchDescription, W)>,

let write_rows = async |rows: &OrdValBatch<_>,
(lower, upper): BatchDescription,
batch_writer: &mut Box<dyn IcebergWriter>|

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants