CREATE SOURCE: MySQL (Legacy syntax)

Creates a new source from MySQL. Materialize supports creating sources from MySQL (8.0.1+).

To connect to a MySQL database, you first need to update its configuration to enable GTID-based binary log (binlog) replication, and then create a connection in Materialize that specifies access and authentication parameters.

Syntax

CREATE SOURCE [IF NOT EXISTS] <src_name>
[IN CLUSTER <cluster_name>]
FROM MYSQL CONNECTION <connection_name> [
  (
    [TEXT COLUMNS ( <col1> [, ...] ) ]
    [, EXCLUDE COLUMNS ( <col1> [, ...] ) ]
  )
]
<FOR ALL TABLES | FOR SCHEMAS ( <schema1> [, ...] ) | FOR TABLES ( <table1> [AS <subsrc_name>] [, ...] )>
[EXPOSE PROGRESS AS <progress_subsource_name>]
[WITH ( <with_option> [, ...] )]

CONNECTION options

Features

Change data capture

The source uses MySQL’s binlog replication protocol to continually ingest changes resulting from INSERT, UPDATE and DELETE operations in the upstream database. This process is known as change data capture.

The replication method used is based on global transaction identifiers (GTIDs), and guarantees transactional consistency — any operation inside a MySQL transaction is assigned the same timestamp in Materialize, which means that the source will never show partial results based on partially replicated transactions.

Before creating a source in Materialize, you must configure the upstream MySQL database for GTID-based binlog replication. Ensure the upstream MySQL database has been configured for GTID-based binlog replication:

If you’re running MySQL using a managed service, additional configuration changes might be required. For step-by-step instructions on enabling GTID-based binlog replication for your MySQL service, see the integration guides.

Binlog retention

By default, MySQL retains binlog files for 30 days (i.e., 2592000 seconds) before automatically removing them. This is configurable via the binlog_expire_logs_seconds system variable. We recommend using the default value for this configuration in order to not compromise Materialize’s ability to resume replication in case of failures or restarts.

In some MySQL managed services, binlog expiration can be overriden by a service-specific configuration parameter. It’s important that you double-check if such a configuration exists, and ensure it’s set to the maximum interval available.

As an example, Amazon RDS for MySQL has its own configuration parameter for binlog retention (binlog retention hours) that overrides binlog_expire_logs_seconds and is set to NULL by default.

Creating a source

Materialize ingests the raw replication stream data for all (or a specific set of) tables in your upstream MySQL database.

CREATE SOURCE mz_source
  FROM MYSQL CONNECTION mysql_connection
  FOR ALL TABLES;

When you define a source, Materialize will automatically:

1. Create a subsource for each original table upstream, and perform an initial, snapshot-based sync of the tables before it starts ingesting change events.

   SHOW SOURCES;
   

            name         |   type    |  cluster  |
   ----------------------+-----------+------------
    mz_source            | mysql     |
    mz_source_progress   | progress  |
    table_1              | subsource |
    table_2              | subsource |
   

2. Incrementally update any materialized or indexed views that depend on the source as change events stream in, as a result of INSERT, UPDATE and DELETE operations in the upstream MySQL database.

MySQL schemas

CREATE SOURCE will attempt to create each upstream table in the same schema as the source. This may lead to naming collisions if, for example, you are replicating schema1.table_1 and schema2.table_1. Use the FOR TABLES clause to provide aliases for each upstream table, in such cases, or to specify an alternative destination schema in Materialize.

CREATE SOURCE mz_source
  FROM MYSQL CONNECTION mysql_connection
  FOR TABLES (schema1.table_1 AS s1_table_1, schema2.table_1 AS s2_table_1);

Monitoring source progress

By default, MySQL sources expose progress metadata as a subsource that you can use to monitor source ingestion progress. The name of the progress subsource can be specified when creating a source using the EXPOSE PROGRESS AS clause; otherwise, it will be named <src_name>_progress.

The following metadata is available for each source as a progress subsource:

And can be queried using:

SELECT transaction_id
FROM <src_name>_progress;

Progress metadata is represented as a GTID set of future possible GTIDs, which is similar to the gtid_executed system variable on a MySQL replica. The reported transaction_id should increase as Materialize consumes new binlog records from the upstream MySQL database. For more details on monitoring source ingestion progress and debugging related issues, see Troubleshooting.

Known limitations

Supported types

Materialize natively supports the following MySQL types:

• bigint
• binary
• bit
• blob
• boolean
• char
• date
• datetime
• decimal
• double
• float
• int
• json
• longblob
• longtext
• mediumblob
• mediumint
• mediumtext
• numeric
• real
• smallint
• text
• time
• timestamp
• tinyblob
• tinyint
• tinytext
• varbinary
• varchar

