CREATE CHANGEFEED

On this page

The CREATE CHANGEFEED statement creates a new Enterprise changefeed, which targets an allowlist of tables called "watched rows". Every change to a watched row is emitted as a record in a configurable format (JSON or Avro) to a configurable sink (Kafka, Google Cloud Pub/Sub, a cloud storage sink, or a webhook sink). You can create, pause, resume, alter, or cancel an Enterprise changefeed.

We recommend reading the Changefeed Messages page for detail on understanding how changefeeds emit messages and Create and Configure Changefeeds for important usage considerations.

The examples on this page provide the foundational syntax of the CREATE CHANGEFEED statement. For examples on more specific use cases with changefeeds see the following pages:

• Change Data Capture Transformations
• Changefeeds on Tables with Column Families
• Export Data with Changefeeds

Required privileges

New in v22.2: You can grant a user the CHANGEFEED privilege to allow them to create changefeeds on a specific table:

GRANT CHANGEFEED ON TABLE example_table TO user;

This privilege model provides a more granular way to grant users the ability to create a changefeed on a table. A user granted the CHANGEFEED privilege can create changefeeds on the target table even if the user does not have the CONTROLCHANGEFEED role option or the SELECT privilege on the table.

Since you can grant the CHANGEFEED privilege to a user or role without them needing the SELECT privilege on a table, these users will be able to create changefeeds, but they will not be able to run a SELECT query on that data directly. However, these users could still read this data indirectly if they have read access to the sink, or create a "sinkless" changefeed that emits messages to the SQL session.

You can add CHANGEFEED to the user or role's default privileges with ALTER DEFAULT PRIVILEGES:

ALTER DEFAULT PRIVILEGES GRANT CHANGEFEED ON TABLES TO user;

Legacy privilege model

To create a changefeed, the user must be a member of the admin role or have the CREATECHANGEFEED parameter set.

Synopsis

Parameters

CDC transformation parameters

Change data capture transformations allow you to define the change data emitted to your sink when you create a changefeed. See the Change Data Capture Transformations page for detail on the functionality, syntax, and use cases for changefeeds created with transformations.

Sink URI

This section provides example URIs for each of the sinks that CockroachDB changefeeds support. For more comprehensive detail of using and configuring each sink, refer to the Changefeed Sinks page.

The sink URI follows the basic format of:

'{scheme}://{host}:{port}?{query_parameters}'

Kafka

Example of a Kafka sink URI:

'kafka://broker.address.com:9092?topic_prefix=bar_&tls_enabled=true&ca_cert=LS0tLS1CRUdJTiBDRVJUSUZ&sasl_enabled=true&sasl_user={sasl user}&sasl_password={url-encoded password}&sasl_mechanism=SCRAM-SHA-256'

Google Cloud Pub/Sub

Example of a Google Cloud Pub/Sub sink URI:

'gcpubsub://{project name}?region={region}&topic_name={topic name}&AUTH=specified&CREDENTIALS={base64-encoded key}'

Use Cloud Storage for Bulk Operations explains the requirements for the authentication parameter with specified or implicit. Refer to Changefeed Sinks for further consideration.

Cloud Storage

The following are example file URLs for each of the cloud storage schemes:

Use Cloud Storage explains the requirements for authentication and encryption for each supported cloud storage sink. Refer to Changefeed Sinks for considerations when using cloud storage.

Webhook

Example of a webhook URI:

'webhook-https://{your-webhook-endpoint}?insecure_tls_skip_verify=true'

Refer to Changefeed Sinks for specifics on webhook sink configuration.

Query parameters

Query parameters include:

Options

Files

The files emitted to a sink use the following naming conventions:

• General file format
• Resolved file format

General file format

/[date]/[timestamp]-[uniquer]-[topic]-[schema-id]

For example:

/2020-04-02/202004022058072107140000000000000-56087568dba1e6b8-1-72-00000000-test_table-1.ndjson

When emitting changefeed messages to a cloud storage sink, you can specify a partition format for your files using the partition_format query parameter. This will result in the following file path formats:

• daily: This is the default option and will follow the same pattern as the previous general file format.

• hourly: This will partition into an hourly directory as the changefeed emits messages, like the following:

  /2020-04-02/20/202004022058072107140000000000000-56087568dba1e6b8-1-72-00000000-test_table-1.ndjson
  

• flat: This will result in no file partitioning. The cloud storage path you specify when creating a changefeed will store all of the message files with no additional directories created.

Resolved file format

/[date]/[timestamp].RESOLVED

For example:

/2020-04-04/202004042351304139680000000000000.RESOLVED

Examples

Before running any of the examples in this section it is necessary to enable the kv.rangefeed.enabled cluster setting. If you are working on a CockroachDB Serverless cluster, this cluster setting is enabled by default.

The following examples show the syntax for managing changefeeds and starting changefeeds to specific sinks. The Options table on this page provides a list of all the available options. For information on sink-specific query parameters and configurations see the Changefeed Sinks page.

Create a changefeed connected to Kafka

> CREATE CHANGEFEED FOR TABLE name, name2, name3
  INTO 'kafka://host:port'
  WITH updated, resolved;

+--------------------+
|       job_id       |
+--------------------+
| 360645287206223873 |
+--------------------+
(1 row)

For step-by-step guidance on creating a changefeed connected to Kafka, see the Create a changefeed connected to Kafka example. The parameters table on the Changefeed Sinks page provides a list of all kafka-specific query parameters.

