# CREATE SOURCE: PostgreSQL (New Syntax) Creates a new source from PostgreSQL 11+. > **Disambiguation:** This page reflects the new syntax which allows Materialize to handle upstream DDL changes, specifically adding or dropping columns, without downtime. For the deprecated syntax, see the [old reference page](/sql/create-source/postgres/). Creates a new source from PostgreSQL. Materialize supports creating sources from PostgreSQL version 11+. Once a new source is created, you can CREATE TABLE FROM SOURCE to create the corresponding tables in Materialize and start the data ingestion process. PostgreSQL 16+ is required for connecting Materialize to a physical replica. ## Prerequisites To create a source from PostgreSQL 11+, you must first: - **Configure upstream PostgreSQL instance** - Set up logical replication. - Create a publication. - Create a replication user and password for Materialize to use to connect. - **Configure network security** - Ensure Materialize can connect to your PostgreSQL instance. - **Create a connection to PostgreSQL in Materialize** - The connection setup depends on the network security configuration. For details, see the [PostgreSQL integration guides](/ingest-data/postgres/#integration-guides). ## Syntax To create a source from an external PostgreSQL: ```mzsql CREATE SOURCE [IF NOT EXISTS] [IN CLUSTER ] FROM POSTGRES CONNECTION (PUBLICATION '') [WITH ( [, ...] )] ; ``` | Syntax element | Description | | --- | --- | | **IF NOT EXISTS** | *Optional.* If specified, do not throw an error if a source with the same name already exists. Instead, issue a notice and skip the source creation. | | `` | The name of the source to create. Names for sources must follow the [naming guidelines](/sql/identifiers/#naming-restrictions). | | **IN CLUSTER** `` | *Optional.* The [cluster](/sql/create-cluster) to maintain this source. Otherwise, the source will be created in the active cluster. {{< tip >}} If possible, use a cluster dedicated just for sources. See also [Operational guidelines](/clusters/operational-guidelines/#sources). {{< /tip >}} | | `` | The name of the PostgreSQL connection to use for the source. For details on creating connections, check the [`CREATE CONNECTION`](/sql/create-connection/#postgresql) documentation page. A connection is **reusable** across multiple `CREATE SOURCE` statements. | | `` | The name of the PostgreSQL publication to associate with the source. For details on creating a publication in your PostgreSQL database, see the [integration guides for your PostgreSQL](/ingest-data/postgres/#integration-guides). | | **WITH** (`` [, ...]) | *Optional.* The following ``s are supported: \| Option \| Description \| \|--------\|-------------\| \| `TIMESTAMP INTERVAL [=] ` \| The interval at which timestamps are assigned to data read from this source. Accepts positive [interval](/sql/types/interval/) values (e.g. `'500ms'`, `'1s'`). The value must be between the system parameters `min_timestamp_interval` and `max_timestamp_interval`. Default: the value of the `default_timestamp_interval` system parameter (`1s`). The interval can also be changed after creation with [`ALTER SOURCE`](/sql/alter-source/). \| | ## Details ### Ingesting data After a source is created, you can create tables from the source, referencing the tables in the publication, to start ingesting data. You can create multiple tables that reference the same table in the publication. See [`CREATE TABLE FROM SOURCE`](/sql/create-table/) for details. #### Handling table schema changes The use of the `CREATE SOURCE` with the new [`CREATE TABLE FROM SOURCE`](/sql/create-table/) allows for the handling of certain upstream DDL changes without downtime. See [`CREATE TABLE FROM SOURCE`](/sql/create-table/postgres/#handling-table-schema-changes) for details. See also [Handling upstream operations](#handling-upstream-operations) for additional upstream operation considerations. #### Supported types With the new syntax, after a PostgreSQL source is created, you [`CREATE TABLE FROM SOURCE`](/sql/create-table/) to create a corresponding table in Materialize and start ingesting data.

