storage: Refresh catalog-vended storage credentials

Vended credentials expire, and nothing refreshed them: `loadTable` hands
their access keys to the FileIO once and OpenDAL keeps signing with them
until the dataflow restarts.

Adds `VendedCredentialLoader`, a `ProvideCredential` implementation that
re-fetches from the catalog's `loadCredentials` endpoint. OpenDAL rebuilds
its `Operator` for every file operation, so reqsign's own credential cache
never survives one call; the loader therefore caches with its own expiry
deadline, taken from `s3.session-token-expires-at-ms` where the catalog
reports one and a short interval where it does not. On a 401 or 403 it
invalidates the catalog token so the next attempt mints a fresh one.

Materialize now builds the OAuth2 provider itself rather than passing a
`credential` catalog property, so one token object serves both catalog
requests and credential refreshes. The catalog client rejects a custom
authenticator combined with that property, so the two cannot coexist.

`connect` grows a table argument because the credentials endpoint is
table-scoped. Installing a loader also means it alone supplies S3
credentials, since OpenDAL replaces its whole provider chain, so one is
installed only when the connection asked for delegation.

Author: patrickwwbutler

src/sql/src/pure.rs

@@ -625,8 +625,10 @@ async fn purify_create_sink(
             // Now that we've validated the sink's storage creds (if they exist)
             // we _could_ use them to build a complete Iceberg client (both catalog and storage).
             // TODO(kynan): Actually use those sink-specific creds here instead of ignoring them.
+            // Purification only proves the catalog is reachable, so it needs no table-scoped
+            // storage credentials.
             let _catalog = connection
-                .connect(storage_configuration, InTask::No)
+                .connect(storage_configuration, InTask::No, None)
                 .await
                 .map_err(|e| IcebergSinkPurificationError::CatalogError(Arc::new(e)))?;
         }