S3-compatible storage and Iceberg REST catalogs

This guide walks you through setting up an Iceberg sink to a catalog that implements the Iceberg REST catalog specification, with tables stored in Amazon S3 or S3-compatible object storage. The examples use Apache Polaris as the catalog and MinIO as the object store.

Materialize does not take a storage endpoint or storage credentials of its own. The catalog supplies both: with credential vending, it hands Materialize temporary credentials scoped to the table it writes.

Prerequisites

• A Polaris deployment, and an S3-compatible bucket for its warehouse. The examples use a bucket named warehouse.

• Polaris configured with storage credentials that can call the object store’s STS AssumeRole API, for example through the AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables.

• Polaris’s API and the object store reachable from Materialize. Iceberg catalog connections cannot tunnel through AWS PrivateLink or an SSH bastion host. If either endpoint sits behind a firewall, allow traffic from the static egress IP addresses of your Materialize region.

• An access token for a Polaris principal that can manage catalogs and principals, such as the root principal created at bootstrap:

  POLARIS=https://<polaris_host>
  read -rs ROOT_CLIENT_SECRET
  
  TOKEN=$(curl -sS "$POLARIS/api/catalog/v1/oauth/tokens" \
    -d grant_type=client_credentials \
    -d client_id=<root_client_id> \
    --data-urlencode "client_secret=$ROOT_CLIENT_SECRET" \
    -d scope=PRINCIPAL_ROLE:ALL | jq -r .access_token)
  

Set up Polaris

Step 1. Create a catalog backed by your bucket

Create a catalog whose storageConfigInfo points at your object store. Polaris vends endpoint to Materialize, so it must be reachable from Materialize. If Polaris reaches the object store at a different address, add that address as endpointInternal. S3-compatible stores such as MinIO generally need pathStyleAccess.

curl -sS --fail-with-body "$POLARIS/api/management/v1/catalogs" \
  -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \
  -d '{
    "catalog": {
      "name": "materialize_catalog",
      "type": "INTERNAL",
      "properties": {
        "default-base-location": "s3://warehouse/materialize_catalog/"
      },
      "storageConfigInfo": {
        "storageType": "S3",
        "allowedLocations": ["s3://warehouse/materialize_catalog/"],
        "endpoint": "https://<minio_host>:9000",
        "pathStyleAccess": true,
        "region": "us-east-1"
      }
    }
  }'

Leave s3.access-key-id and s3.secret-access-key out of the catalog’s properties. Polaris returns catalog properties to every client, so long-lived credentials placed there are shared with anyone who can load a table.

Step 2. Create a principal for Materialize

Create a principal, and save the clientId and clientSecret that Polaris returns. Polaris does not show the secret again.

curl -sS --fail-with-body "$POLARIS/api/management/v1/principals" \
  -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \
  -d '{"principal": {"name": "materialize"}}' | jq .credentials

Step 3. Grant the principal access to the catalog

Polaris grants privileges to catalog roles, which are assigned to principal roles, which are assigned to principals. CATALOG_MANAGE_CONTENT lets the principal create and commit to tables, and includes the TABLE_READ_DATA and TABLE_WRITE_DATA privileges that authorize credential vending.

