# Ingest data from Neon How to stream data from Neon to Materialize > **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). [Neon](https://neon.tech) is a fully managed serverless PostgreSQL provider. It separates compute and storage to offer features like **autoscaling**, **branching** and **bottomless storage**. This page shows you how to stream data from a Neon database to Materialize using the [PostgreSQL source](/sql/create-source/postgres/). ## Before you begin - Make sure you have [a Neon account](https://neon.tech). - Make sure you have access to your Neon instance via [`psql`](https://www.postgresql.org/docs/current/app-psql.html) or the SQL editor in the Neon Console. ## A. Configure Neon The steps in this section are specific to Neon. You can run them by connecting to your Neon database using a `psql` client or the SQL editor in the Neon Console. ### 1. Enable logical replication > **Warning:** Enabling logical replication applies **globally** to all databases in your Neon > project, and **cannot be reverted**. It also **restarts all computes**, which > means that any active connections are dropped and have to reconnect. Materialize uses PostgreSQL's [logical replication](https://www.postgresql.org/docs/current/logical-replication.html) protocol to track changes in your database and propagate them to Materialize. As a first step, you need to make sure logical replication is enabled in Neon. 1. Select your project in the Neon Console. 2. On the Neon **Dashboard**, select **Settings**. 3. Select **Logical Replication**. 4. Click **Enable** to enable logical replication. You can verify that logical replication is enabled by running: ```sql SHOW wal_level; ``` The result should be: ``` wal_level ----------- logical ``` ### 2. Create a publication and a replication user 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. 1. For each table that you want to replicate to Materialize, set the [replica identity](https://www.postgresql.org/docs/current/sql-altertable.html#SQL-ALTERTABLE-REPLICA-IDENTITY) to `FULL`: ```postgres ALTER TABLE REPLICA IDENTITY FULL; ``` ```postgres ALTER TABLE 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 Neon data with minimal in-memory state. However, you should expect increased disk usage in your Neon database. 2. Create a [publication](https://www.postgresql.org/docs/current/logical-replication-publication.html) with the tables you want to replicate: _For specific tables:_ ```postgres CREATE PUBLICATION mz_source FOR TABLE , ; ``` _For all tables in the database:_ ```postgres 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. 3. Create a dedicated user for Materialize, if you don't already have one. The default user created with your Neon project and users created using the Neon CLI, Console or API are granted membership in the [`neon_superuser`](https://neon.tech/docs/manage/roles#the-neonsuperuser-role) role, which has the required `REPLICATION` privilege. While you can use the default user for replication, we recommend creating a dedicated user for security reasons. **Neon CLI:** Use the [`roles create` CLI command](https://neon.tech/docs/reference/cli-roles) to create a new role. ```bash neon roles create --name materialize ``` **Neon Console:** 1. Navigate to the [Neon Console](https://console.neon.tech). 2. Select a project. 3. Select **Branches**. 4. Select the branch where you want to create the role. 5. Select the **Roles & Databases** tab. 6. Click **Add Role**. 7. In the role creation dialog, specify the role name as "materialize". 8. Click **Create**. The role is created, and you are provided with the password for the role. **API:** Use the [`roles` endpoint](https://api-docs.neon.tech/reference/createprojectbranchrole) to create a new role. ```bash curl 'https://console.neon.tech/api/v2/projects//branches//roles' \ -H 'Accept: application/json' \ -H "Authorization: Bearer $NEON_API_KEY" \ -H 'Content-Type: application/json' \ -d '{ "role": { "name": "materialize" } }' | jq ``` 4. Grant the user the required permissions on the schema(s) you want to replicate: ```postgres GRANT USAGE ON SCHEMA public TO materialize; GRANT SELECT ON ALL TABLES IN SCHEMA public TO materialize; ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO materialize; ``` Granting `SELECT ON ALL TABLES IN SCHEMA` instead of on specific tables avoids having to add privileges later if you add tables to your publication. ## B. (Optional) Configure network security > **Note:** If you are prototyping and your Neon instance is publicly accessible, **you can > skip this step**. For production scenarios, we recommend using [**IP Allow**](https://neon.tech/docs/introduction/ip-allow) > to limit the IP addresses that can connect to your Neon instance. **Cloud:** If you use Neon's [**IP Allow**](https://neon.tech/docs/introduction/ip-allow) feature to limit the IP addresses that can connect to your Neon instance, you will need to allow inbound traffic from Materialize IP addresses. 1. In the [Materialize console's SQL Shell](/console/), or your preferred SQL client connected to Materialize, run the following query to find the static egress IP addresses, for the Materialize region you are running in: ```mzsql SELECT * FROM mz_egress_ips; ``` 2. In your Neon project, add the IPs to your **IP Allow** list: 1. Select your project in the Neon Console. 2. On the Neon **Dashboard**, select **Settings**. 3. Select **IP Allow**. 4. Add each Materialize IP address to the list. **Self-Managed:** > **Note:** If you are prototyping and your Neon instance is publicly accessible, **you can > skip this step**. For production scenarios, we recommend using [**IP Allow**](https://neon.tech/docs/introduction/ip-allow) > to limit the IP addresses that can connect to your Neon instance. If you use Neon's [**IP Allow**](https://neon.tech/docs/introduction/ip-allow) feature to limit the IP addresses that can connect to your Neon instance, you will need to allow inbound traffic from Materialize IP addresses. 2. In your Neon project, add the IPs to your **IP Allow** list: 1. Select your project in the Neon Console. 2. On the Neon **Dashboard**, select **Settings**. 3. Select **IP Allow**. 4. Add Materialize IP addresses to the list. ## C. Ingest data in Materialize The steps in this section are specific to Materialize. You can run them in the [Materialize console's SQL Shell](/console/) or your preferred SQL client connected to Materialize. ### 1. (Optional) Create a cluster > **Note:** If you are prototyping and already have a cluster to host your PostgreSQL > source (e.g. `quickstart`), **you can skip this step**. For production > scenarios, we recommend separating your workloads into multiple clusters for > [resource isolation](/sql/create-cluster/#resource-isolation).

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.

  1. 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> );