Create a changefeed connected to Kafka using Avro

> CREATE CHANGEFEED FOR TABLE name, name2, name3
  INTO 'kafka://host:port'
  WITH format = avro, confluent_schema_registry = <schema_registry_address>;

+--------------------+
|       job_id       |
+--------------------+
| 360645287206223873 |
+--------------------+
(1 row)

For more information on how to create a changefeed that emits an Avro record, see this step-by-step example. The parameters table on the Changefeed Sinks page provides a list of all kafka-specific query parameters.

Create a changefeed connected to a cloud storage sink

> CREATE CHANGEFEED FOR TABLE name, name2, name3
  INTO 'scheme://host?parameters'
  WITH updated, resolved;

+--------------------+
|       job_id       |
+--------------------+
| 360645287206223873 |
+--------------------+
(1 row)

For step-by-step guidance on creating a changefeed connected to a cloud storage sink, see the Changefeed Examples page. The parameters table on the Changefeed Sinks page provides a list of the available cloud storage parameters.

Create a changefeed with an S3 storage class

To associate the changefeed message files with a specific storage class in your Amazon S3 bucket, use the S3_STORAGE_CLASS parameter with the class. For example, the following S3 connection URI specifies the INTELLIGENT_TIERING storage class:

CREATE CHANGEFEED FOR TABLE name INTO 's3://{BUCKET NAME}?AWS_ACCESS_KEY_ID={KEY ID}&AWS_SECRET_ACCESS_KEY={SECRET ACCESS KEY}&S3_STORAGE_CLASS=INTELLIGENT_TIERING' WITH resolved;

Use the parameter to set one of these storage classes listed in Amazon's documentation. For more general usage information, see Amazon's Using Amazon S3 storage classes documentation.

Create a changefeed connected to a Google Cloud Pub/Sub

> CREATE CHANGEFEED FOR TABLE name, name2, name3
  INTO 'gcpubsub://project name?parameters'
  WITH resolved;

+--------------------+
|       job_id       |
+--------------------+
| 360645287206223873 |
+--------------------+
(1 row)

For step-by-step guidance on creating a changefeed connected to a Google Cloud Pub/Sub, see the Changefeed Examples page. The parameters table on the Changefeed Sinks page provides a list of the available Google Cloud Pub/Sub parameters.

Create a changefeed connected to a webhook sink

CREATE CHANGEFEED FOR TABLE name, name2, name3
  INTO 'webhook-https://{your-webhook-endpoint}?insecure_tls_skip_verify=true'
  WITH updated;

+---------------------+
|      job_id         |
----------------------+
| 687842491801632769  |
+---------------------+
(1 row)

For step-by-step guidance on creating a changefeed connected to a webhook sink, see the Changefeed Examples page. The parameters table on the Changefeed Sinks page provides a list of the available webhook parameters.

Manage a changefeed

For Enterprise changefeeds, use SHOW CHANGEFEED JOBS to check the status of your changefeed jobs:

> SHOW CHANGEFEED JOBS;

Use the following SQL statements to pause, resume, or cancel a changefeed.

Pause a changefeed

> PAUSE JOB job_id;

For more information, see PAUSE JOB.

Resume a paused changefeed

> RESUME JOB job_id;

For more information, see RESUME JOB.

Cancel a changefeed

> CANCEL JOB job_id;

For more information, see CANCEL JOB.

Modify a changefeed

To modify an Enterprise changefeed, pause the job and then use:

ALTER CHANGEFEED job_id {ADD table DROP table SET option UNSET option};

You can add new table targets, remove them, set new changefeed options, and unset them.

For more information, see ALTER CHANGEFEED.

Configuring all changefeeds

It is useful to be able to pause all running changefeeds during troubleshooting, testing, or when a decrease in CPU load is needed.

To pause all running changefeeds:

PAUSE JOBS (WITH x AS (SHOW CHANGEFEED JOBS) SELECT job_id FROM x WHERE status = ('running'));

This will change the status for each of the running changefeeds to paused, which can be verified with SHOW CHANGEFEED JOBS.

To resume all running changefeeds:

RESUME JOBS (WITH x AS (SHOW CHANGEFEED JOBS) SELECT job_id FROM x WHERE status = ('paused'));

This will resume the changefeeds and update the status for each of the changefeeds to running.

Start a new changefeed where another ended

In some situations, you may want to start a changefeed where a previously running changefeed ended. For example, a changefeed could encounter an error it cannot recover from, such as when a TRUNCATE is performed, and you need to restart the changefeed.

1. Use SHOW CHANGEFEED JOB to find the high-water timestamp for the ended changefeed:

   SHOW CHANGEFEED JOB {job_id};
   

           job_id       | ... |      high_water_timestamp      | ...
   +--------------------+ ... +--------------------------------+ ...
     383870400694353921 | ... | 1537279405671006870.0000000000 | ...
   (1 row)
   

   Note:

   If a changefeed has failed, you must restart the changefeed from a timestamp after the event that caused the failure.

2. Use the high_water_timestamp to start the new changefeed:

   CREATE CHANGEFEED FOR TABLE table_name, table_name2, table_name3
     INTO 'scheme//host:port'
     WITH cursor = '<high_water_timestamp>';
   

When you use the cursor option to start a changefeed, it will not perform an initial scan.

See also

• Change Data Capture Overview
• SQL Statements
• Changefeed Dashboard
• Monitor and Debug Changefeeds

Was this helpful?