H=(-H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json")
API="$POLARIS/api/management/v1"

# Create a principal role and assign it to the principal.
curl -sS --fail-with-body "${H[@]}" "$API/principal-roles" \
  -d '{"principalRole": {"name": "materialize_role"}}'
curl -sS --fail-with-body "${H[@]}" -X PUT "$API/principals/materialize/principal-roles" \
  -d '{"principalRole": {"name": "materialize_role"}}'

# Create a catalog role, grant it privileges, and assign it to the principal role.
curl -sS --fail-with-body "${H[@]}" "$API/catalogs/materialize_catalog/catalog-roles" \
  -d '{"catalogRole": {"name": "materialize_writer"}}'
curl -sS --fail-with-body "${H[@]}" -X PUT \
  "$API/catalogs/materialize_catalog/catalog-roles/materialize_writer/grants" \
  -d '{"grant": {"type": "catalog", "privilege": "CATALOG_MANAGE_CONTENT"}}'
curl -sS --fail-with-body "${H[@]}" -X PUT \
  "$API/principal-roles/materialize_role/catalog-roles/materialize_catalog" \
  -d '{"catalogRole": {"name": "materialize_writer"}}'

Step 4. Create a namespace

Materialize creates the table a sink writes to, but not its namespace.

curl -sS --fail-with-body "$POLARIS/api/catalog/v1/materialize_catalog/namespaces" \
  -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \
  -d '{"namespace": ["materialize"]}'

Create the Iceberg catalog connection in Materialize

Step 1. Store the principal’s credentials

Store the clientId and clientSecret from Step 2 as a single value, separated by a colon:

CREATE SECRET polaris_credential AS '<client_id>:<client_secret>';

Step 2. Create the connection

CREATE CONNECTION iceberg_catalog_connection TO ICEBERG CATALOG (
    CATALOG TYPE = 'rest',
    URL = 'https://<polaris_host>/api/catalog',
    WAREHOUSE = 'materialize_catalog',
    CREDENTIAL = SECRET polaris_credential,
    SCOPE = 'PRINCIPAL_ROLE:ALL',
    ACCESS DELEGATION = 'vended-credentials'
);

CREATE CONNECTION validates the connection by listing the catalog’s namespaces, so a wrong URL, scope, or credential fails here. For all options, see CREATE CONNECTION.

Other REST catalogs

The same connection works with other Iceberg REST catalogs, with these differences:

• Without ACCESS DELEGATION, Materialize uses the storage properties the catalog returns: s3.endpoint, s3.region, s3.path-style-access, s3.access-key-id, and s3.secret-access-key. Materialize adds nothing to them, so they must describe the store completely.
• Materialize exchanges CREDENTIAL at <URL>/v1/oauth/tokens and ignores the token endpoint the catalog advertises. Set OAUTH2 SERVER URL if the catalog serves it elsewhere.
• Materialize refreshes vended credentials through the catalog’s loadCredentials endpoint. Sinks on a catalog that vends credentials but does not implement loadCredentials fail once the first credential expires.
• For storage other than S3, set STORAGE PROVIDER.

Create the Iceberg sink in Materialize

In Materialize, you can sink from a materialized view, table, or source. Use CREATE SINK to create an Iceberg sink, replacing:

• <sink_name> with a name for your sink.
• <sink_cluster> with the name of your sink cluster.
• <my_materialize_object> with the name of your materialized view, table, or source.
• <my_iceberg_namespace> with your catalog namespace.
• <my_iceberg_table> with the name of your Iceberg table. If the Iceberg table does not exist, Materialize creates the table. For details, see CREATE SINK reference page.
• <commit_interval> with your commit interval (e.g., 1m). The commit interval specifies how frequently Materialize commits snapshots to Iceberg. The minimum commit interval is 1s. See Commit interval tradeoffs below.

For the full list of syntax options, see the CREATE SINK reference.

Upsert mode

In upsert mode, replace <key> with the column(s) that uniquely identify rows.

CREATE SINK <sink_name>
  IN CLUSTER <sink_cluster>
  FROM <my_materialize_object>
  INTO ICEBERG CATALOG CONNECTION iceberg_catalog_connection (
    NAMESPACE = '<my_iceberg_namespace>',
    TABLE = '<my_iceberg_table>'
  )
  KEY (<key>)
  MODE UPSERT
  WITH (COMMIT INTERVAL = '<commit_interval>');

Append mode

In append mode, no KEY clause is used. The Iceberg table includes all source columns plus _mz_diff (int) and _mz_timestamp (long).

CREATE SINK <sink_name>
  IN CLUSTER <sink_cluster>
  FROM <my_materialize_object>
  INTO ICEBERG CATALOG CONNECTION iceberg_catalog_connection (
    NAMESPACE = '<my_iceberg_namespace>',
    TABLE = '<my_iceberg_table>'
  )
  MODE APPEND
  WITH (COMMIT INTERVAL = '<commit_interval>');

Verify the sink

Once the first commit interval elapses, check that the sink is running:

SELECT name, status, error
FROM mz_internal.mz_sink_statuses
WHERE name = '<sink_name>';

    name     | status  | error
-------------+---------+-------
 <sink_name> | running |

The table’s data and metadata files appear in your bucket under s3://warehouse/materialize_catalog/<namespace>/<table>/.

Considerations

Commit interval tradeoffs

The COMMIT INTERVAL setting controls how frequently Materialize commits snapshots to your Iceberg table, making the data available to downstream query engines. This setting involves tradeoffs:

Recommendations:

• For production, use intervals of 1m or longer
• For batch analytics, use longer intervals (5m to 15m)

Starting in v26.34, you can change the commit interval of an existing sink with ALTER SINK.

Exactly-once delivery

Iceberg sinks provide exactly-once delivery. After a restart, Materialize resumes from the last committed snapshot without duplicating data.

Materialize stores progress information in Iceberg snapshot metadata properties (mz-frontier and mz-sink-version).

Credential refresh

Materialize refreshes both its OAuth2 token and the vended storage credentials while the sink runs. Rotating the principal’s secret in Polaris requires updating the Materialize secret:

ALTER SECRET polaris_credential AS '<client_id>:<new_client_secret>';

Type mapping

Materialize converts SQL types to Iceberg/Parquet types:

Limitations

• Iceberg catalog connections cannot be modified with ALTER CONNECTION. To change any option, drop the connection and the sinks that depend on it, then recreate them.

• You cannot create an Iceberg sink into an existing Iceberg table. Materialize creates and manages the target Iceberg table itself, so the table named by the sink must not already exist.
• Partitioned tables are not supported.
• Schema evolution of an Iceberg table is not supported. If the SINK FROM object’s schema changes, you must drop and recreate the sink.

Troubleshooting

Sink creation fails with “input compacted past resume upper”

This error occurs when the source data has been compacted beyond the point where the sink last committed. This can happen after a Materialize backup/restore operation. You may need to drop and recreate the sink, which will re-snapshot the entire source relation.

Commit conflicts

If another process modifies the Iceberg table while Materialize is committing, you may see commit conflict errors. Materialize will automatically retry, but if conflicts persist, ensure no other writers are modifying the same table.

Sink falls behind and does not catch up

An Iceberg sink can fall behind its input when each commit to the catalog takes longer than the sink’s COMMIT INTERVAL. In steady state, the sink commits one snapshot per interval of data. If a commit takes longer than the interval, the sink falls further behind with every commit, and its lag keeps growing even though the input relation is up to date. How long a commit takes depends on your catalog and object store, and can be several seconds.

To check whether the sink is behind its input, compare their lag:

SELECT o.name, l.lag
FROM mz_internal.mz_wallclock_global_lag AS l
JOIN mz_catalog.mz_objects AS o ON o.id = l.object_id
WHERE o.name IN ('<sink_name>', '<input_relation_name>');

If the input’s lag is low but the sink’s lag is high and growing, recover the sink in one of the following ways:

• Increase the commit interval to a value that comfortably exceeds the time a commit takes. Outside of development environments, use at least 1m. See Commit interval tradeoffs.

  ALTER SINK <sink_name> SET (COMMIT INTERVAL = '1m');
  

  Changing the commit interval restarts the sink.

• If the commit interval is already appropriate, restart the sink by restarting the replicas of the cluster that hosts it. Restarting the replicas also restarts every other object on that cluster, which must then rehydrate.

  For a managed cluster:

  ALTER CLUSTER <cluster_name> SET (REPLICATION FACTOR = 0);
  ALTER CLUSTER <cluster_name> SET (REPLICATION FACTOR = <original replication factor>);
  

  For an unmanaged cluster, drop and recreate each replica:

  DROP CLUSTER REPLICA <cluster_name>.<replica_name>;
  CREATE CLUSTER REPLICA <cluster_name>.<replica_name> (SIZE = '<original size>');
  

When the sink restarts, it writes all data between its last committed point and the current state of its input in a single catch-up commit, then resumes committing at the configured interval. Restarting the sink without increasing a commit interval that is too short only recovers temporarily, and the sink falls behind again.

Related pages

• CREATE SINK
• CREATE CONNECTION
• Storage access delegation
• Apache Polaris documentation