### 2. Create a connection Once you have configured your network, create a connection in Materialize per your networking configuration. 1. Run the [`CREATE SECRET`](/sql/create-secret/) command to securely store the password for the `materialize` PostgreSQL user you created [earlier](#2-create-a-publication-and-a-replication-user): ```mzsql CREATE SECRET pgpass AS ''; ``` You can access the password for your Neon user from the **Connection Details** widget on the Neon **Dashboard**. 2. Use the [`CREATE CONNECTION`](/sql/create-connection/) command to create a connection object with access and authentication details for Materialize to use: ```mzsql CREATE CONNECTION pg_connection TO POSTGRES ( HOST '', PORT 5432, USER '', PASSWORD SECRET pgpass, SSL MODE 'require', DATABASE '' ); ``` You can find the connection details for your replication user in the **Connection Details** widget on the Neon **Dashboard**. A Neon connection string looks like this: ```bash postgresql://materialize:AbC123dEf@ep-cool-darkness-123456.us-east-2.aws.neon.tech/dbname?sslmode=require ``` - Replace `` with your Neon hostname (e.g., `ep-cool-darkness-123456.us-east-2.aws.neon.tech`). - Replace `` with the dedicated replication user (e.g., `materialize`). - Replace `` with the name of the database containing the tables you want to replicate to Materialize (e.g., `dbname`). ### 3. Start ingesting data {{< tip >}} When snapshotting, Materialize uses PostgreSQL statistics to estimate the amount of data and number of rows to read. Before creating the source in Materialize, check that the PostgreSQL statistics are up to date by running PostgreSQL `ANALYZE`. See [Snapshotting considerations](#snapshotting) for more information. {{< /tip >}} {{< tabs >}} {{< tab "Legacy Syntax" >}} #### Legacy syntax {{% include-example file="examples/ingest_data/postgres/create_source_cloud" example="create-source-legacy" %}} {{% include-example file="examples/ingest_data/postgres/create_source_cloud" example="schema-changes" %}} {{< /tab >}} {{< tab "New Syntax" >}} #### New syntax {{% include-example file="examples/ingest_data/postgres/create_source_cloud" example="create-source" %}} {{% include-example file="examples/ingest_data/postgres/create_source_cloud" example="schema-changes" %}} {{< /tab >}} {{< /tabs >}} ### 4. Monitor the ingestion status

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.

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

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

### 5. Right-size the cluster

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.

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

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

  3. Going forward, you can verify that your new cluster size is sufficient as follows:

    1. 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;

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

## D. Explore your data

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.

## Considerations

Schema changes

Materialize supports schema changes in the upstream database as follows:

Compatible schema changes (Legacy syntax)

Note: This section refer to the legacy CREATE SOURCE ... FOR ... that creates subsources as part of the CREATE SOURCE operation. To be able to handle the upstream column additions and drops, see CREATE SOURCE (New Syntax) and CREATE 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.

Incompatible schema changes

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.

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

Supported types

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

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.

Truncation

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;

Inherited tables

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.

Replication slots

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.

Modifying an existing source

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

Snapshotting

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.