AWS S3 Tables

This guide walks you through the steps required to set up Iceberg sinks in Materialize Cloud.

Prerequisites

• An AWS account with permissions to create and manage IAM policies and roles.
• An AWS S3 Table bucket in your AWS account. The S3 Table bucket must be in the same AWS region as your Materialize deployment.
• A namespace in the AWS S3 Table bucket. For details on creating namespaces, see AWS S3 documentation: Creating a namespace.

Create the Iceberg catalog connection in Materialize

Step 1. Set up permissions in AWS

In AWS, set up permissions to allow Materialize to write data files to the object storage backing your Iceberg catalog. This tutorial uses an IAM policy and IAM role to grant the required permissions. We strongly recommend using role assumption-based authentication to manage access.

Step 1A. Create an IAM policy

Create an IAM policy that allows full access to your S3 Tables API.Replace <S3 table bucket ARN> with the ARN of your S3 table bucket:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "s3tables:*",
            "Resource": [
                "<S3 table bucket ARN>",
                "<S3 table bucket ARN>/table/*"
            ]
        }
    ]
}

Step 1B. Create an IAM role

Create an IAM role that Materialize can assume.

1. For the Trusted entity type, specify Custom trust policy with the following:

   • Principal: The example uses the Materialize Cloud IAM principal. For self-managed deployments and the Emulator, the principal will differ.
   • ExternalId: "PENDING" is a placeholder and will be updated after creating the AWS connection in Materialize.

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Effect": "Allow",
               "Principal": {
                   "AWS": "arn:aws:iam::664411391173:role/MaterializeConnection"
               },
               "Action": "sts:AssumeRole",
               "Condition": {
                   "StringEquals": {
                       "sts:ExternalId": "PENDING"
                   }
               }
           }
       ]
   }
   

2. For permissions, add the IAM policy created earlier to grant access to the S3 Tables.

Once you have created the IAM role, copy the role ARN from the AWS console. You’ll use the ARN in the next step.

Step 2. Create an AWS connection in Materialize

In Materialize, create an AWS connection to authenticate with the object storage:

1. Use CREATE CONNECTION ... TO AWS to create an AWS connection, replacing:

   • <IAM role ARN> with your IAM role ARN from step 1
   • <region> with your AWS region (e.g., us-east-1):

   CREATE CONNECTION aws_connection TO AWS (
       ASSUME ROLE ARN = '<IAM role ARN>',
       REGION = '<region>'
   );
   

   For more details on AWS connection options, see CREATE CONNECTION.

2. Fetch the external_id for your connection, replacing <IAM role ARN> with your IAM role ARN:

   SELECT external_id
   FROM mz_internal.mz_aws_connections awsc
   JOIN mz_connections c ON awsc.id = c.id
   WHERE c.name = 'aws_connection'
   AND awsc.assume_role_arn = '<iam-role-arn>';
   

   You will use the external_id to update the IAM role in the next step.

Step 3. Update the IAM role in AWS

Once you have the external_id, update the trust policy for the IAM role created in step 1. Replace "PENDING" with your external ID value. Your IAM trust policy should look like the following (but with your external ID value):

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "AWS": "arn:aws:iam::664411391173:role/MaterializeConnection"
            },
            "Action": "sts:AssumeRole",
            "Condition": {
                "StringEquals": {
                    "sts:ExternalId": "mz_1234abcd-5678-efgh-9012-ijklmnopqrst_u123"
                }
            }
        }
    ]
}

Step 4. Create an Iceberg catalog connection in Materialize

In Materialize, create an Iceberg catalog connection for the Iceberg sink to use. To create, use CREATE CONNECTION ... TO ICEBERG CATALOG, replacing:

• <region> with your AWS region (e.g., us-east-1) and
• <S3 table bucket ARN> with your AWS S3 Table bucket ARN.

The command uses the AWS connection you created earlier.

CREATE CONNECTION iceberg_catalog_connection TO ICEBERG CATALOG (
    CATALOG TYPE = 's3tablesrest',
    URL = 'https://s3tables.<region>.amazonaws.com/iceberg',
    WAREHOUSE = '<S3 table bucket ARN>',
    AWS CONNECTION = aws_connection
);

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>');

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

Type mapping

Materialize converts SQL types to Iceberg/Parquet types:

Limitations

• Your S3 Tables bucket must be in the same AWS region as your Materialize deployment.
• 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
• Apache Iceberg documentation