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.

To get started with changefeeds, refer to the Create and Configure Changefeeds page for important usage considerations. For detail on how changefeeds emit messages, refer to the Changefeed Messages page.

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

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

Cockroach Labs recommends monitoring your changefeeds to track retryable errors and protected timestamp usage. Refer to the Monitor and Debug Changefeeds page for more information.

Required privileges

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;

When you grant a user the CHANGEFEED privilege on a set of tables, they can:

• Create changefeeds on the target tables even if the user does not have the CONTROLCHANGEFEED role option or the SELECT privilege on the tables.
• Manage the changefeed jobs running on the tables using the SHOW CHANGEFEED JOB, PAUSE JOB, RESUME JOB, and CANCEL JOB commands.

These users will be able to create changefeeds, but they will not be able to run a SELECT query on that data directly. However, they could still read this data indirectly if they have read access to the sink.

To restrict a user's access to changefeed data and sink credentials, enable the changefeed.permissions.require_external_connection_sink.enabled cluster setting. When you enable this setting, users with the CHANGEFEED privilege on a set of tables can only create changefeeds into external connections.

Privilege model

The following summarizes the operations users can run when they have changefeed privileges on a table:

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 query parameters

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

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.

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

You can create an external connection to represent a changefeed sink URI. This allows you to specify the external connection's name in statements rather than the provider-specific URI. For detail on using external connections, see the CREATE EXTERNAL CONNECTION page.

Azure Event Hubs

Example for an Azure Event Hubs URI:

'azure-event-hub://{event-hubs-namespace}.servicebus.windows.net:9093?shared_access_key_name={policy-name}&shared_access_key={url-encoded key}'

You can also use a kafka:// scheme in the URI:

'kafka://{event-hubs-namespace}.servicebus.windows.net:9093?shared_access_key_name={policy-name}&shared_access_key={url-encoded key}'

Cloud Storage

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

For detail on authentication to cloud storage, refer to the Cloud Storage Authentication page. Refer to Changefeed Sinks for considerations when using cloud storage.

Confluent Cloud

Example of a Confluent Cloud sink URI:

'confluent-cloud://pkc-lzvrd.us-west4.gcp.confluent.cloud:9092?api_key={API key}&api_secret={url-encoded API secret}'

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}'

In CockroachDB v23.2 and later, the changefeed.new_pubsub_sink_enabled cluster setting is enabled by default, which provides improved throughput. For details on the changes to the message format, refer to Pub/Sub sink messages.

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

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'

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.

Apache Pulsar

Example for an Apache Pulsar sink URI:

pulsar://{host IP}:6650

By default, Apache Pulsar listens for client connections on port :6650. For more detail on configuration, refer to the Apache Pulsar documentation.

Changefeeds emitting to a Pulsar sink do not support external connections or a number of changefeed options. For a full list, refer to the Changefeed Sinks page.

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 with different use cases and features. The Options table on this page provides a list of all the available options. For information on sink-specific query parameters and configurations, refer to the Changefeed Sinks page.

You can create an external connection to represent a changefeed sink URI. This allows you to specify the external connection's name in statements rather than the provider-specific URI. For detail on using external connections, see the CREATE EXTERNAL CONNECTION page.

Create a changefeed connected to a sink

You can connect a changefeed to the following sinks:

• Kafka
• Cloud storage / HTTP
• Google Cloud Pub/Sub
• Webhook

CREATE CHANGEFEED FOR TABLE table_name, table_name2, table_name3
  INTO 'scheme://host:port'
  WITH updated, resolved;

For guidance on the sink URI, refer to:

• The Changefeed Sinks page for general detail on query parameters and sink configuration.
• The Cloud Storage Authentication page for instructions on setting up each supported cloud storage authentication.

Create a changefeed that filters and transforms change data

CDC queries can filter and transform change data before emitting it to a sink or a SQL client.

You can adapt a changefeed with CDC queries by including SELECT and WHERE clauses in your CREATE statement:

CREATE CHANGEFEED INTO 'scheme://host:port'
  WITH updated, resolved
  AS SELECT owner_id, status
  FROM vehicles
  WHERE status = 'lost';

CDC queries can only run on a single table per changefeed and require an Enterprise license.

Create a sinkless changefeed

You can create a changefeed that will send messages to the SQL client rather than a sink:

CREATE CHANGEFEED FOR TABLE table_name, table_name2, table_name3
  WITH updated, resolved;

Sinkless changefeeds do not require an Enterprise license; however, a sinkless changefeed with CDC queries does require an Enterprise license.

To create a sinkless changefeed using CDC queries:

CREATE CHANGEFEED WITH updated, resolved
  AS SELECT owner_id, status
  FROM vehicles
  WHERE status = 'lost';

Use an external connection to specify a changefeed sink

External connections provide a way to define a name for a sink, which you can use instead of the provider-specific URI.

To restrict a user's access to changefeed data and sink credentials, enable the changefeed.permissions.require_external_connection_sink.enabled cluster setting. When you enable this setting, users with the CHANGEFEED privilege on a set of tables can only create changefeeds into external connections.

External connections support all changefeed sinks.

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

In the changefeed statement, you specify the external connection name:

CREATE CHANGEFEED FOR TABLE table_name INTO 'external://kafka_sink'
  WITH resolved;

Filter changefeeds for tables using row-level TTL

Use the ttl_disable_changefeed_replication table storage parameter to prevent changefeeds from sending DELETE messages issued by row-level TTL jobs for a table. Include the storage parameter when you create or alter the table. For example:

CREATE TABLE tbl (
  id UUID PRIMARY KEY default gen_random_uuid(),
  value TEXT
) WITH (ttl_expire_after = '3 weeks', ttl_job_cron = '@daily', ttl_disable_changefeed_replication = 'true');

ALTER TABLE events SET (ttl_expire_after = '1 year', ttl_disable_changefeed_replication = 'true');

You can also widen the scope to the cluster by setting the sql.ttl.changefeed_replication.disabled cluster setting to true. This will prevent changefeeds from emitting deletes issued by all TTL jobs on a cluster.

If you want to have a changefeed ignore the storage parameter or cluster setting that disables changefeed replication, you can set the changefeed option ignore_disable_changefeed_replication to true:

CREATE CHANGEFEED FOR TABLE table_name INTO 'external://changefeed-sink'
  WITH resolved, ignore_disable_changefeed_replication = true;

This is useful when you have multiple use cases for different changefeeds on the same table. For example, you have a table with a changefeed streaming changes to another database for analytics workflows in which you do not want to reflect row-level TTL deletes. Secondly, you have a changefeed on the same table for audit-logging purposes for which you need to persist every change through the changefeed.

For guidance on how to filter changefeed messages to emit row-level TTL deletes only, refer to Change Data Capture Queries.

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.

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

Define a key to determine the changefeed sink partition

With the key_column option, you can define the key used in message metadata that determines the partition for the changefeed message at your downstream sink. This option overrides the default primary key:

CREATE CHANGEFEED FOR TABLE table_name
  INTO 'external://kafka-sink'
  WITH key_column='partition_column', unordered;

key_column does not preserve ordering of messages from CockroachDB to the downstream sink, therefore you must include the unordered option. It does not affect per-key ordering guarantees or the output of key_in_value.

See also

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

Was this helpful?