When replicating tables that contain the unsupported data types, you can:

• Use TEXT COLUMNS option for the following unsupported MySQL types:

  • enum
  • year

  The specified columns will be treated as text and will not offer the expected MySQL type features.

• Use the EXCLUDE COLUMNS option to exclude any columns that contain unsupported data types.

Zero values for date, datetime, and timestamp

MySQL allows the special “zero” values 0000-00-00, 0000-00-00 00:00:00 in date, datetime, and timestamp columns when the server sql_mode does not include NO_ZERO_DATE or NO_ZERO_IN_DATE. These values are not representable in Materialize’s corresponding native types, so they will cause ingestion to fail for the affected column.

To ingest columns that contain zero values, use TEXT COLUMNS to decode the affected columns as text. The zero values for date, datetime, timestamp, and year are preserved verbatim as strings (e.g. "0000-00-00 00:00:00", "0000").

Snapshot parallelism

When a source is created, Materialize parallelizes the initial snapshot across the cluster’s workers, and can split the read of a large table across workers when the table meets certain requirements. See Snapshot parallelism.

Adding a table to an existing source

When you add a new subsource to an existing source (ALTER SOURCE ... ADD SUBSOURCE ...), Materialize starts the snapshotting process for the new subsource. During this snapshotting, the data ingestion for the existing subsources for the same source is temporarily blocked. As such, if possible, you can resize the cluster to speed up the snapshotting process and once the process finishes, resize the cluster for steady-state.

Handling upstream operations

This section describes how changes to upstream tables that Materialize ingests affect the corresponding Materialize tables.

Adding a column

When you add a new column to your upstream table, Materialize continues to ingest only the existing columns.

To incorporate the new column:

• If using the new CREATE SOURCE and CREATE TABLE FROM SOURCE syntax, create a new table from the source. See Handle upstream column addition.

• If using the legacy CREATE SOURCE ... FOR ... syntax that creates subsources, use DROP SOURCE to drop the affected subsource, and then add the table back to the source using ALTER SOURCE ... ADD SUBSOURCE. The re-added subsource includes the new column.

Dropping a column

Dropping columns that Materialize does not ingest (for example, columns added after the source was created, or columns that are excluded) is supported. As these columns were never ingested, you can drop them without issue.

If your Materialize source ingests a column, dropping that column from your upstream table puts the affected table into an error state.

• If using the new CREATE SOURCE and CREATE TABLE FROM SOURCE syntax, you can safely drop a column by first ignoring it in Materialize. See Handle upstream column drop.

• If using legacy CREATE SOURCE ... FOR ... syntax, use DROP SOURCE to drop the affected subsource, and then add the table back to the source using ALTER SOURCE ... ADD SUBSOURCE.

Changing constraints

Materialize ignores the following constraint changes: foreign key and CHECK. As such, you can add or drop them without affecting ingestion.

Materialize also ignores NOT NULL, UNIQUE, and PRIMARY KEY constraints that are added after the Materialize table is created (that is, the table was created without them). Adding such a constraint, and later dropping it, does not affect ingestion.

Dropping a NOT NULL, UNIQUE, or PRIMARY KEY constraint that existed when the table was created puts the affected table into an error state.

If using the new CREATE SOURCE and CREATE TABLE FROM SOURCE syntax, you can safely drop such a constraint by first excluding it in Materialize. See Handle upstream constraint drop.

Changing a column’s data type

Changing an ingested column’s data type upstream so that it maps to a different Materialize type than before puts the affected Materialize table into an error state. Ingestion for that table stops, and you must drop and recreate the table in Materialize to resume ingestion.

Changing an ingested column’s upstream data type so that it continues to map to the same Materialize type does not interrupt ingestion. For example, changing tinyint to smallint, changing within the text/tinytext/mediumtext/longtext family, and adjusting bit(n) precision are all safe.

Appending new values to the end of an existing enum does not put the table into an error state. However, the newly-added values are not recognized, so rows that use them fail to decode until you drop and recreate the table. Existing enum values remain recognized, and rows that use them continue to decode successfully.

Any other enum change puts the affected Materialize table into an error state, including inserting a value before the end, reordering or renaming values, and removing values.

Renaming a column

Renaming a column that Materialize ingests puts the affected table into an error state. Ingestion for that table stops, and you must drop and recreate the table in Materialize to resume ingestion.

Table-level operations

The following upstream operations put the affected table into an error state. Ingestion for that table stops, and you must drop and recreate the affected table in Materialize to resume:

