SS-369 Support vended credentials for iceberg sinks

[patrickwwbutler]

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.

[ublubu]

It looks like you recreated a bunch of #38084

Was that intentional?

[def-]

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]

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.
   

   materialize/src/storage/src/sink/iceberg.rs

   Line 1705 in ab9a4d0

   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.

materialize/src/storage/src/sink/iceberg.rs

Line 1530 in ab9a4d0

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

---

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

materialize/src/storage/src/sink/iceberg.rs

Lines 283 to 289 in ab9a4d0

	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(),
	);

materialize/src/storage/src/sink/iceberg.rs

Line 1651 in ab9a4d0

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

---

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

materialize/src/storage/src/sink/iceberg.rs

Lines 1830 to 1834 in ab9a4d0

	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)>,

materialize/src/storage/src/sink/iceberg.rs

Lines 1705 to 1707 in ab9a4d0

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