Materialize supports schema changes in the upstream database as follows:
Note: This section refer to the legacy
CREATE SOURCE ... FOR ...that creates subsources as part of theCREATE SOURCEoperation. To be able to handle the upstream column additions and drops, seeCREATE SOURCE (New Syntax)andCREATE TABLE FROM SOURCE.
Adding columns to tables. Materialize will not ingest new columns
added upstream unless you use DROP SOURCE to
first drop the affected subsource, and then add the table back to the source
using ALTER SOURCE...ADD SUBSOURCE.
Dropping columns that were added after the source was created. These columns are never ingested, so you can drop them without issue.
Adding or removing NOT NULL constraints to tables that were nullable
when the source was created.
All other schema changes to upstream tables will set the corresponding Materialize tables into an error state, preventing reads from these tables.
To handle incompatible schema changes, drop
the affected table DROP TABLE , and then, CREATE TABLE FROM SOURCE to recreate the table with the
updated schema.
PostgreSQL’s logical replication API does not provide a signal when users remove tables from publications. Because of this, Materialize relies on periodic checks to determine if a table has been removed from a publication, at which time it generates an irrevocable error, preventing any values from being read from the table.
However, it is possible to remove a table from a publication and then re-add it before Materialize notices that the table was removed. In this case, Materialize can no longer provide any consistency guarantees about the data we present from the table and, unfortunately, is wholly unaware that this occurred.
To mitigate this issue, if you need to drop and re-add a table to a
publication, ensure that you remove the table/subsource from the source
before re-adding it using the DROP SOURCE command.
Materialize natively supports the following PostgreSQL types (including the array type for each of the types):
boolbpcharbyteachardatedaterangefloat4float8int2int2vectorint4int4rangeint8int8rangeintervaljsonjsonbnumericnumrangeoidtexttimetimestamptimestamptztsrangetstzrangeuuidvarcharReplicating tables that contain unsupported data types is
possible via the TEXT COLUMNS option. The specified columns will be
treated as text; i.e., will not have the expected PostgreSQL type
features. For example:
enum: When decoded as text, the implicit ordering of the original
PostgreSQL enum type is not preserved; instead, Materialize will sort values
as text.
money: When decoded as text, resulting text value cannot be cast
back to numeric, since PostgreSQL adds typical currency formatting to the
output.
Avoid truncating upstream tables that are being replicated into Materialize. If a replicated upstream table is truncated, the corresponding subsource(s)/table(s) in Materialize becomes inaccessible and will not produce any data until it is recreated.
Instead of truncating, use an unqualified DELETE to remove all rows from
the upstream table:
DELETE FROM t;
When using PostgreSQL table inheritance,
PostgreSQL serves data from SELECTs as if the inheriting tables’ data is
also present in the inherited table. However, both PostgreSQL’s logical
replication and COPY only present data written to the tables themselves,
i.e. the inheriting data is not treated as part of the inherited table.
PostgreSQL sources use logical replication and COPY to ingest table data,
so inheriting tables’ data will only be ingested as part of the inheriting
table, i.e. in Materialize, the data will not be returned when serving
SELECTs from the inherited table.
If using legacy syntax CREATE SOURCE ... FOR ...:
You can mimic PostgreSQL’s SELECT behavior with inherited tables by
creating a materialized view that unions data from the inherited and
inheriting tables (using UNION ALL). However, if new tables inherit from
the table, data from the inheriting tables will not be available in the
view. You will need to add the inheriting tables via ADD SUBSOURCE and
create a new view (materialized or non-) that unions the new table.
If using new CREATE TABLE FROM SOURCE syntax:
You can mimic PostgreSQL’s SELECT behavior with inherited tables by
creating a materialized view that unions data from the inherited and
inheriting tables (using UNION ALL). However, if new tables inherit from
the table, data from the inheriting tables will not be available in the
view. You will need to add the inheriting tables via CREATE TABLE .. FROM SOURCE and create a new view (materialized or non-) that unions the new
table.
Each source ingests the raw replication stream data for all tables in the specified publication using a single replication slot. To manage replication slots:
For PostgreSQL 13+, set a reasonable value
for max_slot_wal_keep_size
to limit the amount of storage used by replication slots.
If you stop using Materialize, or if either the Materialize instance or
the PostgreSQL instance crash, delete any replication slots. You can query
the mz_internal.mz_postgres_sources table to look up the name of the
replication slot created for each source.
If you delete all objects that depend on a source without also dropping
the source, the upstream replication slot remains and will continue to
accumulate data so that the source can resume in the future. To avoid
unbounded disk space usage, make sure to use DROP SOURCE or manually delete the replication slot.
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.
The PostgreSQL source performs parallel snapshotting of tables by distributing rows among
workers using ranges of
CTID.
Materialize uses
PostgreSQL statistics to estimate
the amount of data and number of rows to read. Missing or stale statistics can result in uneven
work distribution, reducing snapshot performance. They can also cause incorrect snapshot
progress reporting in the Console.
To avoid this situation, before creating the source in Materialize, ensure statistics are up to
date by running PostgreSQL ANALYZE command.
Replicating tables that contain unsupported data types is
possible via the TEXT COLUMNS option. The specified columns will be
treated as text; i.e., will not have the expected PostgreSQL type
features. For example:
enum: When decoded as text, the implicit ordering of the original
PostgreSQL enum type is not preserved; instead, Materialize will sort values
as text.
money: When decoded as text, resulting text value cannot be cast
back to numeric, since PostgreSQL adds typical currency formatting to the
output.
To create a source from PostgreSQL 11+, you must first:
For details, see the PostgreSQL integration guides.
## Create a source using the new syntax In Materialize, create a source using the updated [`CREATE SOURCE` syntax](/sql/create-source/postgres-v2/). ```sql CREATE SOURCE IF NOT EXISTS my_source FROM POSTGRES CONNECTION my_connection (PUBLICATION 'mz_source'); ``` Unlike the [legacy syntax](/sql/create-source/postgres/), the new syntax does not include the `FOR [[ALL] TABLES|SCHEMAS]` clause; i.e., the new syntax does not create corresponding subsources in Materialize automatically. Instead, the new syntax requires a separate [`CREATE TABLE ... FROM SOURCE`](/sql/create-table/), which will create the corresponding tables and start the snapshotting process. See [Create a table from the source](#create-a-table-from-the-source). > **Note:** The [legacy syntax](/sql/create-source/postgres/) is still supported. However, > the legacy syntax doesn't support upstream schema changes. ## Create a table from the source To start ingesting specific tables from your source database, you can create a table in Materialize. We'll add it into the v1 schema in Materialize. ```sql CREATE SCHEMA v1; CREATE TABLE v1.T FROM SOURCE my_source(REFERENCE public.T); ``` Once you've created a table from source, the [initial snapshot](/ingest-data/#snapshotting) of table `v1.T` will begin. > **Note:** During the snapshotting, the data ingestion for the other tables associated with > the source is temporarily blocked. As before, you can monitor progress for the > snapshot operation on the overview page for the source in the Materialize > console. ## Create a view on top of the table. For this guide, add a materialized view `matview` (also in schema `v1`) that sums column `A` from table `T`. ```sql CREATE MATERIALIZED VIEW v1.matview AS SELECT SUM(A) from v1.T; ``` ## Handle upstream column addition ### A. Add a column in your upstream PostgreSQL database In your upstream PostgreSQL database, add a new column `B` to the table `T`: ```sql ALTER TABLE T ADD COLUMN B BOOLEAN DEFAULT false; INSERT INTO T (A, B) VALUES (20, true); ``` This operation will have no immediate effect in Materialize. In Materialize, `v1.T` will continue to ingest only column `A`. The materialized view `v1.matview` will continue to have access to column `A` as well. ### B. Incorporate the new column in Materialize To incorporate the new column into Materialize, create a new `v2` schema and recreate the table in the new schema: ```sql CREATE SCHEMA v2; CREATE TABLE v2.T FROM SOURCE my_source(REFERENCE public.T); ``` The [snapshotting](/ingest-data/#snapshotting) of table `v2.T` will begin. `v2.T` will include columns `A` and `B`. > **Note:** During the snapshotting, the data ingestion for the other tables associated with > the source is temporarily blocked. As before, you can monitor progress for the > snapshot operation on the overview page for the source in the Materialize > console. When the new `v2.T` table has finished snapshotting, create a new materialized view `matview` in the new schema. Since the new `v2.matview` is referencing the new `v2.T`, it can reference column `B`: ```sql {hl_lines="4"} CREATE MATERIALIZED VIEW v2.matview AS SELECT SUM(A) FROM v2.T WHERE B = true; ``` ## Handle upstream column drop ### A. Exclude the column in Materialize To drop a column safely, in Materialize, first, create a new `v3` schema, and recreate table `T` in the new schema but exclude the column to drop. In this example, we'll drop the column B. ```sql CREATE SCHEMA v3; CREATE TABLE v3.T FROM SOURCE my_source(REFERENCE public.T) WITH (EXCLUDE COLUMNS (B)); ``` > **Note:** During the snapshotting, the data ingestion for the other tables associated with > the source is temporarily blocked. As before, you can monitor progress for the > snapshot operation on the overview page for the source in the Materialize > console. ### B. Drop a column in your upstream PostgreSQL database In your upstream PostgreSQL database, drop the column `B` from the table `T`: ```sql ALTER TABLE T DROP COLUMN B; ``` Dropping the column B will have no effect on `v3.T`. However, the drop affects `v2.T` and `v2.matview` from our earlier examples. When the user attempts to read from either, Materialize will report an error that the source table schema has been altered. --- ## Ingest data from AlloyDB This page shows you how to stream data from [AlloyDB for PostgreSQL](https://cloud.google.com/alloydb) to Materialize using the [PostgreSQL source](/sql/create-source/postgres/). > **Tip:** For help getting started with your own data, you can schedule a [free guided > trial](https://materialize.com/demo/?utm_campaign=General&utm_source=documentation). ## Before you beginMake sure you are running PostgreSQL 11 or higher.
Make sure you have access to your PostgreSQL instance via psql,
or your preferred SQL client.
Once logical replication is enabled, the next step is to create a publication with the tables that you want to replicate to Materialize. You’ll also need a user for Materialize with sufficient privileges to manage replication.
For each table that you want to replicate to Materialize, set the
replica identity
to FULL:
ALTER TABLE <table1> REPLICA IDENTITY FULL;
ALTER TABLE <table2> REPLICA IDENTITY FULL;
REPLICA IDENTITY FULL ensures that the replication stream includes the
previous data of changed rows, in the case of UPDATE and DELETE
operations. This setting enables Materialize to ingest PostgreSQL data with
minimal in-memory state. However, you should expect increased disk usage in
your PostgreSQL database.
Create a publication with the tables you want to replicate:
For specific tables:
CREATE PUBLICATION mz_source FOR TABLE <table1>, <table2>;
For all tables in the database:
CREATE PUBLICATION mz_source FOR ALL TABLES;
The mz_source publication will contain the set of change events generated
from the specified tables, and will later be used to ingest the replication
stream.
Be sure to include only the tables you need. If the publication includes additional tables, Materialize will waste resources on ingesting and then immediately discarding the data.
Create a user for Materialize, if you don’t already have one:
CREATE USER materialize PASSWORD '<password>';
Grant the user permission to manage replication:
ALTER ROLE materialize WITH REPLICATION;
Grant the user the required permissions on the tables you want to replicate:
GRANT CONNECT ON DATABASE <dbname> TO materialize;
GRANT USAGE ON SCHEMA <schema> TO materialize;
GRANT SELECT ON <table1> TO materialize;
GRANT SELECT ON <table2> TO materialize;
Once connected to your database, Materialize will take an initial snapshot
of the tables in your publication. SELECT privileges are required for
this initial snapshot.
If you expect to add tables to your publication, you can grant SELECT on
all tables in the schema instead of naming the specific tables:
GRANT SELECT ON ALL TABLES IN SCHEMA <schema> TO materialize;
Configure your network to allow Materialize to connect to your database. For example, you can:
Allow Materialize IPs: Configure your database’s security group to allow connections from Materialize.
Use an SSH tunnel: Use an SSH tunnel to connect Materialize to the database.
The steps to allow Materialize to connect to your database depends on your deployment setup. Refer to your company’s network/security policies and procedures.
In Materialize, a cluster is an isolated environment, similar to a virtual warehouse in Snowflake. When you create a cluster, you choose the size of its compute resource allocation based on the work you need the cluster to do, whether ingesting data from a source, computing always-up-to-date query results, serving results to external clients, or a combination.
In this step, you’ll create a dedicated cluster for ingesting source data from your PostgreSQL database.
In the SQL Shell, or your preferred SQL
client connected to Materialize, use the CREATE CLUSTER
command to create the new cluster:
CREATE CLUSTER ingest_postgres (SIZE = '50cc');
SET CLUSTER = ingest_postgres;
A cluster of size 50cc should be enough to
accommodate multiple PostgreSQL sources, depending on the source
characteristics (e.g., sources with ENVELOPE UPSERT
or ENVELOPE DEBEZIUM will be more
memory-intensive) and the upstream traffic patterns. You can readjust the
size of the cluster at any time using the ALTER CLUSTER command:
ALTER CLUSTER <cluster_name> SET ( SIZE = <new_size> );
Before it starts consuming the replication stream, Materialize takes a snapshot of the relevant tables in your publication. Until this snapshot is complete, Materialize won’t have the same view of your data as your PostgreSQL database.
In this step, you’ll first verify that the source is running and then check the status of the snapshotting process.
Back in the SQL client connected to Materialize, use the
mz_source_statuses
table to check the overall status of your source:
WITH
source_ids AS
(SELECT id FROM mz_sources WHERE name = 'mz_source')
SELECT *
FROM
mz_internal.mz_source_statuses
JOIN
(
SELECT referenced_object_id
FROM mz_internal.mz_object_dependencies
WHERE
object_id IN (SELECT id FROM source_ids)
UNION SELECT id FROM source_ids
)
AS sources
ON mz_source_statuses.id = sources.referenced_object_id;
For each subsource, make sure the status is running. If you see
stalled or failed, there’s likely a configuration issue for you to fix.
Check the error field for details and fix the issue before moving on.
Also, if the status of any subsource is starting for more than a few
minutes, contact our team.
Once the source is running, use the mz_source_statistics
table to check the status of the initial snapshot:
WITH
source_ids AS
(SELECT id FROM mz_sources WHERE name = 'mz_source')
SELECT sources.referenced_object_id AS id, mz_sources.name, snapshot_committed
FROM
mz_internal.mz_source_statistics
JOIN
(
SELECT object_id, referenced_object_id
FROM mz_internal.mz_object_dependencies
WHERE
object_id IN (SELECT id FROM source_ids)
UNION SELECT id, id FROM source_ids
)
AS sources
ON mz_source_statistics.id = sources.referenced_object_id
JOIN mz_sources ON mz_sources.id = sources.referenced_object_id;
object_id | snapshot_committed
----------|------------------
u144 | t
(1 row)
Once snapshot_commited is t, move on to the next step. Snapshotting can
take between a few minutes to several hours, depending on the size of your
dataset and the size of the cluster the source is running in.
After the snapshotting phase, Materialize starts ingesting change events from
the PostgreSQL replication stream. For this work, Materialize generally
performs well with an 100cc replica, so you can resize the cluster
accordingly.
Still in a SQL client connected to Materialize, use the ALTER CLUSTER
command to downsize the cluster to 100cc:
ALTER CLUSTER ingest_postgres SET (SIZE '100cc');
Behind the scenes, this command adds a new 100cc replica and removes the
50cc replica.
Use the SHOW CLUSTER REPLICAS command to
check the status of the new replica:
SHOW CLUSTER REPLICAS WHERE cluster = 'ingest_postgres';
cluster | replica | size | ready
-----------------+---------+--------+-------
ingest_postgres | r1 | 100cc | t
(1 row)
Going forward, you can verify that your new cluster size is sufficient as follows:
In Materialize, get the replication slot name associated with your
PostgreSQL source from the mz_internal.mz_postgres_sources
table:
SELECT
d.name AS database_name,
n.name AS schema_name,
s.name AS source_name,
pgs.replication_slot
FROM
mz_sources AS s
JOIN mz_internal.mz_postgres_sources AS pgs ON s.id = pgs.id
JOIN mz_schemas AS n ON n.id = s.schema_id
JOIN mz_databases AS d ON d.id = n.database_id;
In PostgreSQL, check the replication slot lag, using the replication slot name from the previous step:
SELECT
pg_size_pretty(pg_current_wal_lsn() - confirmed_flush_lsn)
AS replication_lag_bytes
FROM pg_replication_slots
WHERE slot_name = '<slot_name>';
The result of this query is the amount of data your PostgreSQL cluster must retain in its replication log because of this replication slot. Typically, this means Materialize has not yet communicated back to PostgreSQL that it has committed this data. A high value can indicate that the source has fallen behind and that you might need to scale up your ingestion cluster.
With Materialize ingesting your PostgreSQL data into durable storage, you can start exploring the data, computing real-time results that stay up-to-date as new data arrives, and serving results efficiently.
Explore your data with SHOW SOURCES and SELECT.
Compute real-time results in memory with CREATE VIEW
and CREATE INDEX or in durable
storage with CREATE MATERIALIZED VIEW.
Serve results to a PostgreSQL-compatible SQL client or driver with SELECT
or SUBSCRIBE or to an external message broker with
CREATE SINK.
Check out the tools and integrations supported by Materialize.
Materialize supports schema changes in the upstream database as follows:
Note: This section refer to the legacy
CREATE SOURCE ... FOR ...that creates subsources as part of theCREATE SOURCEoperation. To be able to handle the upstream column additions and drops, seeCREATE SOURCE (New Syntax)andCREATE TABLE FROM SOURCE.
Adding columns to tables. Materialize will not ingest new columns
added upstream unless you use DROP SOURCE to
first drop the affected subsource, and then add the table back to the source
using ALTER SOURCE...ADD SUBSOURCE.
Dropping columns that were added after the source was created. These columns are never ingested, so you can drop them without issue.
Adding or removing NOT NULL constraints to tables that were nullable
when the source was created.
All other schema changes to upstream tables will set the corresponding Materialize tables into an error state, preventing reads from these tables.
To handle incompatible schema changes, drop
the affected table DROP TABLE , and then, CREATE TABLE FROM SOURCE to recreate the table with the
updated schema.
PostgreSQL’s logical replication API does not provide a signal when users remove tables from publications. Because of this, Materialize relies on periodic checks to determine if a table has been removed from a publication, at which time it generates an irrevocable error, preventing any values from being read from the table.
However, it is possible to remove a table from a publication and then re-add it before Materialize notices that the table was removed. In this case, Materialize can no longer provide any consistency guarantees about the data we present from the table and, unfortunately, is wholly unaware that this occurred.
To mitigate this issue, if you need to drop and re-add a table to a
publication, ensure that you remove the table/subsource from the source
before re-adding it using the DROP SOURCE command.
Materialize natively supports the following PostgreSQL types (including the array type for each of the types):
boolbpcharbyteachardatedaterangefloat4float8int2int2vectorint4int4rangeint8int8rangeintervaljsonjsonbnumericnumrangeoidtexttimetimestamptimestamptztsrangetstzrangeuuidvarcharReplicating tables that contain unsupported data types is
possible via the TEXT COLUMNS option. The specified columns will be
treated as text; i.e., will not have the expected PostgreSQL type
features. For example:
enum: When decoded as text, the implicit ordering of the original
PostgreSQL enum type is not preserved; instead, Materialize will sort values
as text.
money: When decoded as text, resulting text value cannot be cast
back to numeric, since PostgreSQL adds typical currency formatting to the
output.
Avoid truncating upstream tables that are being replicated into Materialize. If a replicated upstream table is truncated, the corresponding subsource(s)/table(s) in Materialize becomes inaccessible and will not produce any data until it is recreated.
Instead of truncating, use an unqualified DELETE to remove all rows from
the upstream table:
DELETE FROM t;
When using PostgreSQL table inheritance,
PostgreSQL serves data from SELECTs as if the inheriting tables’ data is
also present in the inherited table. However, both PostgreSQL’s logical
replication and COPY only present data written to the tables themselves,
i.e. the inheriting data is not treated as part of the inherited table.
PostgreSQL sources use logical replication and COPY to ingest table data,
so inheriting tables’ data will only be ingested as part of the inheriting
table, i.e. in Materialize, the data will not be returned when serving
SELECTs from the inherited table.
If using legacy syntax CREATE SOURCE ... FOR ...:
You can mimic PostgreSQL’s SELECT behavior with inherited tables by
creating a materialized view that unions data from the inherited and
inheriting tables (using UNION ALL). However, if new tables inherit from
the table, data from the inheriting tables will not be available in the
view. You will need to add the inheriting tables via ADD SUBSOURCE and
create a new view (materialized or non-) that unions the new table.
If using new CREATE TABLE FROM SOURCE syntax:
You can mimic PostgreSQL’s SELECT behavior with inherited tables by
creating a materialized view that unions data from the inherited and
inheriting tables (using UNION ALL). However, if new tables inherit from
the table, data from the inheriting tables will not be available in the
view. You will need to add the inheriting tables via CREATE TABLE .. FROM SOURCE and create a new view (materialized or non-) that unions the new
table.
Each source ingests the raw replication stream data for all tables in the specified publication using a single replication slot. To manage replication slots:
For PostgreSQL 13+, set a reasonable value
for max_slot_wal_keep_size
to limit the amount of storage used by replication slots.
If you stop using Materialize, or if either the Materialize instance or
the PostgreSQL instance crash, delete any replication slots. You can query
the mz_internal.mz_postgres_sources table to look up the name of the
replication slot created for each source.
If you delete all objects that depend on a source without also dropping
the source, the upstream replication slot remains and will continue to
accumulate data so that the source can resume in the future. To avoid
unbounded disk space usage, make sure to use DROP SOURCE or manually delete the replication slot.
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.
The PostgreSQL source performs parallel snapshotting of tables by distributing rows among
workers using ranges of
CTID.
Materialize uses
PostgreSQL statistics to estimate
the amount of data and number of rows to read. Missing or stale statistics can result in uneven
work distribution, reducing snapshot performance. They can also cause incorrect snapshot
progress reporting in the Console.
To avoid this situation, before creating the source in Materialize, ensure statistics are up to
date by running PostgreSQL ANALYZE command.
Make sure you are running PostgreSQL 11 or higher.
Make sure you have access to your PostgreSQL instance via psql,
or your preferred SQL client.
Once logical replication is enabled, create a publication with the tables that you want to replicate to Materialize. You’ll also need a user for Materialize with sufficient privileges to manage replication.
As a superuser, use psql (or your preferred SQL client) to connect to
your database.
For each table that you want to replicate to Materialize, set the
replica identity
to FULL:
ALTER TABLE <table1> REPLICA IDENTITY FULL;
ALTER TABLE <table2> REPLICA IDENTITY FULL;
REPLICA IDENTITY FULL ensures that the replication stream includes the
previous data of changed rows, in the case of UPDATE and DELETE
operations. This setting enables Materialize to ingest PostgreSQL data with
minimal in-memory state. However, you should expect increased disk usage in
your PostgreSQL database.
Create a publication with the tables you want to replicate:
For specific tables:
CREATE PUBLICATION mz_source FOR TABLE <table1>, <table2>;
For all tables in the database:
CREATE PUBLICATION mz_source FOR ALL TABLES;
The mz_source publication will contain the set of change events generated
from the specified tables, and will later be used to ingest the replication
stream.
Be sure to include only the tables you need. If the publication includes additional tables, Materialize will waste resources on ingesting and then immediately discarding the data.
Create a user for Materialize, if you don’t already have one:
CREATE USER materialize PASSWORD '<password>';
Grant the user permission to manage replication:
GRANT rds_replication TO materialize;
Grant the user the required permissions on the tables you want to replicate:
GRANT CONNECT ON DATABASE <dbname> TO materialize;
GRANT USAGE ON SCHEMA <schema> TO materialize;
GRANT SELECT ON <table1> TO materialize;
GRANT SELECT ON <table2> TO materialize;
Once connected to your database, Materialize will take an initial snapshot
of the tables in your publication. SELECT privileges are required for
this initial snapshot.
If you expect to add tables to your publication, you can grant SELECT on
all tables in the schema instead of naming the specific tables:
GRANT SELECT ON ALL TABLES IN SCHEMA <schema> TO materialize;
Configure your network to allow Materialize to connect to your database. For example, you can:
Allow Materialize IPs: Configure your database’s security group to allow connections from Materialize.
Use an SSH tunnel: Use an SSH tunnel to connect Materialize to the database.
The steps to allow Materialize to connect to your database depends on your deployment setup. Refer to your company’s network/security policies and procedures.
In Materialize, a cluster is an isolated environment, similar to a virtual warehouse in Snowflake. When you create a cluster, you choose the size of its compute resource allocation based on the work you need the cluster to do, whether ingesting data from a source, computing always-up-to-date query results, serving results to external clients, or a combination.
In this step, you’ll create a dedicated cluster for ingesting source data from your PostgreSQL database.
In the SQL Shell, or your preferred SQL
client connected to Materialize, use the CREATE CLUSTER
command to create the new cluster:
CREATE CLUSTER ingest_postgres (SIZE = '50cc');
SET CLUSTER = ingest_postgres;
A cluster of size 50cc should be enough to
accommodate multiple PostgreSQL sources, depending on the source
characteristics (e.g., sources with ENVELOPE UPSERT
or ENVELOPE DEBEZIUM will be more
memory-intensive) and the upstream traffic patterns. You can readjust the
size of the cluster at any time using the ALTER CLUSTER command:
ALTER CLUSTER <cluster_name> SET ( SIZE = <new_size> );
Before it starts consuming the replication stream, Materialize takes a snapshot of the relevant tables in your publication. Until this snapshot is complete, Materialize won’t have the same view of your data as your PostgreSQL database.
In this step, you’ll first verify that the source is running and then check the status of the snapshotting process.
Back in the SQL client connected to Materialize, use the
mz_source_statuses
table to check the overall status of your source:
WITH
source_ids AS
(SELECT id FROM mz_sources WHERE name = 'mz_source')
SELECT *
FROM
mz_internal.mz_source_statuses
JOIN
(
SELECT referenced_object_id
FROM mz_internal.mz_object_dependencies
WHERE
object_id IN (SELECT id FROM source_ids)
UNION SELECT id FROM source_ids
)
AS sources
ON mz_source_statuses.id = sources.referenced_object_id;
For each subsource, make sure the status is running. If you see
stalled or failed, there’s likely a configuration issue for you to fix.
Check the error field for details and fix the issue before moving on.
Also, if the status of any subsource is starting for more than a few
minutes, contact our team.
Once the source is running, use the mz_source_statistics
table to check the status of the initial snapshot:
WITH
source_ids AS
(SELECT id FROM mz_sources WHERE name = 'mz_source')
SELECT sources.referenced_object_id AS id, mz_sources.name, snapshot_committed
FROM
mz_internal.mz_source_statistics
JOIN
(
SELECT object_id, referenced_object_id
FROM mz_internal.mz_object_dependencies
WHERE
object_id IN (SELECT id FROM source_ids)
UNION SELECT id, id FROM source_ids
)
AS sources
ON mz_source_statistics.id = sources.referenced_object_id
JOIN mz_sources ON mz_sources.id = sources.referenced_object_id;
object_id | snapshot_committed
----------|------------------
u144 | t
(1 row)
Once snapshot_commited is t, move on to the next step. Snapshotting can
take between a few minutes to several hours, depending on the size of your
dataset and the size of the cluster the source is running in.
After the snapshotting phase, Materialize starts ingesting change events from
the PostgreSQL replication stream. For this work, Materialize generally
performs well with an 100cc replica, so you can resize the cluster
accordingly.
Still in a SQL client connected to Materialize, use the ALTER CLUSTER
command to downsize the cluster to 100cc:
ALTER CLUSTER ingest_postgres SET (SIZE '100cc');
Behind the scenes, this command adds a new 100cc replica and removes the
50cc replica.
Use the SHOW CLUSTER REPLICAS command to
check the status of the new replica:
SHOW CLUSTER REPLICAS WHERE cluster = 'ingest_postgres';
cluster | replica | size | ready
-----------------+---------+--------+-------
ingest_postgres | r1 | 100cc | t
(1 row)
Going forward, you can verify that your new cluster size is sufficient as follows:
In Materialize, get the replication slot name associated with your
PostgreSQL source from the mz_internal.mz_postgres_sources
table:
SELECT
d.name AS database_name,
n.name AS schema_name,
s.name AS source_name,
pgs.replication_slot
FROM
mz_sources AS s
JOIN mz_internal.mz_postgres_sources AS pgs ON s.id = pgs.id
JOIN mz_schemas AS n ON n.id = s.schema_id
JOIN mz_databases AS d ON d.id = n.database_id;
In PostgreSQL, check the replication slot lag, using the replication slot name from the previous step:
SELECT
pg_size_pretty(pg_current_wal_lsn() - confirmed_flush_lsn)
AS replication_lag_bytes
FROM pg_replication_slots
WHERE slot_name = '<slot_name>';
The result of this query is the amount of data your PostgreSQL cluster must retain in its replication log because of this replication slot. Typically, this means Materialize has not yet communicated back to PostgreSQL that it has committed this data. A high value can indicate that the source has fallen behind and that you might need to scale up your ingestion cluster.
With Materialize ingesting your PostgreSQL data into durable storage, you can start exploring the data, computing real-time results that stay up-to-date as new data arrives, and serving results efficiently.
Explore your data with SHOW SOURCES and SELECT.
Compute real-time results in memory with CREATE VIEW
and CREATE INDEX or in durable
storage with CREATE MATERIALIZED VIEW.
Serve results to a PostgreSQL-compatible SQL client or driver with SELECT
or SUBSCRIBE or to an external message broker with
CREATE SINK.
Check out the tools and integrations supported by Materialize.
Materialize supports schema changes in the upstream database as follows:
Note: This section refer to the legacy
CREATE SOURCE ... FOR ...that creates subsources as part of theCREATE SOURCEoperation. To be able to handle the upstream column additions and drops, seeCREATE SOURCE (New Syntax)andCREATE TABLE FROM SOURCE.
Adding columns to tables. Materialize will not ingest new columns
added upstream unless you use DROP SOURCE to
first drop the affected subsource, and then add the table back to the source
using ALTER SOURCE...ADD SUBSOURCE.
Dropping columns that were added after the source was created. These columns are never ingested, so you can drop them without issue.
Adding or removing NOT NULL constraints to tables that were nullable
when the source was created.
All other schema changes to upstream tables will set the corresponding Materialize tables into an error state, preventing reads from these tables.
To handle incompatible schema changes, drop
the affected table DROP TABLE , and then, CREATE TABLE FROM SOURCE to recreate the table with the
updated schema.
PostgreSQL’s logical replication API does not provide a signal when users remove tables from publications. Because of this, Materialize relies on periodic checks to determine if a table has been removed from a publication, at which time it generates an irrevocable error, preventing any values from being read from the table.
However, it is possible to remove a table from a publication and then re-add it before Materialize notices that the table was removed. In this case, Materialize can no longer provide any consistency guarantees about the data we present from the table and, unfortunately, is wholly unaware that this occurred.
To mitigate this issue, if you need to drop and re-add a table to a
publication, ensure that you remove the table/subsource from the source
before re-adding it using the DROP SOURCE command.
Materialize natively supports the following PostgreSQL types (including the array type for each of the types):
boolbpcharbyteachardatedaterangefloat4float8int2int2vectorint4int4rangeint8int8rangeintervaljsonjsonbnumericnumrangeoidtexttimetimestamptimestamptztsrangetstzrangeuuidvarcharReplicating tables that contain unsupported data types is
possible via the TEXT COLUMNS option. The specified columns will be
treated as text; i.e., will not have the expected PostgreSQL type
features. For example:
enum: When decoded as text, the implicit ordering of the original
PostgreSQL enum type is not preserved; instead, Materialize will sort values
as text.
money: When decoded as text, resulting text value cannot be cast
back to numeric, since PostgreSQL adds typical currency formatting to the
output.
Avoid truncating upstream tables that are being replicated into Materialize. If a replicated upstream table is truncated, the corresponding subsource(s)/table(s) in Materialize becomes inaccessible and will not produce any data until it is recreated.
Instead of truncating, use an unqualified DELETE to remove all rows from
the upstream table:
DELETE FROM t;
When using PostgreSQL table inheritance,
PostgreSQL serves data from SELECTs as if the inheriting tables’ data is
also present in the inherited table. However, both PostgreSQL’s logical
replication and COPY only present data written to the tables themselves,
i.e. the inheriting data is not treated as part of the inherited table.
PostgreSQL sources use logical replication and COPY to ingest table data,
so inheriting tables’ data will only be ingested as part of the inheriting
table, i.e. in Materialize, the data will not be returned when serving
SELECTs from the inherited table.
If using legacy syntax CREATE SOURCE ... FOR ...:
You can mimic PostgreSQL’s SELECT behavior with inherited tables by
creating a materialized view that unions data from the inherited and
inheriting tables (using UNION ALL). However, if new tables inherit from
the table, data from the inheriting tables will not be available in the
view. You will need to add the inheriting tables via ADD SUBSOURCE and
create a new view (materialized or non-) that unions the new table.
If using new CREATE TABLE FROM SOURCE syntax:
You can mimic PostgreSQL’s SELECT behavior with inherited tables by
creating a materialized view that unions data from the inherited and
inheriting tables (using UNION ALL). However, if new tables inherit from
the table, data from the inheriting tables will not be available in the
view. You will need to add the inheriting tables via CREATE TABLE .. FROM SOURCE and create a new view (materialized or non-) that unions the new
table.
Each source ingests the raw replication stream data for all tables in the specified publication using a single replication slot. To manage replication slots:
For PostgreSQL 13+, set a reasonable value
for max_slot_wal_keep_size
to limit the amount of storage used by replication slots.
If you stop using Materialize, or if either the Materialize instance or
the PostgreSQL instance crash, delete any replication slots. You can query
the mz_internal.mz_postgres_sources table to look up the name of the
replication slot created for each source.
If you delete all objects that depend on a source without also dropping
the source, the upstream replication slot remains and will continue to
accumulate data so that the source can resume in the future. To avoid
unbounded disk space usage, make sure to use DROP SOURCE or manually delete the replication slot.
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.
The PostgreSQL source performs parallel snapshotting of tables by distributing rows among
workers using ranges of
CTID.
Materialize uses
PostgreSQL statistics to estimate
the amount of data and number of rows to read. Missing or stale statistics can result in uneven
work distribution, reducing snapshot performance. They can also cause incorrect snapshot
progress reporting in the Console.
To avoid this situation, before creating the source in Materialize, ensure statistics are up to
date by running PostgreSQL ANALYZE command.
Make sure you are running PostgreSQL 11 or higher.
Make sure you have access to your PostgreSQL instance via psql,
or your preferred SQL client.
Once logical replication is enabled, create a publication with the tables that you want to replicate to Materialize. You’ll also need a user for Materialize with sufficient privileges to manage replication.
As a superuser, use psql (or your preferred SQL client) to connect to
your database.
For each table that you want to replicate to Materialize, set the
replica identity
to FULL:
ALTER TABLE <table1> REPLICA IDENTITY FULL;
ALTER TABLE <table2> REPLICA IDENTITY FULL;
REPLICA IDENTITY FULL ensures that the replication stream includes the
previous data of changed rows, in the case of UPDATE and DELETE
operations. This setting enables Materialize to ingest PostgreSQL data with
minimal in-memory state. However, you should expect increased disk usage in
your PostgreSQL database.
Create a publication with the tables you want to replicate:
For specific tables:
CREATE PUBLICATION mz_source FOR TABLE <table1>, <table2>;
For all tables in the database:
CREATE PUBLICATION mz_source FOR ALL TABLES;
The mz_source publication will contain the set of change events generated
from the specified tables, and will later be used to ingest the replication
stream.
Be sure to include only the tables you need. If the publication includes additional tables, Materialize will waste resources on ingesting and then immediately discarding the data.
Create a user for Materialize, if you don’t already have one:
CREATE USER materialize PASSWORD '<password>';
Grant the user permission to manage replication:
GRANT rds_replication TO materialize;
Grant the user the required permissions on the tables you want to replicate:
GRANT CONNECT ON DATABASE <dbname> TO materialize;
GRANT USAGE ON SCHEMA <schema> TO materialize;
GRANT SELECT ON <table1> TO materialize;
GRANT SELECT ON <table2> TO materialize;
Once connected to your database, Materialize will take an initial snapshot
of the tables in your publication. SELECT privileges are required for
this initial snapshot.
If you expect to add tables to your publication, you can grant SELECT on
all tables in the schema instead of naming the specific tables:
GRANT SELECT ON ALL TABLES IN SCHEMA <schema> TO materialize;
Configure your network to allow Materialize to connect to your database. For example, you can:
Allow Materialize IPs: Configure your database’s security group to allow connections from Materialize.
Use an SSH tunnel: Use an SSH tunnel to connect Materialize to the database.
The steps to allow Materialize to connect to your database depends on your deployment setup. Refer to your company’s network/security policies and procedures.
In Materialize, a cluster is an isolated environment, similar to a virtual warehouse in Snowflake. When you create a cluster, you choose the size of its compute resource allocation based on the work you need the cluster to do, whether ingesting data from a source, computing always-up-to-date query results, serving results to external clients, or a combination.
In this step, you’ll create a dedicated cluster for ingesting source data from your PostgreSQL database.
In the SQL Shell, or your preferred SQL
client connected to Materialize, use the CREATE CLUSTER
command to create the new cluster:
CREATE CLUSTER ingest_postgres (SIZE = '50cc');
SET CLUSTER = ingest_postgres;
A cluster of size 50cc should be enough to
accommodate multiple PostgreSQL sources, depending on the source
characteristics (e.g., sources with ENVELOPE UPSERT
or ENVELOPE DEBEZIUM will be more
memory-intensive) and the upstream traffic patterns. You can readjust the
size of the cluster at any time using the ALTER CLUSTER command:
ALTER CLUSTER <cluster_name> SET ( SIZE = <new_size> );
Before it starts consuming the replication stream, Materialize takes a snapshot of the relevant tables in your publication. Until this snapshot is complete, Materialize won’t have the same view of your data as your PostgreSQL database.
In this step, you’ll first verify that the source is running and then check the status of the snapshotting process.
Back in the SQL client connected to Materialize, use the
mz_source_statuses
table to check the overall status of your source:
WITH
source_ids AS
(SELECT id FROM mz_sources WHERE name = 'mz_source')
SELECT *
FROM
mz_internal.mz_source_statuses
JOIN
(
SELECT referenced_object_id
FROM mz_internal.mz_object_dependencies
WHERE
object_id IN (SELECT id FROM source_ids)
UNION SELECT id FROM source_ids
)
AS sources
ON mz_source_statuses.id = sources.referenced_object_id;
For each subsource, make sure the status is running. If you see
stalled or failed, there’s likely a configuration issue for you to fix.
Check the error field for details and fix the issue before moving on.
Also, if the status of any subsource is starting for more than a few
minutes, contact our team.
Once the source is running, use the mz_source_statistics
table to check the status of the initial snapshot:
WITH
source_ids AS
(SELECT id FROM mz_sources WHERE name = 'mz_source')
SELECT sources.referenced_object_id AS id, mz_sources.name, snapshot_committed
FROM
mz_internal.mz_source_statistics
JOIN
(
SELECT object_id, referenced_object_id
FROM mz_internal.mz_object_dependencies
WHERE
object_id IN (SELECT id FROM source_ids)
UNION SELECT id, id FROM source_ids
)
AS sources
ON mz_source_statistics.id = sources.referenced_object_id
JOIN mz_sources ON mz_sources.id = sources.referenced_object_id;
object_id | snapshot_committed
----------|------------------
u144 | t
(1 row)
Once snapshot_commited is t, move on to the next step. Snapshotting can
take between a few minutes to several hours, depending on the size of your
dataset and the size of the cluster the source is running in.
After the snapshotting phase, Materialize starts ingesting change events from
the PostgreSQL replication stream. For this work, Materialize generally
performs well with an 100cc replica, so you can resize the cluster
accordingly.
Still in a SQL client connected to Materialize, use the ALTER CLUSTER
command to downsize the cluster to 100cc:
ALTER CLUSTER ingest_postgres SET (SIZE '100cc');
Behind the scenes, this command adds a new 100cc replica and removes the
50cc replica.
Use the SHOW CLUSTER REPLICAS command to
check the status of the new replica:
SHOW CLUSTER REPLICAS WHERE cluster = 'ingest_postgres';
cluster | replica | size | ready
-----------------+---------+--------+-------
ingest_postgres | r1 | 100cc | t
(1 row)
Going forward, you can verify that your new cluster size is sufficient as follows:
In Materialize, get the replication slot name associated with your
PostgreSQL source from the mz_internal.mz_postgres_sources
table:
SELECT
d.name AS database_name,
n.name AS schema_name,
s.name AS source_name,
pgs.replication_slot
FROM
mz_sources AS s
JOIN mz_internal.mz_postgres_sources AS pgs ON s.id = pgs.id
JOIN mz_schemas AS n ON n.id = s.schema_id
JOIN mz_databases AS d ON d.id = n.database_id;
In PostgreSQL, check the replication slot lag, using the replication slot name from the previous step:
SELECT
pg_size_pretty(pg_current_wal_lsn() - confirmed_flush_lsn)
AS replication_lag_bytes
FROM pg_replication_slots
WHERE slot_name = '<slot_name>';
The result of this query is the amount of data your PostgreSQL cluster must retain in its replication log because of this replication slot. Typically, this means Materialize has not yet communicated back to PostgreSQL that it has committed this data. A high value can indicate that the source has fallen behind and that you might need to scale up your ingestion cluster.
With Materialize ingesting your PostgreSQL data into durable storage, you can start exploring the data, computing real-time results that stay up-to-date as new data arrives, and serving results efficiently.
Explore your data with SHOW SOURCES and SELECT.
Compute real-time results in memory with CREATE VIEW
and CREATE INDEX or in durable
storage with CREATE MATERIALIZED VIEW.
Serve results to a PostgreSQL-compatible SQL client or driver with SELECT
or SUBSCRIBE or to an external message broker with
CREATE SINK.
Check out the tools and integrations supported by Materialize.
Materialize supports schema changes in the upstream database as follows:
Note: This section refer to the legacy
CREATE SOURCE ... FOR ...that creates subsources as part of theCREATE SOURCEoperation. To be able to handle the upstream column additions and drops, seeCREATE SOURCE (New Syntax)andCREATE TABLE FROM SOURCE.
Adding columns to tables. Materialize will not ingest new columns
added upstream unless you use DROP SOURCE to
first drop the affected subsource, and then add the table back to the source
using ALTER SOURCE...ADD SUBSOURCE.
Dropping columns that were added after the source was created. These columns are never ingested, so you can drop them without issue.
Adding or removing NOT NULL constraints to tables that were nullable
when the source was created.
All other schema changes to upstream tables will set the corresponding Materialize tables into an error state, preventing reads from these tables.
To handle incompatible schema changes, drop
the affected table DROP TABLE , and then, CREATE TABLE FROM SOURCE to recreate the table with the
updated schema.
PostgreSQL’s logical replication API does not provide a signal when users remove tables from publications. Because of this, Materialize relies on periodic checks to determine if a table has been removed from a publication, at which time it generates an irrevocable error, preventing any values from being read from the table.
However, it is possible to remove a table from a publication and then re-add it before Materialize notices that the table was removed. In this case, Materialize can no longer provide any consistency guarantees about the data we present from the table and, unfortunately, is wholly unaware that this occurred.
To mitigate this issue, if you need to drop and re-add a table to a
publication, ensure that you remove the table/subsource from the source
before re-adding it using the DROP SOURCE command.
Materialize natively supports the following PostgreSQL types (including the array type for each of the types):
boolbpcharbyteachardatedaterangefloat4float8int2int2vectorint4int4rangeint8int8rangeintervaljsonjsonbnumericnumrangeoidtexttimetimestamptimestamptztsrangetstzrangeuuidvarcharReplicating tables that contain unsupported data types is
possible via the TEXT COLUMNS option. The specified columns will be
treated as text; i.e., will not have the expected PostgreSQL type
features. For example:
enum: When decoded as text, the implicit ordering of the original
PostgreSQL enum type is not preserved; instead, Materialize will sort values
as text.
money: When decoded as text, resulting text value cannot be cast
back to numeric, since PostgreSQL adds typical currency formatting to the
output.
Avoid truncating upstream tables that are being replicated into Materialize. If a replicated upstream table is truncated, the corresponding subsource(s)/table(s) in Materialize becomes inaccessible and will not produce any data until it is recreated.
Instead of truncating, use an unqualified DELETE to remove all rows from
the upstream table:
DELETE FROM t;
When using PostgreSQL table inheritance,
PostgreSQL serves data from SELECTs as if the inheriting tables’ data is
also present in the inherited table. However, both PostgreSQL’s logical
replication and COPY only present data written to the tables themselves,
i.e. the inheriting data is not treated as part of the inherited table.
PostgreSQL sources use logical replication and COPY to ingest table data,
so inheriting tables’ data will only be ingested as part of the inheriting
table, i.e. in Materialize, the data will not be returned when serving
SELECTs from the inherited table.
If using legacy syntax CREATE SOURCE ... FOR ...:
You can mimic PostgreSQL’s SELECT behavior with inherited tables by
creating a materialized view that unions data from the inherited and
inheriting tables (using UNION ALL). However, if new tables inherit from
the table, data from the inheriting tables will not be available in the
view. You will need to add the inheriting tables via ADD SUBSOURCE and
create a new view (materialized or non-) that unions the new table.
If using new CREATE TABLE FROM SOURCE syntax:
You can mimic PostgreSQL’s SELECT behavior with inherited tables by
creating a materialized view that unions data from the inherited and
inheriting tables (using UNION ALL). However, if new tables inherit from
the table, data from the inheriting tables will not be available in the
view. You will need to add the inheriting tables via CREATE TABLE .. FROM SOURCE and create a new view (materialized or non-) that unions the new
table.
Each source ingests the raw replication stream data for all tables in the specified publication using a single replication slot. To manage replication slots:
For PostgreSQL 13+, set a reasonable value
for max_slot_wal_keep_size
to limit the amount of storage used by replication slots.
If you stop using Materialize, or if either the Materialize instance or
the PostgreSQL instance crash, delete any replication slots. You can query
the mz_internal.mz_postgres_sources table to look up the name of the
replication slot created for each source.
If you delete all objects that depend on a source without also dropping
the source, the upstream replication slot remains and will continue to
accumulate data so that the source can resume in the future. To avoid
unbounded disk space usage, make sure to use DROP SOURCE or manually delete the replication slot.
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.
The PostgreSQL source performs parallel snapshotting of tables by distributing rows among
workers using ranges of
CTID.
Materialize uses
PostgreSQL statistics to estimate
the amount of data and number of rows to read. Missing or stale statistics can result in uneven
work distribution, reducing snapshot performance. They can also cause incorrect snapshot
progress reporting in the Console.
To avoid this situation, before creating the source in Materialize, ensure statistics are up to
date by running PostgreSQL ANALYZE command.
Make sure you are running PostgreSQL 11 or higher.
Make sure you have access to your PostgreSQL instance via psql,
or your preferred SQL client.
Once logical replication is enabled, the next step is to create a publication with the tables that you want to replicate to Materialize. You’ll also need a user for Materialize with sufficient privileges to manage replication.
For each table that you want to replicate to Materialize, set the
replica identity
to FULL:
ALTER TABLE <table1> REPLICA IDENTITY FULL;
ALTER TABLE <table2> REPLICA IDENTITY FULL;
REPLICA IDENTITY FULL ensures that the replication stream includes the
previous data of changed rows, in the case of UPDATE and DELETE
operations. This setting enables Materialize to ingest PostgreSQL data with
minimal in-memory state. However, you should expect increased disk usage in
your PostgreSQL database.
Create a publication with the tables you want to replicate:
For specific tables:
CREATE PUBLICATION mz_source FOR TABLE <table1>, <table2>;
For all tables in the database:
CREATE PUBLICATION mz_source FOR ALL TABLES;
The mz_source publication will contain the set of change events generated
from the specified tables, and will later be used to ingest the replication
stream.
Be sure to include only the tables you need. If the publication includes additional tables, Materialize will waste resources on ingesting and then immediately discarding the data.
Create a user for Materialize, if you don’t already have one:
CREATE USER materialize PASSWORD '<password>';
Grant the user permission to manage replication:
ALTER ROLE materialize WITH REPLICATION;
Grant the user the required permissions on the tables you want to replicate:
GRANT CONNECT ON DATABASE <dbname> TO materialize;
GRANT USAGE ON SCHEMA <schema> TO materialize;
GRANT SELECT ON <table1> TO materialize;
GRANT SELECT ON <table2> TO materialize;
Once connected to your database, Materialize will take an initial snapshot
of the tables in your publication. SELECT privileges are required for
this initial snapshot.
If you expect to add tables to your publication, you can grant SELECT on
all tables in the schema instead of naming the specific tables:
GRANT SELECT ON ALL TABLES IN SCHEMA <schema> TO materialize;
Configure your network to allow Materialize to connect to your database. For example, you can:
Allow Materialize IPs: Configure your database’s security group to allow connections from Materialize.
Use an SSH tunnel: Use an SSH tunnel to connect Materialize to the database.
The steps to allow Materialize to connect to your database depends on your deployment setup. Refer to your company’s network/security policies and procedures.
In Materialize, a cluster is an isolated environment, similar to a virtual warehouse in Snowflake. When you create a cluster, you choose the size of its compute resource allocation based on the work you need the cluster to do, whether ingesting data from a source, computing always-up-to-date query results, serving results to external clients, or a combination.
In this step, you’ll create a dedicated cluster for ingesting source data from your PostgreSQL database.
In the SQL Shell, or your preferred SQL
client connected to Materialize, use the CREATE CLUSTER
command to create the new cluster:
CREATE CLUSTER ingest_postgres (SIZE = '50cc');
SET CLUSTER = ingest_postgres;
A cluster of size 50cc should be enough to
accommodate multiple PostgreSQL sources, depending on the source
characteristics (e.g., sources with ENVELOPE UPSERT
or ENVELOPE DEBEZIUM will be more
memory-intensive) and the upstream traffic patterns. You can readjust the
size of the cluster at any time using the ALTER CLUSTER command:
ALTER CLUSTER <cluster_name> SET ( SIZE = <new_size> );
Before it starts consuming the replication stream, Materialize takes a snapshot of the relevant tables in your publication. Until this snapshot is complete, Materialize won’t have the same view of your data as your PostgreSQL database.
In this step, you’ll first verify that the source is running and then check the status of the snapshotting process.
Back in the SQL client connected to Materialize, use the
mz_source_statuses
table to check the overall status of your source:
WITH
source_ids AS
(SELECT id FROM mz_sources WHERE name = 'mz_source')
SELECT *
FROM
mz_internal.mz_source_statuses
JOIN
(
SELECT referenced_object_id
FROM mz_internal.mz_object_dependencies
WHERE
object_id IN (SELECT id FROM source_ids)
UNION SELECT id FROM source_ids
)
AS sources
ON mz_source_statuses.id = sources.referenced_object_id;
For each subsource, make sure the status is running. If you see
stalled or failed, there’s likely a configuration issue for you to fix.
Check the error field for details and fix the issue before moving on.
Also, if the status of any subsource is starting for more than a few
minutes, contact our team.
Once the source is running, use the mz_source_statistics
table to check the status of the initial snapshot:
WITH
source_ids AS
(SELECT id FROM mz_sources WHERE name = 'mz_source')
SELECT sources.referenced_object_id AS id, mz_sources.name, snapshot_committed
FROM
mz_internal.mz_source_statistics
JOIN
(
SELECT object_id, referenced_object_id
FROM mz_internal.mz_object_dependencies
WHERE
object_id IN (SELECT id FROM source_ids)
UNION SELECT id, id FROM source_ids
)
AS sources
ON mz_source_statistics.id = sources.referenced_object_id
JOIN mz_sources ON mz_sources.id = sources.referenced_object_id;
object_id | snapshot_committed
----------|------------------
u144 | t
(1 row)
Once snapshot_commited is t, move on to the next step. Snapshotting can
take between a few minutes to several hours, depending on the size of your
dataset and the size of the cluster the source is running in.
After the snapshotting phase, Materialize starts ingesting change events from
the PostgreSQL replication stream. For this work, Materialize generally
performs well with an 100cc replica, so you can resize the cluster
accordingly.
Still in a SQL client connected to Materialize, use the ALTER CLUSTER
command to downsize the cluster to 100cc:
ALTER CLUSTER ingest_postgres SET (SIZE '100cc');
Behind the scenes, this command adds a new 100cc replica and removes the
50cc replica.
Use the SHOW CLUSTER REPLICAS command to
check the status of the new replica:
SHOW CLUSTER REPLICAS WHERE cluster = 'ingest_postgres';
cluster | replica | size | ready
-----------------+---------+--------+-------
ingest_postgres | r1 | 100cc | t
(1 row)
Going forward, you can verify that your new cluster size is sufficient as follows:
In Materialize, get the replication slot name associated with your
PostgreSQL source from the mz_internal.mz_postgres_sources
table:
SELECT
d.name AS database_name,
n.name AS schema_name,
s.name AS source_name,
pgs.replication_slot
FROM
mz_sources AS s
JOIN mz_internal.mz_postgres_sources AS pgs ON s.id = pgs.id
JOIN mz_schemas AS n ON n.id = s.schema_id
JOIN mz_databases AS d ON d.id = n.database_id;
In PostgreSQL, check the replication slot lag, using the replication slot name from the previous step:
SELECT
pg_size_pretty(pg_current_wal_lsn() - confirmed_flush_lsn)
AS replication_lag_bytes
FROM pg_replication_slots
WHERE slot_name = '<slot_name>';
The result of this query is the amount of data your PostgreSQL cluster must retain in its replication log because of this replication slot. Typically, this means Materialize has not yet communicated back to PostgreSQL that it has committed this data. A high value can indicate that the source has fallen behind and that you might need to scale up your ingestion cluster.
With Materialize ingesting your PostgreSQL data into durable storage, you can start exploring the data, computing real-time results that stay up-to-date as new data arrives, and serving results efficiently.
Explore your data with SHOW SOURCES and SELECT.
Compute real-time results in memory with CREATE VIEW
and CREATE INDEX or in durable
storage with CREATE MATERIALIZED VIEW.
Serve results to a PostgreSQL-compatible SQL client or driver with SELECT
or SUBSCRIBE or to an external message broker with
CREATE SINK.
Check out the tools and integrations supported by Materialize.
Materialize supports schema changes in the upstream database as follows:
Note: This section refer to the legacy
CREATE SOURCE ... FOR ...that creates subsources as part of theCREATE SOURCEoperation. To be able to handle the upstream column additions and drops, seeCREATE SOURCE (New Syntax)andCREATE TABLE FROM SOURCE.
Adding columns to tables. Materialize will not ingest new columns
added upstream unless you use DROP SOURCE to
first drop the affected subsource, and then add the table back to the source
using ALTER SOURCE...ADD SUBSOURCE.
Dropping columns that were added after the source was created. These columns are never ingested, so you can drop them without issue.
Adding or removing NOT NULL constraints to tables that were nullable
when the source was created.
All other schema changes to upstream tables will set the corresponding Materialize tables into an error state, preventing reads from these tables.
To handle incompatible schema changes, drop
the affected table DROP TABLE , and then, CREATE TABLE FROM SOURCE to recreate the table with the
updated schema.
PostgreSQL’s logical replication API does not provide a signal when users remove tables from publications. Because of this, Materialize relies on periodic checks to determine if a table has been removed from a publication, at which time it generates an irrevocable error, preventing any values from being read from the table.
However, it is possible to remove a table from a publication and then re-add it before Materialize notices that the table was removed. In this case, Materialize can no longer provide any consistency guarantees about the data we present from the table and, unfortunately, is wholly unaware that this occurred.
To mitigate this issue, if you need to drop and re-add a table to a
publication, ensure that you remove the table/subsource from the source
before re-adding it using the DROP SOURCE command.
Materialize natively supports the following PostgreSQL types (including the array type for each of the types):
boolbpcharbyteachardatedaterangefloat4float8int2int2vectorint4int4rangeint8int8rangeintervaljsonjsonbnumericnumrangeoidtexttimetimestamptimestamptztsrangetstzrangeuuidvarcharReplicating tables that contain unsupported data types is
possible via the TEXT COLUMNS option. The specified columns will be
treated as text; i.e., will not have the expected PostgreSQL type
features. For example:
enum: When decoded as text, the implicit ordering of the original
PostgreSQL enum type is not preserved; instead, Materialize will sort values
as text.
money: When decoded as text, resulting text value cannot be cast
back to numeric, since PostgreSQL adds typical currency formatting to the
output.
Avoid truncating upstream tables that are being replicated into Materialize. If a replicated upstream table is truncated, the corresponding subsource(s)/table(s) in Materialize becomes inaccessible and will not produce any data until it is recreated.
Instead of truncating, use an unqualified DELETE to remove all rows from
the upstream table:
DELETE FROM t;
When using PostgreSQL table inheritance,
PostgreSQL serves data from SELECTs as if the inheriting tables’ data is
also present in the inherited table. However, both PostgreSQL’s logical
replication and COPY only present data written to the tables themselves,
i.e. the inheriting data is not treated as part of the inherited table.
PostgreSQL sources use logical replication and COPY to ingest table data,
so inheriting tables’ data will only be ingested as part of the inheriting
table, i.e. in Materialize, the data will not be returned when serving
SELECTs from the inherited table.
If using legacy syntax CREATE SOURCE ... FOR ...:
You can mimic PostgreSQL’s SELECT behavior with inherited tables by
creating a materialized view that unions data from the inherited and
inheriting tables (using UNION ALL). However, if new tables inherit from
the table, data from the inheriting tables will not be available in the
view. You will need to add the inheriting tables via ADD SUBSOURCE and
create a new view (materialized or non-) that unions the new table.
If using new CREATE TABLE FROM SOURCE syntax:
You can mimic PostgreSQL’s SELECT behavior with inherited tables by
creating a materialized view that unions data from the inherited and
inheriting tables (using UNION ALL). However, if new tables inherit from
the table, data from the inheriting tables will not be available in the
view. You will need to add the inheriting tables via CREATE TABLE .. FROM SOURCE and create a new view (materialized or non-) that unions the new
table.
Each source ingests the raw replication stream data for all tables in the specified publication using a single replication slot. To manage replication slots:
For PostgreSQL 13+, set a reasonable value
for max_slot_wal_keep_size
to limit the amount of storage used by replication slots.
If you stop using Materialize, or if either the Materialize instance or
the PostgreSQL instance crash, delete any replication slots. You can query
the mz_internal.mz_postgres_sources table to look up the name of the
replication slot created for each source.
If you delete all objects that depend on a source without also dropping
the source, the upstream replication slot remains and will continue to
accumulate data so that the source can resume in the future. To avoid
unbounded disk space usage, make sure to use DROP SOURCE or manually delete the replication slot.
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.
The PostgreSQL source performs parallel snapshotting of tables by distributing rows among
workers using ranges of
CTID.
Materialize uses
PostgreSQL statistics to estimate
the amount of data and number of rows to read. Missing or stale statistics can result in uneven
work distribution, reducing snapshot performance. They can also cause incorrect snapshot
progress reporting in the Console.
To avoid this situation, before creating the source in Materialize, ensure statistics are up to
date by running PostgreSQL ANALYZE command.
Make sure you are running PostgreSQL 11 or higher.
Make sure you have access to your PostgreSQL instance via psql,
or your preferred SQL client.
Once logical replication is enabled, the next step is to create a publication with the tables that you want to replicate to Materialize. You’ll also need a user for Materialize with sufficient privileges to manage replication.
For each table that you want to replicate to Materialize, set the
replica identity
to FULL:
ALTER TABLE <table1> REPLICA IDENTITY FULL;
ALTER TABLE <table2> REPLICA IDENTITY FULL;
REPLICA IDENTITY FULL ensures that the replication stream includes the
previous data of changed rows, in the case of UPDATE and DELETE
operations. This setting enables Materialize to ingest PostgreSQL data with
minimal in-memory state. However, you should expect increased disk usage in
your PostgreSQL database.
Create a publication with the tables you want to replicate:
For specific tables:
CREATE PUBLICATION mz_source FOR TABLE <table1>, <table2>;
For all tables in the database:
CREATE PUBLICATION mz_source FOR ALL TABLES;
The mz_source publication will contain the set of change events generated
from the specified tables, and will later be used to ingest the replication
stream.
Be sure to include only the tables you need. If the publication includes additional tables, Materialize will waste resources on ingesting and then immediately discarding the data.
Create a user for Materialize, if you don’t already have one:
CREATE USER materialize PASSWORD '<password>';
Grant the user permission to manage replication:
ALTER ROLE materialize WITH REPLICATION;
Grant the user the required permissions on the tables you want to replicate:
GRANT CONNECT ON DATABASE <dbname> TO materialize;
GRANT USAGE ON SCHEMA <schema> TO materialize;
GRANT SELECT ON <table1> TO materialize;
GRANT SELECT ON <table2> TO materialize;
Once connected to your database, Materialize will take an initial snapshot
of the tables in your publication. SELECT privileges are required for
this initial snapshot.
If you expect to add tables to your publication, you can grant SELECT on
all tables in the schema instead of naming the specific tables:
GRANT SELECT ON ALL TABLES IN SCHEMA <schema> TO materialize;
Configure your network to allow Materialize to connect to your database. For example, you can:
Allow Materialize IPs: Configure your database’s security group to allow connections from Materialize.
Use an SSH tunnel: Use an SSH tunnel to connect Materialize to the database.
The steps to allow Materialize to connect to your database depends on your deployment setup. Refer to your company’s network/security policies and procedures.
In Materialize, a cluster is an isolated environment, similar to a virtual warehouse in Snowflake. When you create a cluster, you choose the size of its compute resource allocation based on the work you need the cluster to do, whether ingesting data from a source, computing always-up-to-date query results, serving results to external clients, or a combination.
In this step, you’ll create a dedicated cluster for ingesting source data from your PostgreSQL database.
In the SQL Shell, or your preferred SQL
client connected to Materialize, use the CREATE CLUSTER
command to create the new cluster:
CREATE CLUSTER ingest_postgres (SIZE = '50cc');
SET CLUSTER = ingest_postgres;
A cluster of size 50cc should be enough to
accommodate multiple PostgreSQL sources, depending on the source
characteristics (e.g., sources with ENVELOPE UPSERT
or ENVELOPE DEBEZIUM will be more
memory-intensive) and the upstream traffic patterns. You can readjust the
size of the cluster at any time using the ALTER CLUSTER command:
ALTER CLUSTER <cluster_name> SET ( SIZE = <new_size> );
Before it starts consuming the replication stream, Materialize takes a snapshot of the relevant tables in your publication. Until this snapshot is complete, Materialize won’t have the same view of your data as your PostgreSQL database.
In this step, you’ll first verify that the source is running and then check the status of the snapshotting process.
Back in the SQL client connected to Materialize, use the
mz_source_statuses
table to check the overall status of your source:
WITH
source_ids AS
(SELECT id FROM mz_sources WHERE name = 'mz_source')
SELECT *
FROM
mz_internal.mz_source_statuses
JOIN
(
SELECT referenced_object_id
FROM mz_internal.mz_object_dependencies
WHERE
object_id IN (SELECT id FROM source_ids)
UNION SELECT id FROM source_ids
)
AS sources
ON mz_source_statuses.id = sources.referenced_object_id;
For each subsource, make sure the status is running. If you see
stalled or failed, there’s likely a configuration issue for you to fix.
Check the error field for details and fix the issue before moving on.
Also, if the status of any subsource is starting for more than a few
minutes, contact our team.
Once the source is running, use the mz_source_statistics
table to check the status of the initial snapshot:
WITH
source_ids AS
(SELECT id FROM mz_sources WHERE name = 'mz_source')
SELECT sources.referenced_object_id AS id, mz_sources.name, snapshot_committed
FROM
mz_internal.mz_source_statistics
JOIN
(
SELECT object_id, referenced_object_id
FROM mz_internal.mz_object_dependencies
WHERE
object_id IN (SELECT id FROM source_ids)
UNION SELECT id, id FROM source_ids
)
AS sources
ON mz_source_statistics.id = sources.referenced_object_id
JOIN mz_sources ON mz_sources.id = sources.referenced_object_id;
object_id | snapshot_committed
----------|------------------
u144 | t
(1 row)
Once snapshot_commited is t, move on to the next step. Snapshotting can
take between a few minutes to several hours, depending on the size of your
dataset and the size of the cluster the source is running in.
After the snapshotting phase, Materialize starts ingesting change events from
the PostgreSQL replication stream. For this work, Materialize generally
performs well with an 100cc replica, so you can resize the cluster
accordingly.
Still in a SQL client connected to Materialize, use the ALTER CLUSTER
command to downsize the cluster to 100cc:
ALTER CLUSTER ingest_postgres SET (SIZE '100cc');
Behind the scenes, this command adds a new 100cc replica and removes the
50cc replica.
Use the SHOW CLUSTER REPLICAS command to
check the status of the new replica:
SHOW CLUSTER REPLICAS WHERE cluster = 'ingest_postgres';
cluster | replica | size | ready
-----------------+---------+--------+-------
ingest_postgres | r1 | 100cc | t
(1 row)
Going forward, you can verify that your new cluster size is sufficient as follows:
In Materialize, get the replication slot name associated with your
PostgreSQL source from the mz_internal.mz_postgres_sources
table:
SELECT
d.name AS database_name,
n.name AS schema_name,
s.name AS source_name,
pgs.replication_slot
FROM
mz_sources AS s
JOIN mz_internal.mz_postgres_sources AS pgs ON s.id = pgs.id
JOIN mz_schemas AS n ON n.id = s.schema_id
JOIN mz_databases AS d ON d.id = n.database_id;
In PostgreSQL, check the replication slot lag, using the replication slot name from the previous step:
SELECT
pg_size_pretty(pg_current_wal_lsn() - confirmed_flush_lsn)
AS replication_lag_bytes
FROM pg_replication_slots
WHERE slot_name = '<slot_name>';
The result of this query is the amount of data your PostgreSQL cluster must retain in its replication log because of this replication slot. Typically, this means Materialize has not yet communicated back to PostgreSQL that it has committed this data. A high value can indicate that the source has fallen behind and that you might need to scale up your ingestion cluster.
With Materialize ingesting your PostgreSQL data into durable storage, you can start exploring the data, computing real-time results that stay up-to-date as new data arrives, and serving results efficiently.
Explore your data with SHOW SOURCES and SELECT.
Compute real-time results in memory with CREATE VIEW
and CREATE INDEX or in durable
storage with CREATE MATERIALIZED VIEW.
Serve results to a PostgreSQL-compatible SQL client or driver with SELECT
or SUBSCRIBE or to an external message broker with
CREATE SINK.
Check out the tools and integrations supported by Materialize.
Materialize supports schema changes in the upstream database as follows:
Note: This section refer to the legacy
CREATE SOURCE ... FOR ...that creates subsources as part of theCREATE SOURCEoperation. To be able to handle the upstream column additions and drops, seeCREATE SOURCE (New Syntax)andCREATE TABLE FROM SOURCE.
Adding columns to tables. Materialize will not ingest new columns
added upstream unless you use DROP SOURCE to
first drop the affected subsource, and then add the table back to the source
using ALTER SOURCE...ADD SUBSOURCE.
Dropping columns that were added after the source was created. These columns are never ingested, so you can drop them without issue.
Adding or removing NOT NULL constraints to tables that were nullable
when the source was created.
All other schema changes to upstream tables will set the corresponding Materialize tables into an error state, preventing reads from these tables.
To handle incompatible schema changes, drop
the affected table DROP TABLE , and then, CREATE TABLE FROM SOURCE to recreate the table with the
updated schema.
PostgreSQL’s logical replication API does not provide a signal when users remove tables from publications. Because of this, Materialize relies on periodic checks to determine if a table has been removed from a publication, at which time it generates an irrevocable error, preventing any values from being read from the table.
However, it is possible to remove a table from a publication and then re-add it before Materialize notices that the table was removed. In this case, Materialize can no longer provide any consistency guarantees about the data we present from the table and, unfortunately, is wholly unaware that this occurred.
To mitigate this issue, if you need to drop and re-add a table to a
publication, ensure that you remove the table/subsource from the source
before re-adding it using the DROP SOURCE command.
Materialize natively supports the following PostgreSQL types (including the array type for each of the types):
boolbpcharbyteachardatedaterangefloat4float8int2int2vectorint4int4rangeint8int8rangeintervaljsonjsonbnumericnumrangeoidtexttimetimestamptimestamptztsrangetstzrangeuuidvarcharReplicating tables that contain unsupported data types is
possible via the TEXT COLUMNS option. The specified columns will be
treated as text; i.e., will not have the expected PostgreSQL type
features. For example:
enum: When decoded as text, the implicit ordering of the original
PostgreSQL enum type is not preserved; instead, Materialize will sort values
as text.
money: When decoded as text, resulting text value cannot be cast
back to numeric, since PostgreSQL adds typical currency formatting to the
output.
Avoid truncating upstream tables that are being replicated into Materialize. If a replicated upstream table is truncated, the corresponding subsource(s)/table(s) in Materialize becomes inaccessible and will not produce any data until it is recreated.
Instead of truncating, use an unqualified DELETE to remove all rows from
the upstream table:
DELETE FROM t;
When using PostgreSQL table inheritance,
PostgreSQL serves data from SELECTs as if the inheriting tables’ data is
also present in the inherited table. However, both PostgreSQL’s logical
replication and COPY only present data written to the tables themselves,
i.e. the inheriting data is not treated as part of the inherited table.
PostgreSQL sources use logical replication and COPY to ingest table data,
so inheriting tables’ data will only be ingested as part of the inheriting
table, i.e. in Materialize, the data will not be returned when serving
SELECTs from the inherited table.
If using legacy syntax CREATE SOURCE ... FOR ...:
You can mimic PostgreSQL’s SELECT behavior with inherited tables by
creating a materialized view that unions data from the inherited and
inheriting tables (using UNION ALL). However, if new tables inherit from
the table, data from the inheriting tables will not be available in the
view. You will need to add the inheriting tables via ADD SUBSOURCE and
create a new view (materialized or non-) that unions the new table.
If using new CREATE TABLE FROM SOURCE syntax:
You can mimic PostgreSQL’s SELECT behavior with inherited tables by
creating a materialized view that unions data from the inherited and
inheriting tables (using UNION ALL). However, if new tables inherit from
the table, data from the inheriting tables will not be available in the
view. You will need to add the inheriting tables via CREATE TABLE .. FROM SOURCE and create a new view (materialized or non-) that unions the new
table.
Each source ingests the raw replication stream data for all tables in the specified publication using a single replication slot. To manage replication slots:
For PostgreSQL 13+, set a reasonable value
for max_slot_wal_keep_size
to limit the amount of storage used by replication slots.
If you stop using Materialize, or if either the Materialize instance or
the PostgreSQL instance crash, delete any replication slots. You can query
the mz_internal.mz_postgres_sources table to look up the name of the
replication slot created for each source.
If you delete all objects that depend on a source without also dropping
the source, the upstream replication slot remains and will continue to
accumulate data so that the source can resume in the future. To avoid
unbounded disk space usage, make sure to use DROP SOURCE or manually delete the replication slot.
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.
The PostgreSQL source performs parallel snapshotting of tables by distributing rows among
workers using ranges of
CTID.
Materialize uses
PostgreSQL statistics to estimate
the amount of data and number of rows to read. Missing or stale statistics can result in uneven
work distribution, reducing snapshot performance. They can also cause incorrect snapshot
progress reporting in the Console.
To avoid this situation, before creating the source in Materialize, ensure statistics are up to
date by running PostgreSQL ANALYZE command.
In Materialize, a cluster is an isolated environment, similar to a virtual warehouse in Snowflake. When you create a cluster, you choose the size of its compute resource allocation based on the work you need the cluster to do, whether ingesting data from a source, computing always-up-to-date query results, serving results to external clients, or a combination.
In this step, you’ll create a dedicated cluster for ingesting source data from your PostgreSQL database.
In the SQL Shell, or your preferred SQL
client connected to Materialize, use the CREATE CLUSTER
command to create the new cluster:
CREATE CLUSTER ingest_postgres (SIZE = '50cc');
SET CLUSTER = ingest_postgres;
A cluster of size 50cc should be enough to
accommodate multiple PostgreSQL sources, depending on the source
characteristics (e.g., sources with ENVELOPE UPSERT
or ENVELOPE DEBEZIUM will be more
memory-intensive) and the upstream traffic patterns. You can readjust the
size of the cluster at any time using the ALTER CLUSTER command:
ALTER CLUSTER <cluster_name> SET ( SIZE = <new_size> );
Before it starts consuming the replication stream, Materialize takes a snapshot of the relevant tables in your publication. Until this snapshot is complete, Materialize won’t have the same view of your data as your PostgreSQL database.
In this step, you’ll first verify that the source is running and then check the status of the snapshotting process.
Back in the SQL client connected to Materialize, use the
mz_source_statuses
table to check the overall status of your source:
WITH
source_ids AS
(SELECT id FROM mz_sources WHERE name = 'mz_source')
SELECT *
FROM
mz_internal.mz_source_statuses
JOIN
(
SELECT referenced_object_id
FROM mz_internal.mz_object_dependencies
WHERE
object_id IN (SELECT id FROM source_ids)
UNION SELECT id FROM source_ids
)
AS sources
ON mz_source_statuses.id = sources.referenced_object_id;
For each subsource, make sure the status is running. If you see
stalled or failed, there’s likely a configuration issue for you to fix.
Check the error field for details and fix the issue before moving on.
Also, if the status of any subsource is starting for more than a few
minutes, contact our team.
Once the source is running, use the mz_source_statistics
table to check the status of the initial snapshot:
WITH
source_ids AS
(SELECT id FROM mz_sources WHERE name = 'mz_source')
SELECT sources.referenced_object_id AS id, mz_sources.name, snapshot_committed
FROM
mz_internal.mz_source_statistics
JOIN
(
SELECT object_id, referenced_object_id
FROM mz_internal.mz_object_dependencies
WHERE
object_id IN (SELECT id FROM source_ids)
UNION SELECT id, id FROM source_ids
)
AS sources
ON mz_source_statistics.id = sources.referenced_object_id
JOIN mz_sources ON mz_sources.id = sources.referenced_object_id;
object_id | snapshot_committed
----------|------------------
u144 | t
(1 row)
Once snapshot_commited is t, move on to the next step. Snapshotting can
take between a few minutes to several hours, depending on the size of your
dataset and the size of the cluster the source is running in.
After the snapshotting phase, Materialize starts ingesting change events from
the PostgreSQL replication stream. For this work, Materialize generally
performs well with an 100cc replica, so you can resize the cluster
accordingly.
Still in a SQL client connected to Materialize, use the ALTER CLUSTER
command to downsize the cluster to 100cc:
ALTER CLUSTER ingest_postgres SET (SIZE '100cc');
Behind the scenes, this command adds a new 100cc replica and removes the
50cc replica.
Use the SHOW CLUSTER REPLICAS command to
check the status of the new replica:
SHOW CLUSTER REPLICAS WHERE cluster = 'ingest_postgres';
cluster | replica | size | ready
-----------------+---------+--------+-------
ingest_postgres | r1 | 100cc | t
(1 row)
Going forward, you can verify that your new cluster size is sufficient as follows:
In Materialize, get the replication slot name associated with your
PostgreSQL source from the mz_internal.mz_postgres_sources
table:
SELECT
d.name AS database_name,
n.name AS schema_name,
s.name AS source_name,
pgs.replication_slot
FROM
mz_sources AS s
JOIN mz_internal.mz_postgres_sources AS pgs ON s.id = pgs.id
JOIN mz_schemas AS n ON n.id = s.schema_id
JOIN mz_databases AS d ON d.id = n.database_id;
In PostgreSQL, check the replication slot lag, using the replication slot name from the previous step:
SELECT
pg_size_pretty(pg_current_wal_lsn() - confirmed_flush_lsn)
AS replication_lag_bytes
FROM pg_replication_slots
WHERE slot_name = '<slot_name>';
The result of this query is the amount of data your PostgreSQL cluster must retain in its replication log because of this replication slot. Typically, this means Materialize has not yet communicated back to PostgreSQL that it has committed this data. A high value can indicate that the source has fallen behind and that you might need to scale up your ingestion cluster.
With Materialize ingesting your PostgreSQL data into durable storage, you can start exploring the data, computing real-time results that stay up-to-date as new data arrives, and serving results efficiently.
Explore your data with SHOW SOURCES and SELECT.
Compute real-time results in memory with CREATE VIEW
and CREATE INDEX or in durable
storage with CREATE MATERIALIZED VIEW.
Serve results to a PostgreSQL-compatible SQL client or driver with SELECT
or SUBSCRIBE or to an external message broker with
CREATE SINK.
Check out the tools and integrations supported by Materialize.
Materialize supports schema changes in the upstream database as follows:
Note: This section refer to the legacy
CREATE SOURCE ... FOR ...that creates subsources as part of theCREATE SOURCEoperation. To be able to handle the upstream column additions and drops, seeCREATE SOURCE (New Syntax)andCREATE TABLE FROM SOURCE.
Adding columns to tables. Materialize will not ingest new columns
added upstream unless you use DROP SOURCE to
first drop the affected subsource, and then add the table back to the source
using ALTER SOURCE...ADD SUBSOURCE.
Dropping columns that were added after the source was created. These columns are never ingested, so you can drop them without issue.
Adding or removing NOT NULL constraints to tables that were nullable
when the source was created.
All other schema changes to upstream tables will set the corresponding Materialize tables into an error state, preventing reads from these tables.
To handle incompatible schema changes, drop
the affected table DROP TABLE , and then, CREATE TABLE FROM SOURCE to recreate the table with the
updated schema.
PostgreSQL’s logical replication API does not provide a signal when users remove tables from publications. Because of this, Materialize relies on periodic checks to determine if a table has been removed from a publication, at which time it generates an irrevocable error, preventing any values from being read from the table.
However, it is possible to remove a table from a publication and then re-add it before Materialize notices that the table was removed. In this case, Materialize can no longer provide any consistency guarantees about the data we present from the table and, unfortunately, is wholly unaware that this occurred.
To mitigate this issue, if you need to drop and re-add a table to a
publication, ensure that you remove the table/subsource from the source
before re-adding it using the DROP SOURCE command.
Materialize natively supports the following PostgreSQL types (including the array type for each of the types):
boolbpcharbyteachardatedaterangefloat4float8int2int2vectorint4int4rangeint8int8rangeintervaljsonjsonbnumericnumrangeoidtexttimetimestamptimestamptztsrangetstzrangeuuidvarcharReplicating tables that contain unsupported data types is
possible via the TEXT COLUMNS option. The specified columns will be
treated as text; i.e., will not have the expected PostgreSQL type
features. For example:
enum: When decoded as text, the implicit ordering of the original
PostgreSQL enum type is not preserved; instead, Materialize will sort values
as text.
money: When decoded as text, resulting text value cannot be cast
back to numeric, since PostgreSQL adds typical currency formatting to the
output.
Avoid truncating upstream tables that are being replicated into Materialize. If a replicated upstream table is truncated, the corresponding subsource(s)/table(s) in Materialize becomes inaccessible and will not produce any data until it is recreated.
Instead of truncating, use an unqualified DELETE to remove all rows from
the upstream table:
DELETE FROM t;
When using PostgreSQL table inheritance,
PostgreSQL serves data from SELECTs as if the inheriting tables’ data is
also present in the inherited table. However, both PostgreSQL’s logical
replication and COPY only present data written to the tables themselves,
i.e. the inheriting data is not treated as part of the inherited table.
PostgreSQL sources use logical replication and COPY to ingest table data,
so inheriting tables’ data will only be ingested as part of the inheriting
table, i.e. in Materialize, the data will not be returned when serving
SELECTs from the inherited table.
If using legacy syntax CREATE SOURCE ... FOR ...:
You can mimic PostgreSQL’s SELECT behavior with inherited tables by
creating a materialized view that unions data from the inherited and
inheriting tables (using UNION ALL). However, if new tables inherit from
the table, data from the inheriting tables will not be available in the
view. You will need to add the inheriting tables via ADD SUBSOURCE and
create a new view (materialized or non-) that unions the new table.
If using new CREATE TABLE FROM SOURCE syntax:
You can mimic PostgreSQL’s SELECT behavior with inherited tables by
creating a materialized view that unions data from the inherited and
inheriting tables (using UNION ALL). However, if new tables inherit from
the table, data from the inheriting tables will not be available in the
view. You will need to add the inheriting tables via CREATE TABLE .. FROM SOURCE and create a new view (materialized or non-) that unions the new
table.
Each source ingests the raw replication stream data for all tables in the specified publication using a single replication slot. To manage replication slots:
For PostgreSQL 13+, set a reasonable value
for max_slot_wal_keep_size
to limit the amount of storage used by replication slots.
If you stop using Materialize, or if either the Materialize instance or
the PostgreSQL instance crash, delete any replication slots. You can query
the mz_internal.mz_postgres_sources table to look up the name of the
replication slot created for each source.
If you delete all objects that depend on a source without also dropping
the source, the upstream replication slot remains and will continue to
accumulate data so that the source can resume in the future. To avoid
unbounded disk space usage, make sure to use DROP SOURCE or manually delete the replication slot.
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.
The PostgreSQL source performs parallel snapshotting of tables by distributing rows among
workers using ranges of
CTID.
Materialize uses
PostgreSQL statistics to estimate
the amount of data and number of rows to read. Missing or stale statistics can result in uneven
work distribution, reducing snapshot performance. They can also cause incorrect snapshot
progress reporting in the Console.
To avoid this situation, before creating the source in Materialize, ensure statistics are up to
date by running PostgreSQL ANALYZE command.
Make sure you are running PostgreSQL 11 or higher.
Make sure you have access to your PostgreSQL instance via psql,
or your preferred SQL client.
Once logical replication is enabled, the next step is to create a publication with the tables that you want to replicate to Materialize. You’ll also need a user for Materialize with sufficient privileges to manage replication.
For each table that you want to replicate to Materialize, set the
replica identity
to FULL:
ALTER TABLE <table1> REPLICA IDENTITY FULL;
ALTER TABLE <table2> REPLICA IDENTITY FULL;
REPLICA IDENTITY FULL ensures that the replication stream includes the
previous data of changed rows, in the case of UPDATE and DELETE
operations. This setting enables Materialize to ingest PostgreSQL data with
minimal in-memory state. However, you should expect increased disk usage in
your PostgreSQL database.
Create a publication with the tables you want to replicate:
For specific tables:
CREATE PUBLICATION mz_source FOR TABLE <table1>, <table2>;
For all tables in the database:
CREATE PUBLICATION mz_source FOR ALL TABLES;
The mz_source publication will contain the set of change events generated
from the specified tables, and will later be used to ingest the replication
stream.
Be sure to include only the tables you need. If the publication includes additional tables, Materialize will waste resources on ingesting and then immediately discarding the data.
Create a user for Materialize, if you don’t already have one:
CREATE USER materialize PASSWORD '<password>';
Grant the user permission to manage replication:
ALTER ROLE materialize WITH REPLICATION;
Grant the user the required permissions on the tables you want to replicate:
GRANT CONNECT ON DATABASE <dbname> TO materialize;
GRANT USAGE ON SCHEMA <schema> TO materialize;
GRANT SELECT ON <table1> TO materialize;
GRANT SELECT ON <table2> TO materialize;
Once connected to your database, Materialize will take an initial snapshot
of the tables in your publication. SELECT privileges are required for
this initial snapshot.
If you expect to add tables to your publication, you can grant SELECT on
all tables in the schema instead of naming the specific tables:
GRANT SELECT ON ALL TABLES IN SCHEMA <schema> TO materialize;
Configure your network to allow Materialize to connect to your database. For example, you can:
Allow Materialize IPs: Configure your database’s security group to allow connections from Materialize.
Use an SSH tunnel: Use an SSH tunnel to connect Materialize to the database.
The steps to allow Materialize to connect to your database depends on your deployment setup. Refer to your company’s network/security policies and procedures.
In Materialize, a cluster is an isolated environment, similar to a virtual warehouse in Snowflake. When you create a cluster, you choose the size of its compute resource allocation based on the work you need the cluster to do, whether ingesting data from a source, computing always-up-to-date query results, serving results to external clients, or a combination.
In this step, you’ll create a dedicated cluster for ingesting source data from your PostgreSQL database.
In the SQL Shell, or your preferred SQL
client connected to Materialize, use the CREATE CLUSTER
command to create the new cluster:
CREATE CLUSTER ingest_postgres (SIZE = '50cc');
SET CLUSTER = ingest_postgres;
A cluster of size 50cc should be enough to
accommodate multiple PostgreSQL sources, depending on the source
characteristics (e.g., sources with ENVELOPE UPSERT
or ENVELOPE DEBEZIUM will be more
memory-intensive) and the upstream traffic patterns. You can readjust the
size of the cluster at any time using the ALTER CLUSTER command:
ALTER CLUSTER <cluster_name> SET ( SIZE = <new_size> );
Before it starts consuming the replication stream, Materialize takes a snapshot of the relevant tables in your publication. Until this snapshot is complete, Materialize won’t have the same view of your data as your PostgreSQL database.
In this step, you’ll first verify that the source is running and then check the status of the snapshotting process.
Back in the SQL client connected to Materialize, use the
mz_source_statuses
table to check the overall status of your source:
WITH
source_ids AS
(SELECT id FROM mz_sources WHERE name = 'mz_source')
SELECT *
FROM
mz_internal.mz_source_statuses
JOIN
(
SELECT referenced_object_id
FROM mz_internal.mz_object_dependencies
WHERE
object_id IN (SELECT id FROM source_ids)
UNION SELECT id FROM source_ids
)
AS sources
ON mz_source_statuses.id = sources.referenced_object_id;
For each subsource, make sure the status is running. If you see
stalled or failed, there’s likely a configuration issue for you to fix.
Check the error field for details and fix the issue before moving on.
Also, if the status of any subsource is starting for more than a few
minutes, contact our team.
Once the source is running, use the mz_source_statistics
table to check the status of the initial snapshot:
WITH
source_ids AS
(SELECT id FROM mz_sources WHERE name = 'mz_source')
SELECT sources.referenced_object_id AS id, mz_sources.name, snapshot_committed
FROM
mz_internal.mz_source_statistics
JOIN
(
SELECT object_id, referenced_object_id
FROM mz_internal.mz_object_dependencies
WHERE
object_id IN (SELECT id FROM source_ids)
UNION SELECT id, id FROM source_ids
)
AS sources
ON mz_source_statistics.id = sources.referenced_object_id
JOIN mz_sources ON mz_sources.id = sources.referenced_object_id;
object_id | snapshot_committed
----------|------------------
u144 | t
(1 row)
Once snapshot_commited is t, move on to the next step. Snapshotting can
take between a few minutes to several hours, depending on the size of your
dataset and the size of the cluster the source is running in.
After the snapshotting phase, Materialize starts ingesting change events from
the PostgreSQL replication stream. For this work, Materialize generally
performs well with an 100cc replica, so you can resize the cluster
accordingly.
Still in a SQL client connected to Materialize, use the ALTER CLUSTER
command to downsize the cluster to 100cc:
ALTER CLUSTER ingest_postgres SET (SIZE '100cc');
Behind the scenes, this command adds a new 100cc replica and removes the
50cc replica.
Use the SHOW CLUSTER REPLICAS command to
check the status of the new replica:
SHOW CLUSTER REPLICAS WHERE cluster = 'ingest_postgres';
cluster | replica | size | ready
-----------------+---------+--------+-------
ingest_postgres | r1 | 100cc | t
(1 row)
Going forward, you can verify that your new cluster size is sufficient as follows:
In Materialize, get the replication slot name associated with your
PostgreSQL source from the mz_internal.mz_postgres_sources
table:
SELECT
d.name AS database_name,
n.name AS schema_name,
s.name AS source_name,
pgs.replication_slot
FROM
mz_sources AS s
JOIN mz_internal.mz_postgres_sources AS pgs ON s.id = pgs.id
JOIN mz_schemas AS n ON n.id = s.schema_id
JOIN mz_databases AS d ON d.id = n.database_id;
In PostgreSQL, check the replication slot lag, using the replication slot name from the previous step:
SELECT
pg_size_pretty(pg_current_wal_lsn() - confirmed_flush_lsn)
AS replication_lag_bytes
FROM pg_replication_slots
WHERE slot_name = '<slot_name>';
The result of this query is the amount of data your PostgreSQL cluster must retain in its replication log because of this replication slot. Typically, this means Materialize has not yet communicated back to PostgreSQL that it has committed this data. A high value can indicate that the source has fallen behind and that you might need to scale up your ingestion cluster.
With Materialize ingesting your PostgreSQL data into durable storage, you can start exploring the data, computing real-time results that stay up-to-date as new data arrives, and serving results efficiently.
Explore your data with SHOW SOURCES and SELECT.
Compute real-time results in memory with CREATE VIEW
and CREATE INDEX or in durable
storage with CREATE MATERIALIZED VIEW.
Serve results to a PostgreSQL-compatible SQL client or driver with SELECT
or SUBSCRIBE or to an external message broker with
CREATE SINK.
Check out the tools and integrations supported by Materialize.
Materialize supports schema changes in the upstream database as follows:
Note: This section refer to the legacy
CREATE SOURCE ... FOR ...that creates subsources as part of theCREATE SOURCEoperation. To be able to handle the upstream column additions and drops, seeCREATE SOURCE (New Syntax)andCREATE TABLE FROM SOURCE.
Adding columns to tables. Materialize will not ingest new columns
added upstream unless you use DROP SOURCE to
first drop the affected subsource, and then add the table back to the source
using ALTER SOURCE...ADD SUBSOURCE.
Dropping columns that were added after the source was created. These columns are never ingested, so you can drop them without issue.
Adding or removing NOT NULL constraints to tables that were nullable
when the source was created.
All other schema changes to upstream tables will set the corresponding Materialize tables into an error state, preventing reads from these tables.
To handle incompatible schema changes, drop
the affected table DROP TABLE , and then, CREATE TABLE FROM SOURCE to recreate the table with the
updated schema.
PostgreSQL’s logical replication API does not provide a signal when users remove tables from publications. Because of this, Materialize relies on periodic checks to determine if a table has been removed from a publication, at which time it generates an irrevocable error, preventing any values from being read from the table.
However, it is possible to remove a table from a publication and then re-add it before Materialize notices that the table was removed. In this case, Materialize can no longer provide any consistency guarantees about the data we present from the table and, unfortunately, is wholly unaware that this occurred.
To mitigate this issue, if you need to drop and re-add a table to a
publication, ensure that you remove the table/subsource from the source
before re-adding it using the DROP SOURCE command.
Materialize natively supports the following PostgreSQL types (including the array type for each of the types):
boolbpcharbyteachardatedaterangefloat4float8int2int2vectorint4int4rangeint8int8rangeintervaljsonjsonbnumericnumrangeoidtexttimetimestamptimestamptztsrangetstzrangeuuidvarcharReplicating tables that contain unsupported data types is
possible via the TEXT COLUMNS option. The specified columns will be
treated as text; i.e., will not have the expected PostgreSQL type
features. For example:
enum: When decoded as text, the implicit ordering of the original
PostgreSQL enum type is not preserved; instead, Materialize will sort values
as text.
money: When decoded as text, resulting text value cannot be cast
back to numeric, since PostgreSQL adds typical currency formatting to the
output.
Avoid truncating upstream tables that are being replicated into Materialize. If a replicated upstream table is truncated, the corresponding subsource(s)/table(s) in Materialize becomes inaccessible and will not produce any data until it is recreated.
Instead of truncating, use an unqualified DELETE to remove all rows from
the upstream table:
DELETE FROM t;
When using PostgreSQL table inheritance,
PostgreSQL serves data from SELECTs as if the inheriting tables’ data is
also present in the inherited table. However, both PostgreSQL’s logical
replication and COPY only present data written to the tables themselves,
i.e. the inheriting data is not treated as part of the inherited table.
PostgreSQL sources use logical replication and COPY to ingest table data,
so inheriting tables’ data will only be ingested as part of the inheriting
table, i.e. in Materialize, the data will not be returned when serving
SELECTs from the inherited table.
If using legacy syntax CREATE SOURCE ... FOR ...:
You can mimic PostgreSQL’s SELECT behavior with inherited tables by
creating a materialized view that unions data from the inherited and
inheriting tables (using UNION ALL). However, if new tables inherit from
the table, data from the inheriting tables will not be available in the
view. You will need to add the inheriting tables via ADD SUBSOURCE and
create a new view (materialized or non-) that unions the new table.
If using new CREATE TABLE FROM SOURCE syntax:
You can mimic PostgreSQL’s SELECT behavior with inherited tables by
creating a materialized view that unions data from the inherited and
inheriting tables (using UNION ALL). However, if new tables inherit from
the table, data from the inheriting tables will not be available in the
view. You will need to add the inheriting tables via CREATE TABLE .. FROM SOURCE and create a new view (materialized or non-) that unions the new
table.
Each source ingests the raw replication stream data for all tables in the specified publication using a single replication slot. To manage replication slots:
For PostgreSQL 13+, set a reasonable value
for max_slot_wal_keep_size
to limit the amount of storage used by replication slots.
If you stop using Materialize, or if either the Materialize instance or
the PostgreSQL instance crash, delete any replication slots. You can query
the mz_internal.mz_postgres_sources table to look up the name of the
replication slot created for each source.
If you delete all objects that depend on a source without also dropping
the source, the upstream replication slot remains and will continue to
accumulate data so that the source can resume in the future. To avoid
unbounded disk space usage, make sure to use DROP SOURCE or manually delete the replication slot.
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.
The PostgreSQL source performs parallel snapshotting of tables by distributing rows among
workers using ranges of
CTID.
Materialize uses
PostgreSQL statistics to estimate
the amount of data and number of rows to read. Missing or stale statistics can result in uneven
work distribution, reducing snapshot performance. They can also cause incorrect snapshot
progress reporting in the Console.
To avoid this situation, before creating the source in Materialize, ensure statistics are up to
date by running PostgreSQL ANALYZE command.