• Dropping a table (DROP TABLE).
• Renaming a table or moving it to a different schema.
• Truncating a table (TRUNCATE). To clear a table without putting it into an error state, use an unqualified DELETE FROM t; instead.

Source failure states and recovery

Operations that do not require re-creating the source

Materialize tracks its position in the upstream binary log as a set of global transaction identifiers (GTIDs), which it persists alongside the ingested data. A GTID identifies a transaction across the whole replication topology, not a byte offset in a particular binlog file, so it survives routine operational events: after an interruption the source reconnects, asks the server for the transactions after its last committed GTID, and catches up.

The source recovers on its own. It briefly reports a stalled status while the condition persists, then returns to running and catches up for all of the following:

• Restarting or patching MySQL (including OS-level restarts).
• Restarting Materialize. The source resumes from its tracked GTID set and does not re-snapshot already-ingested data.
• Transient network interruptions between Materialize and MySQL.
• The upstream server running out of disk space, until space is reclaimed.
• A long-running upstream transaction blocking the initial snapshot. The source stalls until that transaction commits or rolls back.
• Failing over to a replica, with the checks described below.

Operations that require re-creating the source

A smaller set of events breaks GTID continuity or makes the binlog stream unreadable. Materialize cannot then guarantee a correct, gap-free view of your data, so it puts the entire source into an error state that requires re-creating the source. Upstream changes to an individual table’s schema are handled separately, and do not error the entire source.

In each case, the remediation is to drop the source and create it again with the statements you originally used, which triggers a fresh snapshot:

DROP SOURCE mz_source CASCADE;

Once the source is back, it is healthy when status is running and error is NULL:

SELECT status, error FROM mz_internal.mz_source_statuses WHERE name = 'mz_source';

Binlog files removed before the resume point

MySQL expires binlog files on a retention schedule, and PURGE BINARY LOGS removes them on demand. If Materialize is disconnected or lagging long enough that the files holding its resume point are removed, the source fails with one of:

mysql server does not have the binlog available at the requested gtid set

mysql server binlog frontier at <frontier> is beyond required frontier <frontier>

To avoid this, keep planned outages shorter than the retention window and monitor source lag against it. For how retention is configured, including the service-specific parameters that override it, see Binlog retention.

Resetting the binary log

RESET BINARY LOGS AND GTIDS (RESET MASTER before MySQL 8.4) discards the binlog files and the server’s GTID history, so the source’s resume point no longer exists. The source fails with:

mysql server does not have the binlog available at the requested gtid set

or, if the server then reissues GTIDs the source has already seen, with an out-of-order GTID error.

Changing a required replication setting

Materialize re-validates the upstream replication settings each time it (re-)establishes the replication stream. If one no longer holds its required value, the source fails with:

mysql server configuration: invalid mysql system setting '<setting>'. Expected '<expected>'. Got '<actual>'.

The validated settings are log_bin, binlog_format, binlog_row_image, gtid_mode, enforce_gtid_consistency, gtid_next, and, when replica_parallel_workers is greater than 1, replica_preserve_commit_order (slave_preserve_commit_order on servers older than MySQL 8.0). For the required values, see Change data capture. Restore the setting upstream, then re-create the source.

Restoring the upstream database

Materialize has no dedicated check for a restore of the upstream database. Depending on how the restore handles GTIDs, the source either fails with one of the errors above or with:

received a gtid set from the server that violates our requirements: <detail>

or, if the restore reuses GTIDs the source has already ingested, it keeps replicating against a history that no longer matches its state. Materialize cannot detect that case, so re-create the source after any restore of the upstream database, including a disaster-recovery restore onto a new server, rather than relying on an error.

Out-of-order GTIDs

If Materialize observes GTIDs in an order it cannot reconcile, the source fails with:

received out of order gtids for source <source_id> at transaction-id <transaction_id>

This is most common when Materialize replicates from a MySQL replica that applies transactions with multiple threads. See Troubleshooting: Received out of order GTIDs for the diagnosis steps and the upstream settings that make it less likely.

Operations that require re-creating only the affected tables

Lowering binlog_row_metadata

A table that Materialize began ingesting while binlog_row_metadata was FULL can no longer be decoded once the setting is lowered, and enters an error state with:

unable to decode: Table <table> was created with binlog_row_metadata=FULL but
binlog_row_metadata has since been set to a different value, meaning we cannot
reliably decode the columns

Restoring binlog_row_metadata=FULL does not clear the error. Set it back to FULL and re-create the affected tables. Tables that were created while the setting was lower keep replicating.

Failovers

