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 or a cloud storage sink). You can create, pause, resume, or cancel an enterprise changefeed.

For more information, see Change Data Capture.

Required privileges

Changefeeds can only be created by superusers, i.e., members of the admin role. The admin role exists by default with root as the member.

Considerations

• In most cases, each version of a row will be emitted once. However, some infrequent conditions (e.g., node failures, network partitions) will cause them to be repeated. This gives our changefeeds an at-least-once delivery guarantee. For more information, see Change Data Capture - Ordering Guarantees.

Synopsis

Parameters

Sink URI

The sink URI follows the basic format of:

'[scheme]://[host]:[port]?[query_parameters]'

The scheme can be kafka or any cloud storage sink.

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=petee&sasl_password=bones'

Query parameters include:

Cloud storage sink

Use a cloud storage sink to deliver changefeed data to OLAP or big data systems without requiring transport via Kafka.

Example of a cloud storage sink (i.e., AWS S3) URI:

'experimental-s3://test-s3encryption/test?AWS_ACCESS_KEY_ID=ABCDEFGHIJKLMNOPQ&AWS_SECRET_ACCESS_KEY=LS0tLS1CRUdJTiBDRVJUSUZ'

Any of the cloud storages below can be used as a sink:

[scheme]://[host]/[path]?[parameters]

• ¹ If the AUTH parameter is not provided, AWS connections default to specified and the access keys must be provided in the URI parameters. If the AUTH parameter is implicit, the access keys can be omitted and the credentials will be loaded from the environment.

• ² If the AUTH parameter is not specified, the cloudstorage.gs.default.key cluster setting will be used if it is non-empty, otherwise the implicit behavior is used. If the AUTH parameter is implicit, all GCS connections use Google's default authentication strategy. If the AUTH parameter is default, the cloudstorage.gs.default.key cluster setting must be set to the contents of a service account file which will be used during authentication. If the AUTH parameter is specified, GCS connections are authenticated on a per-statement basis, which allows the JSON key object to be sent in the CREDENTIALS parameter. The JSON key object should be base64-encoded (using the standard encoding in RFC 4648).

• ³ You can create your own HTTP server with Caddy or nginx. A custom root CA can be appended to the system's default CAs by setting the cloudstorage.http.custom_ca cluster setting, which will be used when verifying certificates from HTTPS URLs.

• ⁴ The file system backup location on the NFS drive is relative to the path specified by the --external-io-dir flag set while starting the node. If the flag is set to disabled, then imports from local directories and NFS drives are disabled.

• ⁵ The host component of NFS/Local can either be empty or the nodeID. If the nodeID is specified, it is currently ignored (i.e., any node can be sent work and it will look in its local input/output directory); however, the nodeID will likely be required in the future.

• ⁶ A custom root CA can be appended to the system's default CAs by setting the cloudstorage.http.custom_ca cluster setting, which will be used when verifying certificates from an S3-compatible service.

• ⁷ The AWS_REGION parameter is optional since it is not a required parameter for most S3-compatible services. Specify the parameter only if your S3-compatible service requires it.

Example file URLs

Options

Avro limitations

Currently, support for Avro is limited and experimental. Below is a list of unsupported SQL types and values for Avro changefeeds:

• Decimals must have precision specified.

• Decimals with NaN or infinite values cannot be written in Avro.

  Note:

  To avoid NaN or infinite values, add a CHECK constraint to prevent these values from being inserted into decimal columns.

• TIME, DATE, INTERVAL, UUID, INET, ARRAY, JSONB, BIT, and collated STRING are not supported in Avro yet.

Avro types

Below is a mapping of CockroachDB types to Avro types:

Responses

The messages (i.e., keys and values) emitted to a Kafka topic are specific to the envelope. The default format is wrapped, and the output messages are composed of the following:

• Key: An array always composed of the row's PRIMARY KEY field(s) (e.g., [1] for JSON or {"id":{"long":1}} for Avro).
• Value:
  • One of three possible top-level fields:
    • after, which contains the state of the row after the update (or null' for DELETEs).
    • updated, which contains the updated timestamp.
    • resolved, which is emitted for records representing resolved timestamps. These records do not include an "after" value since they only function as checkpoints.
  • For INSERT and UPDATE, the current state of the row inserted or updated.
  • For DELETE, null.

For example:

Examples

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 more information on how to create a changefeed connected to Kafka, see Change Data Capture.

Create a changefeed connected to Kafka using Avro

> CREATE CHANGEFEED FOR TABLE name, name2, name3
  INTO 'kafka://host:port'
  WITH format = experimental_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 Change Data Capture.

Create a changefeed connected to a cloud storage sink

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

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

For more information on how to create a changefeed connected to a cloud storage sink, see Change Data Capture.

Manage a changefeed

Use the following SQL statements to pause, resume, and 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.

Start a new changefeed where another ended

Find the high-water timestamp for the ended changefeed:

> SELECT * FROM crdb_internal.jobs WHERE job_id = <job_id>;

        job_id       |  job_type  | ... |      high_water_timestamp      | error | coordinator_id
+--------------------+------------+ ... +--------------------------------+-------+----------------+
  383870400694353921 | CHANGEFEED | ... | 1537279405671006870.0000000000 |       |              1
(1 row)

Use the high_water_timestamp to start the new changefeed:

> CREATE CHANGEFEED FOR TABLE name, name2, name3
  INTO 'kafka//host:port'
  WITH cursor = '<high_water_timestamp>';

Note that because the cursor is provided, the initial scan is not performed.

See also

• Change Data Capture
• Other SQL Statements
• Changefeed Dashboard

Was this helpful?