Materialize natively supports the following PostgreSQL types (including the array type for each of the types):

  • bool
  • bpchar
  • bytea
  • char
  • date
  • daterange
  • float4
  • float8
  • int2
  • int2vector
  • int4
  • int4range
  • int8
  • int8range
  • interval
  • json
  • jsonb
  • numeric
  • numrange
  • oid
  • text
  • time
  • timestamp
  • timestamptz
  • tsrange
  • tstzrange
  • uuid
  • varchar
For more information, including strategies for handling unsupported types, see [`CREATE TABLE FROM SOURCE`](/sql/create-table/). ### Publication membership 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`](/sql/drop-source/) command. ### PostgreSQL replication slots When you define a source, Materialize will automatically create a **replication slot** in the upstream PostgreSQL database (see [PostgreSQL replication slots](#postgresql-replication-slots)). Each source ingests the raw replication stream data for all tables in the specified publication using **a single** replication slot. This allows you to minimize the performance impact on the upstream database as well as reuse the same source across multiple materializations. The name of the replication slot created by Materialize is prefixed with `materialize_`. In Materialize, you can query the `mz_internal.mz_postgres_sources` to find the replication slots created: ```mzsql SELECT id, replication_slot FROM mz_internal.mz_postgres_sources; ``` ``` id | replication_slot ---------+---------------------------------------------- u8 | materialize_7f8a72d0bf2a4b6e9ebc4e61ba769b71 ``` > **Tip:** - For PostgreSQL 13+, set a reasonable value > for [`max_slot_wal_keep_size`](https://www.postgresql.org/docs/13/runtime-config-replication.html#GUC-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`](/sql/drop-source/) or manually delete the replication slot. ### Reading from a physical standby Materialize can replicate from a PostgreSQL physical standby (read replica) instead of the primary, using logical decoding on the standby. This requires **PostgreSQL 16+** on both the primary and the standby, since earlier versions do not support creating logical replication slots on a standby. When the upstream is a standby, the replication slot is created on the standby and Materialize only connects to the standby. Note that slot creation on a standby can block until the primary emits a standby snapshot (a `RUNNING_XACTS` WAL record). On an idle primary, run [`SELECT pg_log_standby_snapshot()`](https://www.postgresql.org/docs/16/functions-admin.html#FUNCTIONS-SNAPSHOT-SYNCHRONIZATION) on the primary to unblock source creation. ## Handling upstream operations This section describes how changes to upstream tables that Materialize ingests affect the corresponding Materialize tables. ### Adding a column When you add a new column to your upstream table, Materialize continues to ingest only the existing columns. To incorporate the new column: - If using the new [`CREATE SOURCE` and `CREATE TABLE FROM SOURCE`](/sql/create-source/postgres-v2/) syntax, create a new table from the source. See [Handle upstream column addition](/ingest-data/postgres/source-versioning/#handle-upstream-column-addition). - If using the legacy [`CREATE SOURCE ... FOR ...`](/sql/create-source/postgres/) syntax that creates subsources, use [`DROP SOURCE`](/sql/drop-source/) to drop the affected subsource, and then add the table back to the source using [`ALTER SOURCE ... ADD SUBSOURCE`](/sql/alter-source/). The re-added subsource includes the new column. ### Dropping a column Dropping columns that Materialize does not ingest (for example, columns added after the source was created, or columns that are excluded) is supported. As these columns were never ingested, you can drop them without issue. If your Materialize source ingests a column, dropping that column from your upstream table puts the affected table into an error state. - If using the new [`CREATE SOURCE` and `CREATE TABLE FROM SOURCE`](/sql/create-source/postgres-v2/) syntax, you can safely drop a column by first ignoring it in Materialize. See [Handle upstream column drop](/ingest-data/postgres/source-versioning/#handle-upstream-column-drop). - If using legacy [`CREATE SOURCE ... FOR ...`](/sql/create-source/postgres/) syntax, use [`DROP SOURCE`](/sql/drop-source/) to drop the affected subsource, and then add the table back to the source using [`ALTER SOURCE ... ADD SUBSOURCE`](/sql/alter-source/). ### Changing constraints Materialize ignores the following constraint changes: foreign key, `CHECK`, and `EXCLUSION`. As such, you can add or drop them without affecting ingestion. Materialize also ignores `NOT NULL`, `UNIQUE`, and `PRIMARY KEY` constraints that are added after the Materialize table is created (that is, the table was created without them). Adding such a constraint, and later dropping it, does not affect ingestion. Dropping a `NOT NULL`, `UNIQUE`, or `PRIMARY KEY` constraint that existed when the table was created puts the affected table into an error state. If using the new [`CREATE SOURCE` and `CREATE TABLE FROM SOURCE`](/sql/create-source/postgres-v2/) syntax, you can safely drop such a constraint by first excluding it in Materialize. See [Handle upstream constraint drop](/ingest-data/postgres/source-versioning/#handle-upstream-constraint-drop). ### Changing a column's data type Changing an ingested column's data type upstream puts the affected Materialize table into an error state unless the column was ingested as `text` via the `TEXT COLUMNS` option. Ingestion for that table stops, and you must drop and recreate the table in Materialize to resume ingestion. ### Renaming a column Renaming a column that Materialize ingests puts the affected table into an error state. Ingestion for that table stops, and you must drop and recreate the table in Materialize to resume ingestion. ### Table-level operations The following upstream operations put the affected table into an error state. Ingestion for that table stops, and you must drop and recreate the affected table in Materialize to resume: - Dropping a table (`DROP TABLE`), or removing it from the publication (`ALTER PUBLICATION ... DROP TABLE`). - Renaming a table or moving it to a different schema. - Setting a table's replica identity to anything other than `FULL` (`ALTER TABLE ... REPLICA IDENTITY`). - Truncating a table (`TRUNCATE`). To clear a table without putting it into an error state, use an unqualified `DELETE FROM t;` instead. ## Source failure states and recovery ### Operations that do not require re-creating the source Materialize tracks a [log sequence number (LSN)](https://www.postgresql.org/docs/current/wal-internals.html) as it consumes the upstream write-ahead log (WAL), and the source's replication slot retains the WAL that Materialize has not yet consumed. Because the slot outlives the connection, routine operational events do not lose data: after a transient interruption the source stalls, then resumes from its committed LSN and catches up automatically. **No action is required** for the following operations: - Restarting or patching PostgreSQL (including OS-level restarts). - Restarting Materialize. The source resumes from its committed LSN and does **not** re-snapshot already-ingested data. - Transient network interruptions between Materialize and PostgreSQL. These surface as [`connection closed`](/ingest-data/postgres/connection-closed/). - Resizing the cluster that hosts the source, or changing its replication factor. Briefly, the source may report [`replication slot ... is active`](/ingest-data/postgres/replication-slot-active/) while the upstream releases the slot from the previous connection. - The upstream database running out of disk space, once space is freed. > **Note:** Recovery after an interruption depends on the WAL that the replication slot is > holding still being available upstream. An interruption long enough for the slot > to be invalidated, or for the slot to be dropped, is not recoverable. See > [Replication slot invalidated](#replication-slot-invalidated) and [Replication > slot dropped or rewound](#replication-slot-dropped-or-rewound). > **Warning:** While a source is disconnected, the upstream WAL accumulates behind its > replication slot and cannot be reclaimed. A long outage, an undersized source > cluster, or a source cluster stuck in a restart loop can therefore consume > significant upstream disk. Monitor `restart_lsn` in > [`pg_replication_slots`](https://www.postgresql.org/docs/current/view-pg-replication-slots.html) > during planned maintenance. ### Operations that require re-creating the source A smaller set of events breaks LSN continuity or destroys the replication slot. When this happens, Materialize cannot guarantee a correct, gap-free view of your data. Most of these put the **entire source** into an error or permanently stalled state. One, [restoring from a volume or disk snapshot](#restoring-from-a-volume-or-disk-snapshot), cannot be detected at all, so the source keeps running on diverged data. Every event in this section requires **re-creating** the source. Upstream changes to an individual table's schema are handled separately, and do not error the entire source. In each case below, the remediation is to drop and re-create the source: ```mzsql DROP SOURCE mz_source CASCADE; CREATE SOURCE mz_source FROM POSTGRES CONNECTION pg_connection (PUBLICATION 'mz_source'); -- Re-create the tables you were ingesting. CREATE TABLE table_1 FROM SOURCE mz_source (REFERENCE public.table_1); ``` If you are using the legacy `CREATE SOURCE ... FOR TABLES` syntax, re-create the source with the same `FOR TABLES` list instead of adding tables separately. Because a re-created source snapshots from the current state of the upstream database, any changes it missed while it was in an error state are reflected in the snapshot rather than replayed as individual updates. > **Warning:** `CASCADE` drops every object that depends on the source, including its tables, > views, materialized views, indexes, and sinks. Capture their definitions before > you run it, and re-create them once the new source has finished snapshotting. #### Point-in-time restore Restoring the source database from a backup, including restoring to a different server for disaster recovery, increments the PostgreSQL timeline and is detected as a discontinuity. The source fails with an error of the form: ``` unsupported action: database restored from point-in-time backup. Expected timeline ID 8 but got 9 ``` The same error covers other events that change the timeline, such as a managed failover between replicas. To see the timeline a source is pinned to, query [`mz_internal.mz_postgres_sources`](/sql/system-catalog/mz_internal/#mz_postgres_sources): ```mzsql SELECT s.name, p.replication_slot, p.timeline_id FROM mz_internal.mz_postgres_sources p JOIN mz_catalog.mz_sources s ON s.id = p.id; ``` If your upstream fails over between replicas as part of routine maintenance, see [High-availability failovers](#high-availability-failovers). #### Restoring from a volume or disk snapshot Restoring the upstream data directory from a crash-consistent volume or disk snapshot rolls the database back, but preserves the timeline ID and the replication slot. Materialize cannot detect this kind of restore. The source keeps running without an error, but its contents diverge from upstream. This can surface later as incorrect results, or as negative-accumulation errors in queries such as `Non-positive multiplicity`. > **Warning:** After any restore of this kind, drop and re-create the source even if it reports > as `running`. Do not wait for the source to enter an error state, because it > will not. #### Promotion of a physical replica When a source reads from a physical standby (read replica) rather than the primary, promoting that standby to a primary fails the source with: ``` unsupported action: upstream physical replica status changed (e.g. a physical replica was promoted to a primary). Expected pg_is_in_recovery()=true but got false ``` Materialize detects the promotion while the replication stream is live, without waiting for a restart. Re-create the source against the promoted node. #### Replication slot invalidated PostgreSQL invalidates a replication slot once the WAL it holds exceeds [`max_slot_wal_keep_size`](https://www.postgresql.org/docs/current/runtime-config-replication.html#GUC-MAX-SLOT-WAL-KEEP-SIZE). This protects the upstream from running out of disk, at the cost of ending replication. The source fails with: ``` replication slot has been invalidated because it exceeded the maximum reserved size ``` To avoid this, size the source cluster so that it keeps up with the upstream write rate, and set `max_slot_wal_keep_size` high enough to cover your longest expected outage. Some hosted PostgreSQL services set this value for you and do not allow it to be raised. #### Replication slot dropped or rewound If the slot Materialize is using is dropped upstream, or the upstream is rebuilt from a base backup (which does not carry replication slots), a new slot starts at the current LSN, past the point the source needs to resume from. The source stalls with: ``` slot overcompacted. Requested LSN ... but only LSNs >= ... are available ``` For diagnosis steps, see [Slot overcompacted](/ingest-data/postgres/slot-overcompacted/). PostgreSQL refuses to drop a slot that is in use, so this generally happens only while the source is paused or disconnected. Not every rewind is caught this way. A rewind that leaves the slot able to serve the LSN the source asks for, such as [restoring from a volume or disk snapshot](#restoring-from-a-volume-or-disk-snapshot), raises no error at all. #### Dropping the publication Running `DROP PUBLICATION` upstream stalls the source, and all of its tables, with: ``` publication "mz_source" does not exist ``` Re-create the publication upstream, then re-create the source. #### Major version upgrades A PostgreSQL major version upgrade rewrites the on-disk format and does not preserve the replication slot, so there is no in-place recovery. To upgrade without a gap in your downstream views, run a second source against the upgraded instance in parallel and cut over once it has hydrated. See [Upgrade the major version of your PostgreSQL source](/ingest-data/postgres/major-version-upgrade/). ### High-availability failovers Some managed PostgreSQL services increment the timeline during routine high-availability operations, such as maintenance, a machine-tier change, or an automatic failover between replicas. Materialize cannot distinguish these from a genuine restore, so by default they fail the source with the [`Expected timeline ID`](#point-in-time-restore) error. On self-managed Materialize, where the upstream service guarantees that a failover is a contiguous fork of the WAL with no data loss, you can disable timeline validation with the [`pg_source_validate_timeline`](/sql/alter-system-set/) system parameter: ```mzsql ALTER SYSTEM SET pg_source_validate_timeline = false; ``` This parameter is not available on Materialize Cloud. There, a high-availability failover that changes the timeline requires re-creating the source. > **Warning:** Disabling this check is a trade-off. With it off, Materialize also does **not** > detect a genuine [point-in-time restore](#point-in-time-restore) or any other > discontinuous timeline change, and silently ingesting across one can corrupt the > contents of the source. Only disable it when your provider documents that its > failovers preserve WAL continuity for logical replication subscribers, and > re-create the source manually after any operation that does not. ## Examples ### Prerequisites To create a source from PostgreSQL 11+, you must first: - **Configure upstream PostgreSQL instance** - Set up logical replication. - Create a publication. - Create a replication user and password for Materialize to use to connect. - **Configure network security** - Ensure Materialize can connect to your PostgreSQL instance. - **Create a connection to PostgreSQL in Materialize** - The connection setup depends on the network security configuration. For details, see the [PostgreSQL integration guides](/ingest-data/postgres/#integration-guides). ### Create a source {#create-source-example} Once you have configured the upstream PostgreSQL, network security, and created the connection, you can create the source. In this example, the PostgreSQL publication is `mz_source` and the connection to PostgreSQL is `pg_connection`. ```mzsql /* This example assumes: - In the upstream PostgreSQL, you have defined: - replication user and password with the appropriate access. - a publication named `mz_source` for the `public.items` and `public.orders` tables. - In Materialize: - You have created a secret for the PostgreSQL password. - You have defined the connection to the upstream PostgreSQL. - You have used the connection to create a source. For example (substitute with your configuration): CREATE SECRET pgpass AS ''; -- substitute CREATE CONNECTION pg_connection TO POSTGRES ( HOST '', -- substitute DATABASE , -- substitute USER , -- substitute PASSWORD SECRET pgpass -- [, ] ); */ CREATE SOURCE pg_source FROM POSTGRES CONNECTION pg_connection ( PUBLICATION 'mz_source' ); ``` After a source is created, you can create tables from the source, referencing specific upstream table(s). Use a [DDL transaction block](/sql/begin/#ddl-only-transactions) to create multiple tables from the same source. ```mzsql BEGIN; CREATE TABLE items FROM SOURCE pg_source(REFERENCE public.items) ; CREATE TABLE orders FROM SOURCE pg_source(REFERENCE public.orders) ; COMMIT; ``` {{< note >}} - Although the example creates the tables with the same name as the upstream tables, the tables in Materialize can have different names. - You can create multiple tables that reference the same upstream table. {{< /note >}} For more information, see [`CREATE TABLE`](/sql/create-table/). ## Related pages - [`CREATE TABLE`](/sql/create-table/) - [`CREATE SECRET`](/sql/create-secret) - [`CREATE CONNECTION`](/sql/create-connection) - [PostgreSQL integration guides](/ingest-data/postgres/#integration-guides) [`enum`]: https://www.postgresql.org/docs/current/datatype-enum.html [`money`]: https://www.postgresql.org/docs/current/datatype-money.html