Because Materialize replicates by GTID rather than by binlog file and position, it can follow a failover to a replica that was replicating from the failed server with GTIDs enabled. Point the connection at the new primary with ALTER CONNECTION, or at an endpoint that always resolves to the current primary.

On the new primary, confirm that:

• The binlog files covering the source’s resume point are present. A replica configured with a shorter retention than the old primary can leave the source with no way to resume. See Binlog files removed before the resume point.
• The required replication settings hold, including replica_preserve_commit_order while the server is still applying replication from another server. Multi-threaded apply is the main source of out-of-order GTIDs; preserving commit order reduces but does not eliminate the risk, so keep replica_parallel_workers at 0 or 1 where you can.

Examples

Creating a connection

A connection describes how to connect and authenticate to an external system you want Materialize to read data from.

Once created, a connection is reusable across multiple CREATE SOURCE statements. For more details on creating connections, check the CREATE CONNECTION documentation page.

CREATE SECRET mysqlpass AS '<MYSQL_PASSWORD>';

CREATE CONNECTION mysql_connection TO MYSQL (
    HOST 'instance.foo000.us-west-1.rds.amazonaws.com',
    PORT 3306,
    USER 'materialize',
    PASSWORD SECRET mysqlpass
);

If your MySQL server is not exposed to the public internet, you can tunnel the connection through an AWS PrivateLink service (Materialize Cloud) or an SSH bastion host SSH bastion host.

CREATE CONNECTION privatelink_svc TO AWS PRIVATELINK (
   SERVICE NAME 'com.amazonaws.vpce.us-east-1.vpce-svc-0e123abc123198abc',
   AVAILABILITY ZONES ('use1-az1', 'use1-az4')
);

CREATE CONNECTION mysql_connection TO MYSQL (
    HOST 'instance.foo000.us-west-1.rds.amazonaws.com',
    PORT 3306,
    USER 'root',
    PASSWORD SECRET mysqlpass,
    AWS PRIVATELINK privatelink_svc
);

CREATE CONNECTION ssh_connection TO SSH TUNNEL (
    HOST 'bastion-host',
    PORT 22,
    USER 'materialize'
);

CREATE CONNECTION mysql_connection TO MYSQL (
    HOST 'instance.foo000.us-west-1.rds.amazonaws.com',
    SSH TUNNEL ssh_connection
);

Creating a source

Create subsources for all tables in MySQL

CREATE SOURCE mz_source
    FROM MYSQL CONNECTION mysql_connection
    FOR ALL TABLES;

Create subsources for all tables from specific schemas in MySQL

CREATE SOURCE mz_source
  FROM MYSQL CONNECTION mysql_connection
  FOR SCHEMAS (mydb, project);

Create subsources for specific tables in MySQL

CREATE SOURCE mz_source
  FROM MYSQL CONNECTION mysql_connection
  FOR TABLES (mydb.table_1, mydb.table_2 AS alias_table_2);

Handling unsupported types

If you’re replicating tables that use data types unsupported by Materialize, use the TEXT COLUMNS option to decode data as text for the affected columns. TEXT COLUMNS should also be used for columns that contain MySQL zero-value DATE, DATETIME, or TIMESTAMP data.

This option expects the upstream fully-qualified names of the replicated table and column (i.e. as defined in your MySQL database).

CREATE SOURCE mz_source
  FROM MYSQL CONNECTION mysql_connection (
    TEXT COLUMNS (mydb.table_1.column_of_unsupported_type)
  )
  FOR ALL TABLES;

Excluding columns

MySQL doesn’t provide a way to filter out columns from the replication stream. To exclude specific upstream columns from being ingested, use the EXCLUDE COLUMNS option.

CREATE SOURCE mz_source
  FROM MYSQL CONNECTION mysql_connection (
    EXCLUDE COLUMNS (mydb.table_1.column_to_ignore)
  )
  FOR ALL TABLES;

Handling errors and schema changes

To handle upstream schema changes or errored subsources, use the DROP SOURCE syntax to drop the affected subsource, and then ALTER SOURCE...ADD SUBSOURCE to add the subsource back to the source.

-- List all subsources in mz_source
SHOW SUBSOURCES ON mz_source;

-- Get rid of an outdated or errored subsource
DROP SOURCE table_1;

-- Start ingesting the table with the updated schema or fix
ALTER SOURCE mz_source ADD SUBSOURCE table_1;

Related pages

• CREATE SECRET
• CREATE CONNECTION
• CREATE SOURCE
• MySQL integration guides:
  • Amazon RDS
  • Amazon Aurora
  • Azure DB
  • Google Cloud SQL
  • Self-hosted