> ## Documentation Index > Fetch the complete documentation index at: https://docs.cockroachlabs.com/llms.txt > Use this file to discover all available pages before exploring further. # CREATE CHANGEFEED The `CREATE CHANGEFEED` [statement](/docs/stable/sql-statements.md) 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](https://kafka.apache.org/), [Google Cloud Pub/Sub](https://cloud.google.com/pubsub), a [cloud storage sink](/docs/stable/changefeed-sinks.md#cloud-storage-sink), or a [webhook sink](/docs/stable/changefeed-sinks.md#webhook-sink)). You can [create](#examples), [pause](#pause-a-changefeed), [resume](#resume-a-paused-changefeed), [alter](/docs/stable/alter-changefeed.md), or [cancel](#cancel-a-changefeed) an Enterprise changefeed. To get started with changefeeds, refer to the [Create and Configure Changefeeds](/docs/stable/create-and-configure-changefeeds.md) page for important usage considerations. For detail on how changefeeds emit messages, refer to the [Changefeed Messages](/docs/stable/changefeed-messages.md#ordering-and-delivery-guarantees) page. The [examples](#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](/docs/stable/cdc-queries.md#filter-columns) * [Changefeeds on Tables with Column Families](/docs/stable/changefeeds-on-tables-with-column-families.md) * [Export Data with Changefeeds](/docs/stable/export-data-with-changefeeds.md) Cockroach Labs recommends monitoring your changefeeds to track [retryable errors](/docs/stable/monitor-and-debug-changefeeds.md#changefeed-retry-errors) and [protected timestamp](/docs/stable/architecture/storage-layer.md#protected-timestamps) usage. Refer to the [Monitor and Debug Changefeeds](/docs/stable/monitor-and-debug-changefeeds.md#changefeed-retry-errors) page for more information. ## Required privileges You can [grant](/docs/stable/grant.md#grant-privileges-on-specific-tables-in-a-database) a user the `CHANGEFEED` privilege to allow them to create changefeeds on a specific table: ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} 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](/docs/stable/alter-role#role-options) or the `SELECT` privilege on the tables. * Manage the changefeed jobs running on the tables using the [`SHOW CHANGEFEED JOB`](/docs/stable/show-jobs#show-changefeed-jobs), [`PAUSE JOB`](/docs/stable/pause-job), [`RESUME JOB`](/docs/stable/resume-job), and [`CANCEL JOB`](/docs/stable/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](/docs/stable/changefeed-sinks.md). 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](/docs/stable/create-changefeed#required-privileges) on a set of tables can only create changefeeds into [external connections](/docs/stable/create-external-connection.md). You can add `CHANGEFEED` to the user or role's [default privileges](/docs/stable/security-reference/authorization.md#default-privileges) with [`ALTER DEFAULT PRIVILEGES`](/docs/stable/alter-default-privileges#grant-default-privileges-to-a-specific-role): ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} ALTER DEFAULT PRIVILEGES GRANT CHANGEFEED ON TABLES TO user; ``` ### Privilege model > [!TIP] > > For fine-grained access control, we recommend using the system-level privileges `CHANGEFEED` and `CONTROLJOB` / `VIEWJOB`. The following summarizes the operations users can run when they have changefeed privileges on a table: | Granted privileges | Usage | | - | - | | `CHANGEFEED` | Create changefeeds on tables. View and manage changefeed jobs on tables. | | `CHANGEFEED` + [`USAGE`](/docs/stable/create-external-connection#required-privileges) on external connection | Create changefeeds on tables to an external connection URI. Manage changefeed jobs on tables. **Note:** If you need to manage access to changefeed sink URIs, set the `changefeed.permissions.require_external_connection_sink.enabled=true` cluster setting. This will mean that users with these privileges can **only** create changefeeds on external connections. | | `SELECT` | Create a sinkless changefeed that emits messages to a SQL client. | | **Deprecated** `CONTROLCHANGEFEED` role option + `SELECT` | Create changefeeds on tables. Users with the `CONTROLCHANGEFEED` role option must have `SELECT` on each table, even if they are also granted the `CHANGEFEED` privilege. The `CONTROLCHANGEFEED` role option will be removed in a future release. We recommend using the [system-level privileges](/docs/stable/security-reference/authorization.md#supported-privileges) `CHANGEFEED` and `CONTROLJOB`/ `VIEWJOB` for fine-grained access control. | | `admin` | Create, view, and manage changefeed jobs. | ## Synopsis ![create\_changefeed syntax diagram](https://mintcdn.com/cockroachlabs/O8LdfYQ-dz7oGPVh/images/sql-diagrams/v24.2/create_changefeed.svg?fit=max\&auto=format\&n=O8LdfYQ-dz7oGPVh\&q=85\&s=b82608efec16a4c2fd2d7ced9762962c) ## Parameters | Parameter | Description | | - | - | | `table_name` | The name of the table (or tables in a comma separated list) to create a changefeed for. **Note:** Before creating a changefeed, consider the number of changefeeds versus the number of tables to include in a single changefeed. Each scenario can have an impact on total memory usage or changefeed performance. Refer to [Create and Configure Changefeeds](/docs/stable/create-and-configure-changefeeds.md) for more detail. | | `sink` | The location of the configurable sink. The scheme of the URI indicates the type. For more information, refer to [Sink URI](#sink-uri). **Note:** If you create a changefeed without a sink, your changefeed will run like a [basic changefeed](/docs/stable/changefeed-for.md) sending messages to the SQL client. For more detail, refer to the [Create and Configure Changefeeds](/docs/stable/create-and-configure-changefeeds.md#create) page. | | `option` / `value` | For a list of available options and their values, refer to [Options](#options). | ### 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](/docs/stable/cdc-queries.md#filter-columns) page for detail on the functionality, syntax, and use cases for changefeeds created with queries. | Parameter | Description | | - | - | | `sink` | The location of the configurable sink. The scheme of the URI indicates the type. For more information, see [Sink URI](#sink-uri). | | `option` / `value` | For a list of available options and their values, see [Options](#options). | | `target_list` | The columns to emit data from. | | `changefeed_target_expr` | The target table for the changefeed. | | `opt_where_clause` | An optional `WHERE` clause to apply filters to the table. | ### Sink URI To form the URI for each sink: ``` '{scheme}://{host}:{port}?{query_parameters}' ``` | URI Component | Description | | - | - | | `scheme` | The type of sink, e.g., [`kafka`](/docs/stable/changefeed-sinks#kafka), [`gcpubsub`](/docs/stable/changefeed-sinks). | | `host` | The sink's hostname or IP address. | | `port` | The sink's port. | | `query_parameters` | The sink's [query parameters](#query-parameters). | For more comprehensive detail of using and configuring each sink, refer to: * [Amazon MSK](#amazon-msk) [Amazon MSK](/docs/stable/changefeed-sinks.md#amazon-msk) * [Apache Pulsar](#apache-pulsar) (in Preview) [Apache Pulsar](/docs/stable/changefeed-sinks.md#apache-pulsar) (in Preview) * [Azure Event Hubs](#azure-event-hubs) [Azure Event Hubs](/docs/stable/changefeed-sinks.md#azure-event-hubs) * [Cloud Storage](#cloud-storage-sink) / HTTP [Cloud Storage](/docs/stable/changefeed-sinks.md#cloud-storage-sink) / HTTP * [Confluent Cloud](#confluent-cloud) [Confluent Cloud](/docs/stable/changefeed-sinks.md#confluent-cloud) * [Google Cloud Pub/Sub](#google-cloud-pub-sub) [Google Cloud Pub/Sub](/docs/stable/changefeed-sinks.md) * [Kafka](#kafka) [Kafka](/docs/stable/changefeed-sinks.md#kafka) * [Webhook](#webhook-sink) [Webhook](/docs/stable/changefeed-sinks.md#webhook-sink) 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`](/docs/stable/create-external-connection) page. ### Query parameters > [!NOTE] > > Parameters should always be URI-encoded before they are included the changefeed's URI, as they often contain special characters. Use Javascript's [encodeURIComponent](https://developer.mozilla.org/docs/Web/JavaScript/Reference/Global_Objects/encodeURIComponent) function or Go language's [url.QueryEscape](https://golang.org/pkg/net/url/#QueryEscape) function to URI-encode the parameters. Other languages provide similar functions to URI-encode special characters. Query parameters include: | Parameter | Sink Type | Type | Description | | - | - | - | - | | `ASSUME_ROLE` | [Amazon S3](/docs/stable/changefeed-sinks.md), [Google Cloud Storage](/docs/stable/changefeed-sinks.md#cloud-storage-sink), [Google Cloud Pub/Sub](/docs/stable/changefeed-sinks.md) | [`STRING`](/docs/stable/string) | Pass the [ARN](https://docs.aws.amazon.com/general/latest/gr/aws-arns-and-namespaces.html) of the role to assume. Use in combination with `AUTH=implicit` or `specified`. `external_id`: Use as a value to `ASSUME_ROLE` to specify the [external ID](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-user_externalid.html) for third-party access to your S3 bucket. | | `AUTH` | [Amazon S3](/docs/stable/changefeed-sinks.md), [Google Cloud Storage](/docs/stable/changefeed-sinks.md#cloud-storage-sink), [Google Cloud Pub/Sub](/docs/stable/changefeed-sinks.md), [Azure Blob Storage](/docs/stable/changefeed-sinks.md#cloud-storage-sink) | [`STRING`](/docs/stable/string) | The authentication parameter can define either `specified` (default) or `implicit` authentication. To use `specified` authentication, pass your [Service Account](https://cloud.google.com/iam/docs/understanding-service-accounts) credentials with the URI. To use `implicit` authentication, configure these credentials via an environment variable. Refer to the [Cloud Storage Authentication page](/docs/stable/cloud-storage-authentication.md) page for examples of each of these. | | `api_key` | [Confluent Cloud](/docs/stable/changefeed-sinks.md#confluent-cloud) | [`STRING`](/docs/stable/string) | The API key created for the cluster in Confluent Cloud. | | `api_secret` | [Confluent Cloud](/docs/stable/changefeed-sinks.md#confluent-cloud) | [`STRING`](/docs/stable/string) | The API key's secret generated in Confluent Cloud. **Note:** This must be [URL-encoded](https://www.urlencoder.org/) before passing into the connection string. | | `ca_cert` | [Kafka](/docs/stable/changefeed-sinks.md#kafka), [webhook](/docs/stable/changefeed-sinks.md#webhook-sink), [Confluent schema registry](https://docs.confluent.io/platform/current/schema-registry/index.html) | [`STRING`](/docs/stable/string) | The base64-encoded `ca_cert` file. Specify `ca_cert` for a Kafka sink, webhook sink, and/or a Confluent schema registry. For usage with a Kafka sink, see [Kafka Sink URI](/docs/stable/changefeed-sinks.md#kafka). It's necessary to state `https` in the schema registry's address when passing `ca_cert`: `confluent_schema_registry='https://schema_registry:8081?ca_cert=LS0tLS1CRUdJTiBDRVJUSUZ'` See [`confluent_schema_registry`](#confluent-schema-registry) for more detail on using this option. Note: To encode your `ca.cert`, run `base64 -w 0 ca.cert`. | | `client_cert` | [Kafka](/docs/stable/changefeed-sinks.md#kafka), [webhook](/docs/stable/changefeed-sinks.md#webhook-sink), [Confluent schema registry](https://docs.confluent.io/platform/current/schema-registry/index.html) | [`STRING`](/docs/stable/string) | The base64-encoded Privacy Enhanced Mail (PEM) certificate. This is used with `client_key`. | | `client_key` | [Kafka](/docs/stable/changefeed-sinks.md#kafka), [webhook](/docs/stable/changefeed-sinks.md#webhook-sink), [Confluent schema registry](https://docs.confluent.io/platform/current/schema-registry/index.html) | [`STRING`](/docs/stable/string) | The base64-encoded private key for the PEM certificate. This is used with `client_cert`. **Note:** Client keys are often encrypted. You will receive an error if you pass an encrypted client key in your changefeed statement. To decrypt the client key, run: `openssl rsa -in key.pem -out key.decrypt.pem -passin pass:{PASSWORD}`. Once decrypted, be sure to update your changefeed statement to use the new `key.decrypt.pem` file instead. | | `file_size` | [cloud](/docs/stable/changefeed-sinks.md#cloud-storage-sink) | [`STRING`](/docs/stable/string) | The file will be flushed (i.e., written to the sink) when it exceeds the specified file size. This can be used with the [`WITH resolved` option](#options), which flushes on a specified cadence. When you change or increase `file_size`, ensure that you adjust the `changefeed.memory.per_changefeed_limit` [cluster setting](/docs/stable/cluster-settings.md), which has a default of `512MiB`. Buffering messages can quickly reach this limit if you have increased the file size. Refer to [Advanced Changefeed Configuration](/docs/stable/advanced-changefeed-configuration.md) for more detail. **Default:** `16MB` | | `insecure_tls_skip_verify` | [Kafka](/docs/stable/changefeed-sinks.md#kafka), [webhook](/docs/stable/changefeed-sinks.md#webhook-sink) | [`BOOL`](/docs/stable/bool) | If `true`, disable client-side validation of responses. Note that a CA certificate is still required; this parameter means that the client will not verify the certificate. **Warning:** Use this query parameter with caution, as it creates [MITM](https://wikipedia.org/wiki/Man-in-the-middle_attack) vulnerabilities unless combined with another method of authentication. **Default:** `false` | | `partition_format` | [cloud](/docs/stable/changefeed-sinks.md#cloud-storage-sink) | [`STRING`](/docs/stable/string) | Specify how changefeed [file paths](#general-file-format) are partitioned in cloud storage sinks. Use `partition_format` with the following values: For example: `CREATE CHANGEFEED FOR TABLE users INTO 'gs://...?AUTH...&partition_format=hourly'` **Default:** `daily` | For example: `CREATE CHANGEFEED FOR TABLE users INTO 'gs://...?AUTH...&partition_format=hourly'` \ \ **Default:** `daily` `S3_STORAGE_CLASS` | [Amazon S3 cloud storage sink](/docs/stable/changefeed-sinks.md#amazon-s3) | [`STRING`](/docs/stable/string) | Specify the Amazon S3 storage class for files created by the changefeed. See [Create a changefeed with an S3 storage class](#create-a-changefeed-with-an-s3-storage-class) for the available classes and an example. \ \ **Default:** `STANDARD` New in v24.2:`sasl_aws_iam_role_arn` | [Amazon MSK](/docs/stable/changefeed-sinks.md#amazon-msk) | [`STRING`](/docs/stable/string) | The ARN for the IAM role that has the permissions to create a topic and send data to the topic. For more details on setting up an Amazon MSK cluster with an IAM role, refer to [the AWS documentation](https://docs.aws.amazon.com/msk/latest/developerguide/serverless-getting-started.html). New in v24.2:`sasl_aws_iam_session_name` | [Amazon MSK](/docs/stable/changefeed-sinks.md#amazon-msk) | [`STRING`](/docs/stable/string) | The user-specified string that identifies the session in AWS. New in v24.2:`sasl_aws_region` | [Amazon MSK](/docs/stable/changefeed-sinks.md#amazon-msk) | [`STRING`](/docs/stable/string) | The region of the Amazon MSK cluster. `sasl_client_id` | [Kafka](/docs/stable/changefeed-sinks.md#kafka) | [`STRING`](/docs/stable/string) | Client ID for OAuth authentication from a third-party provider. This parameter is only applicable with `sasl_mechanism=OAUTHBEARER`. `sasl_client_secret` | [Kafka](/docs/stable/changefeed-sinks.md#kafka) | [`STRING`](/docs/stable/string) | Client secret for OAuth authentication from a third-party provider. This parameter is only applicable with `sasl_mechanism=OAUTHBEARER`. **Note:** You must [base64 encode](https://www.base64encode.org/) this value when passing it in as part of a sink URI. `sasl_enabled` | [Amazon MSK](/docs/stable/changefeed-sinks.md#amazon-msk), [Azure Event Hubs](/docs/stable/changefeed-sinks.md#azure-event-hubs), [Kafka](/docs/stable/changefeed-sinks.md#kafka), [Confluent Cloud](/docs/stable/changefeed-sinks.md#confluent-cloud) | [`BOOL`](/docs/stable/bool) | If `true`, set the authentication protocol with the [`sasl_mechanism`](#sasl-mechanism) parameter. You must have `tls_enabled` set to `true` to use SASL.\ \ For Confluent Cloud and Azure Event Hubs sinks, this is set to `true` by default.\ \ **Default:** `false` `sasl_grant_type` | [Kafka](/docs/stable/changefeed-sinks.md#kafka) | [`STRING`](/docs/stable/string) | Override the default OAuth client credentials grant type for other implementations. This parameter is only applicable with `sasl_mechanism=OAUTHBEARER`. `sasl_handshake` | [Azure Event Hubs](/docs/stable/changefeed-sinks.md#azure-event-hubs), [Kafka](/docs/stable/changefeed-sinks.md#kafka), [Confluent Cloud](/docs/stable/changefeed-sinks.md#confluent-cloud) | [`BOOL`](/docs/stable/bool) | For Confluent Cloud and Azure Event Hubs sinks, this is set to `true` by default. `sasl_mechanism` | [Amazon MSK](/docs/stable/changefeed-sinks.md#amazon-msk), [Azure Event Hubs](/docs/stable/changefeed-sinks.md#azure-event-hubs), [Kafka](/docs/stable/changefeed-sinks.md#kafka), [Confluent Cloud](/docs/stable/changefeed-sinks.md#confluent-cloud) | [`STRING`](/docs/stable/string) | Can be set to [`OAUTHBEARER`](https://docs.confluent.io/platform/current/kafka/authentication_sasl/authentication_sasl_oauth.html), [`SCRAM-SHA-256`](https://docs.confluent.io/platform/current/kafka/authentication_sasl/authentication_sasl_scram.html), [`SCRAM-SHA-512`](https://docs.confluent.io/platform/current/kafka/authentication_sasl/authentication_sasl_scram.html), or [`PLAIN`](https://docs.confluent.io/current/kafka/authentication_sasl/authentication_sasl_plain.html). A `sasl_user` and `sasl_password` are required for `PLAIN` and `SCRAM` authentication.\ \ For Amazon MSK clusters, set to [`AWS_MSK_IAM`](/docs/stable/changefeed-sinks#amazon-msk). [`sasl_aws_iam_role_arn`](#sasl-aws-iam-role-arn), [`sasl_aws_iam_session_name`](#sasl-aws-iam-session-name), and [`sasl_aws_region`](#sasl-aws-region) are also required in the sink uri.\ \ Refer to the [Connect to a Changefeed Kafka sink with OAuth Using Okta](/docs/stable/connect-to-a-changefeed-kafka-sink-with-oauth-using-okta.md) tutorial for detail setting up OAuth using Okta.\ \ For Confluent Cloud and Azure Event Hubs sinks, `sasl_mechanism=PLAIN` is required but set automatically by CockroachDB.\ \ **Default:** `PLAIN` `sasl_scopes` | [Kafka](/docs/stable/changefeed-sinks.md#kafka) | [`STRING`](/docs/stable/string) | A list of scopes that the OAuth token should have access for. This parameter is only applicable with `sasl_mechanism=OAUTHBEARER`. `sasl_token_url` | [Kafka](/docs/stable/changefeed-sinks.md#kafka) | [`STRING`](/docs/stable/string) | Client token URL for OAuth authentication from a third-party provider. **Note:** You must [URL encode](https://www.urlencoder.org/) this value before passing in a URI. This parameter is only applicable with `sasl_mechanism=OAUTHBEARER`. `sasl_user` | [Amazon MSK](/docs/stable/changefeed-sinks.md#amazon-msk), [Kafka](/docs/stable/changefeed-sinks.md#kafka) | [`STRING`](/docs/stable/string) | Your SASL username. `sasl_password` | [Amazon MSK](/docs/stable/changefeed-sinks.md#amazon-msk), [Kafka](/docs/stable/changefeed-sinks.md#kafka) | [`STRING`](/docs/stable/string) | Your SASL password. **Note:** Passwords should be [URL encoded](https://wikipedia.org/wiki/Percent-encoding) since the value can contain characters that would cause authentication to fail. `shared_access_key` | [Azure Event Hubs](/docs/stable/changefeed-sinks.md#azure-event-hubs) | [`STRING`](/docs/stable/string) | The URL-encoded key for your Event Hub shared access policy. `shared_access_key_name` | [Azure Event Hubs](/docs/stable/changefeed-sinks.md#azure-event-hubs) | [`STRING`](/docs/stable/string) | The name of your Event Hub shared access policy. `tls_enabled` | [Amazon MSK](/docs/stable/changefeed-sinks.md#amazon-msk), [Kafka](/docs/stable/changefeed-sinks.md#kafka), [Confluent Cloud](/docs/stable/changefeed-sinks.md#confluent-cloud) | [`BOOL`](/docs/stable/bool) | If `true`, enable Transport Layer Security (TLS) on the connection to Kafka. This can be used with a `ca_cert` (see below).\ \ For Confluent Cloud and Azure Event Hubs sinks, this is set to `true` by default.\ \ **Default:** `false` `topic_name` | [Azure Event Hubs](/docs/stable/changefeed-sinks.md#azure-event-hubs), [Kafka](/docs/stable/changefeed-sinks.md#kafka), [Confluent Cloud](/docs/stable/changefeed-sinks.md#confluent-cloud), [GC Pub/Sub](/docs/stable/changefeed-sinks.md) | [`STRING`](/docs/stable/string) | Allows arbitrary topic naming for Kafka and GC Pub/Sub topics. See the [Kafka topic naming limitations](/docs/stable/changefeed-sinks.md#topic-naming) or [GC Pub/Sub topic naming](/docs/stable/changefeed-sinks.md) for detail on supported characters etc. \ \ For example, `CREATE CHANGEFEED FOR foo,bar INTO 'kafka://sink?topic_name=all'` will emit all records to a topic named `all`. Note that schemas will still be registered separately. When using Kafka, this parameter can be combined with the [`topic_prefix` parameter](#topic-prefix) (this is not supported for GC Pub/Sub). \ \ **Default:** table name. `topic_prefix` | [Azure Event Hubs](/docs/stable/changefeed-sinks.md#azure-event-hubs), [Kafka](/docs/stable/changefeed-sinks.md#kafka), [Confluent Cloud](/docs/stable/changefeed-sinks.md#confluent-cloud) | [`STRING`](/docs/stable/string) | Adds a prefix to all topic names.\ \ For example, `CREATE CHANGEFEED FOR TABLE foo INTO 'kafka://...?topic_prefix=bar_'` would emit rows under the topic `bar_foo` instead of `foo`. ### Options | Option | Value | Description | | - | - | - | | `avro_schema_prefix` | Schema prefix name | Provide a namespace for the schema of a table in addition to the default, the table name. This allows multiple databases or clusters to share the same schema registry when the same table name is present in multiple databases. Example: `CREATE CHANGEFEED FOR foo WITH format=avro, confluent_schema_registry='registry_url', avro_schema_prefix='super'` will register subjects as `superfoo-key` and `superfoo-value` with the namespace `super`. | | `compression` | `gzip`, `zstd` | Compress changefeed data files written to a [cloud storage sink](/docs/stable/changefeed-sinks.md#cloud-storage-sink). For compression options when using a Kafka sink, see [Kafka sink configuration](/docs/stable/changefeed-sinks.md#kafka-sink-configuration). | | `confluent_schema_registry` | Schema Registry address | The [Schema Registry](https://docs.confluent.io/current/schema-registry/docs/index.html#sr) address is required to use `avro`. Use the `timeout={duration}` query parameter [`timeout={duration}` query parameter](/docs/stable/create-changefeed) ([duration string](https://pkg.go.dev/time#ParseDuration)) in your Confluent Schema Registry URI to change the default timeout for contacting the schema registry. By default, the timeout is 30 seconds. To connect to Confluent Cloud, use the following URL structure: `'https://{API_KEY_ID}:{API_SECRET_URL_ENCODED}@{CONFLUENT_REGISTRY_URL}:443'`. See the [Stream a Changefeed to a Confluent Cloud Kafka Cluster](/docs/stable/stream-a-changefeed-to-a-confluent-cloud-kafka-cluster.md#step-8-create-a-changefeed) tutorial for further detail. Use the `changefeed.schema_registry.retry_count` metric to measure the number of request retries performed when sending requests to the schema registry. For more detail on monitoring changefeeds, refer to [Monitor and Debug Changefeeds](/docs/stable/monitor-and-debug-changefeeds.md#changefeed-retry-errors). | | `cursor` | [Timestamp](/docs/stable/as-of-system-time.md#parameters) | Emit any changes after the given timestamp. `cursor` does not output the current state of the table first. When `cursor` is not specified, the changefeed starts by doing an initial scan of all the watched rows and emits the current value, then moves to emitting any changes that happen after the scan. The changefeed will encounter an error if you specify a timestamp that is before the configured garbage collection window for the target table. (Refer to [`gc.ttlseconds`](/docs/stable/configure-replication-zones#replication-zone-variables).) With default garbage collection settings, this means you cannot create a changefeed that starts more than [the-default-MVCC-garbage-collection-interval](/docs/stable/configure-replication-zones.md) in the past. You can use `cursor` to [start-a-new-changefeed-where-a-previous-changefeed-ended](#start-a-new-changefeed-where-another-ended). Example: `cursor='1536242855577149065.0000000000'` | | `diff` | N/A | Publish a [`before` field](/docs/stable/changefeed-messages#wrapped-and-diff) with each message, which includes the value of the row before the update was applied. Changefeeds must use the `diff` option with the default [`wrapped` envelope](#envelope) to emit the `before` field. | | New in v24.2: `encode_json_value_null_as_object` | N/A | Emit JSON `NULL` values as `{"__crdb_json_null__": true}` to distinguish these values from SQL `NULL` values. Refer to the [Changefeed Messages](/docs/stable/changefeed-messages.md#json) page for an example. **Note:** When this option is enabled, if the changefeed encounters the literal value `{"__crdb_json_null__": true}` in JSON, it will have the same representation as a JSON `NULL` value and a warning will be printed to the [`DEV` logging channel](/docs/stable/logging#dev). | | `end_time` | [Timestamp](/docs/stable/as-of-system-time.md#parameters) | Indicate the timestamp up to which the changefeed will emit all events and then complete with a `successful` status. Provide a future timestamp to `end_time` in number of nanoseconds since the [Unix epoch](https://wikipedia.org/wiki/Unix_time). For example, `end_time="1655402400000000000"`. You cannot use `end_time` and [`initial_scan = 'only'`](#initial-scan) simultaneously. | | `envelope` | `wrapped` / `bare` / `key_only` / `row` | `wrapped` the default envelope structure for changefeed messages containing an array of the primary key, a top-level field for the type of message, and the current state of the row (or `null` for deleted rows). `bare` removes the `after` key from the changefeed message and stores any metadata in a `crdb` field. When used with `avro` format, `record` will replace the `after` key. **Note:** This is the default envelope format for [CDC-queries](/docs/stable/cdc-queries.md#filter-columns). For an example, refer to [Filter-columns](/docs/stable/cdc-queries.md#filter-columns). `key_only` emits only the key and no value, which is faster if you only need to know the key of the changed row. This envelope option is only supported for [Kafka-sinks](/docs/stable/changefeed-sinks.md#kafka) or sinkless changefeeds. `row` emits the row without any additional metadata fields in the message. This envelope option is only supported in [Kafka-sinks](/docs/stable/changefeed-sinks.md#kafka) or sinkless changefeeds. `row` does not support [`avro` format](#format). Refer to [Responses](/docs/stable/changefeed-messages.md#responses) for more detail on message format. Default: `envelope=wrapped`. Default for [CDC-queries](/docs/stable/cdc-queries.md#filter-columns): `envelope=bare`. | | `execution_locality` | Key-value pairs | Restricts the execution of a changefeed to nodes that match the defined locality filter requirements, e.g., `WITH execution_locality = 'region=us-west-1a,cloud=aws'`. See [Run a changefeed job by locality](/docs/stable/changefeeds-in-multi-region-deployments.md#run-a-changefeed-job-by-locality) for usage and reference detail. | | `format` | `json` / `avro` / `csv` / `parquet` | Format of the emitted message. `avro`: For mappings of CockroachDB types to Avro types, [refer-to-the-table](/docs/stable/changefeed-messages.md#avro-types) and detail on [Avro-limitations](/docs/stable/changefeed-messages.md#avro-limitations). **Note:** [`confluent_schema_registry`](#confluent-schema-registry) is required with `format=avro`. `csv`: You cannot combine `format=csv` with the [`diff`](#diff) or [`resolved`](#resolved) options. Changefeeds use the same CSV format as the [`EXPORT`](/docs/stable/export) statement. Refer to [Export-data-with-changefeeds](/docs/stable/export-data-with-changefeeds.md) for details using these options to create a changefeed as an alternative to `EXPORT`. **Note:** [`initial_scan = 'only'`](#initial-scan) is required with `format=csv`. `parquet`: Cloud storage is the only supported sink. The [`topic_in_value`](#topic-in-value) option is not compatible with `parquet` format. Default: `format=json`. | | `full_table_name` | N/A | Use fully qualified table name in topics, subjects, schemas, and record output instead of the default table name. This can prevent unintended behavior when the same table name is present in multiple databases. **Note:** This option cannot modify existing table names used as topics, subjects, etc., as part of an [`ALTER CHANGEFEED`](/docs/stable/alter-changefeed) statement. To modify a topic, subject, etc., to use a fully qualified table name, create a new changefeed with this option. Example: `CREATE CHANGEFEED FOR foo... WITH full_table_name` will create the topic name `defaultdb.public.foo` instead of `foo`. | | `gc_protect_expires_after` | [Duration string](https://pkg.go.dev/time#ParseDuration) | Automatically expires protected timestamp records that are older than the defined duration. In the case where a changefeed job remains paused, `gc_protect_expires_after` will trigger the underlying protected timestamp record to expire and cancel the changefeed job to prevent accumulation of protected data. Refer to [Protect-Changefeed-Data-from-Garbage-Collection](/docs/stable/protect-changefeed-data.md) for more detail on protecting changefeed data. | | `ignore_disable_changefeed_replication` | [`BOOL`](/docs/stable/bool) | When set to `true`, the changefeed **will emit** events even if CDC filtering for TTL jobs is configured using the `disable_changefeed_replication` [session variable](/docs/stable/set-vars.md), `sql.ttl.changefeed_replication.disabled` [cluster setting](/docs/stable/cluster-settings.md), or the `ttl_disable_changefeed_replication` [table storage parameter](/docs/stable/row-level-ttl.md). Refer to [Filter changefeeds for tables using TTL](#filter-changefeeds-for-tables-using-row-level-ttl) for usage details. | | `initial_scan` | `yes`/`no`/`only` | Control whether or not an initial scan will occur at the start time of a changefeed. Only one `initial_scan` option (`yes`, `no`, or `only`) can be used. If none of these are set, an initial scan will occur if there is no [`cursor`](#cursor), and will not occur if there is one. This preserves the behavior from previous releases. With `initial_scan = 'only'` set, the changefeed job will end with a successful status (`succeeded`) after the initial scan completes. You cannot specify `yes`, `no`, `only` simultaneously. If used in conjunction with `cursor`, an initial scan will be performed at the cursor timestamp. If no `cursor` is specified, the initial scan is performed at `now()`. Although the [`initial_scan` / `no_initial_scan`](/docs/stable/create-changefeed) syntax from previous versions is still supported, you cannot combine the previous and current syntax. Default: `initial_scan = 'yes'` | | `kafka_sink_config` | [`STRING`](/docs/stable/string) | Set fields to configure the required level of message acknowledgement from the Kafka server, the version of the server, and batching parameters for Kafka sinks. Set the message file compression type. See [Kafka sink configuration](/docs/stable/changefeed-sinks.md#kafka-sink-configuration) for more detail on configuring all the available fields for this option. Example: `CREATE CHANGEFEED FOR table INTO 'kafka://localhost:9092' WITH kafka_sink_config='{"Flush": {"MaxMessages": 1, "Frequency": "1s"}, "RequiredAcks": "ONE"}'` | | `key_column` | `'column'` | Override the key used in [message metadata](/docs/stable/changefeed-messages.md#ordering-and-delivery-guarantees). This changes the key hashed to determine downstream partitions. In sinks that support partitioning by message, CockroachDB uses the [32-bit FNV-1a](https://wikipedia.org/wiki/Fowler%E2%80%93Noll%E2%80%93Vo_hash_function) hashing algorithm to determine which partition to send to. **Note:** `key_column` does not preserve ordering of messages from CockroachDB to the downstream sink, therefore you must also include the [`unordered`](#unordered) option in your changefeed creation statement. It does not affect per-key [ordering guarantees](/docs/stable/changefeed-messages.md#ordering-and-delivery-guarantees) or the output of [`key_in_value`](#key-in-value). See the [Define a key to determine the changefeed sink partition](#define-a-key-to-determine-the-changefeed-sink-partition) example. | | `key_in_value` | N/A | Add a primary key array to the emitted message. This makes the [primary key](/docs/stable/primary-key.md) of a deleted row recoverable in sinks where each message has a value but not a key (most have a key and value in each message). `key_in_value` is automatically used for [cloud storage sinks](/docs/stable/changefeed-sinks.md#cloud-storage-sink), [webhook sinks](/docs/stable/changefeed-sinks.md#webhook-sink), and [GC Pub/Sub sinks](/docs/stable/changefeed-sinks.md). | | `lagging_ranges_threshold` | [Duration string](https://pkg.go.dev/time#ParseDuration) | Set a duration from the present that determines the length of time a range is considered to be lagging behind, which will then track in the [`lagging_ranges`](/docs/stable/monitor-and-debug-changefeeds#changefeed-retry-errors) metric. Note that ranges undergoing an [initial scan](#initial-scan) for longer than the threshold duration are considered to be lagging. Starting a changefeed with an initial scan on a large table will likely increment the metric for each range in the table. As ranges complete the initial scan, the number of ranges lagging behind will decrease. **Default:** `3m` | | `lagging_ranges_polling_interval` | [Duration string](https://pkg.go.dev/time#ParseDuration) | Set the interval rate for when lagging ranges are checked and the `lagging_ranges` metric is updated. Polling adds latency to the `lagging_ranges` metric being updated. For example, if a range falls behind by 3 minutes, the metric may not update until an additional minute afterward. **Default:** `1m` | | `metrics_label` | [`STRING`](/docs/stable/string) | Define a metrics label to which the metrics for one or multiple changefeeds increment. All changefeeds also have their metrics aggregated. The maximum length of a label is 128 bytes. There is a limit of 1024 unique labels. `WITH metrics_label=label_name` For more detail on usage and considerations, see [Using changefeed metrics labels](/docs/stable/monitor-and-debug-changefeeds.md#using-changefeed-metrics-labels). | | `min_checkpoint_frequency` | [Duration string](https://pkg.go.dev/time#ParseDuration) | Controls how often a node's changefeed [aggregator](/docs/stable/how-does-an-enterprise-changefeed-work) will flush their progress to the [coordinating changefeed node](/docs/stable/how-does-an-enterprise-changefeed-work). A node's changefeed aggregator will wait at least the specified duration between sending progress updates for the ranges it is watching to the coordinator. This can help you control the flush frequency of higher latency sinks to achieve better throughput. However, more frequent checkpointing can increase CPU usage. If this is set to `0s`, a node will flush messages as long as the high-water mark has increased for the ranges that particular node is processing. If a changefeed is resumed, then `min_checkpoint_frequency` is the amount of time that changefeed will need to catch up. That is, it could emit [duplicate messages](/docs/stable/changefeed-messages.md#duplicate-messages) during this time. **Note:** [`resolved`](#resolved) messages will not be emitted more frequently than the configured `min_checkpoint_frequency` (but may be emitted less frequently). If you require `resolved` messages more frequently than `30s`, you must configure `min_checkpoint_frequency` to at least the desired `resolved` message frequency. For more details, refer to [Resolved message frequency](/docs/stable/changefeed-messages.md#resolved-timestamp-frequency). **Default:** `30s` | | `mvcc_timestamp` | N/A | Include the [MVCC](/docs/stable/architecture/storage-layer.md#mvcc) timestamp for each emitted row in a changefeed. With the `mvcc_timestamp` option, each emitted row will always contain its MVCC timestamp, even during the changefeed's initial backfill. | | `on_error` | `pause` / `fail` | Use `on_error=pause` to pause the changefeed when encountering **non**-retryable errors. `on_error=pause` will pause the changefeed instead of sending it into a terminal failure state. **Note:** Retryable errors will continue to be retried with this option specified. Use with [`protect_data_from_gc_on_pause`](#protect-data-from-gc-on-pause) to protect changes from [garbage collection](/docs/stable/configure-replication-zones.md). If a changefeed with `on_error=pause` is running when a watched table is [truncated](/docs/stable/truncate.md), the changefeed will pause but will not be able to resume reads from that table. Using [`ALTER CHANGEFEED`](/docs/stable/alter-changefeed) to drop the table from the changefeed and then [resuming the job](/docs/stable/resume-job.md) will work, but you cannot add the same table to the changefeed again. Instead, you will need to [create a new changefeed](#start-a-new-changefeed-where-another-ended) for that table. Default: `on_error=fail` | | `protect_data_from_gc_on_pause` | N/A | This option is deprecated as of v23.2 and will be removed in a future release. When a [changefeed is paused](/docs/stable/pause-job.md), ensure that the data needed to [resume the changefeed](/docs/stable/resume-job.md) is not garbage collected. If `protect_data_from_gc_on_pause` is **unset**, pausing the changefeed will release the existing protected timestamp records. It is also important to note that pausing and adding `protect_data_from_gc_on_pause` to a changefeed will not protect data if the [garbage collection](/docs/stable/configure-replication-zones.md) window has already passed. Use with [`on_error=pause`](#on-error) to protect changes from garbage collection when encountering non-retryable errors. Refer to [Protect Changefeed Data from Garbage Collection](/docs/stable/protect-changefeed-data.md) for more detail on protecting changefeed data. **Note:** If you use this option, changefeeds that are left paused for long periods of time can prevent garbage collection. Use with the [`gc_protect_expires_after`](#gc-protect-expires-after) option to set a limit for protected data and for how long a changefeed will remain paused. | | `pubsub_sink_config` | [`STRING`](/docs/stable/string) | Set fields to configure sink batching and retries. The schema is as follows: `{ "Flush": { "Messages": ..., "Bytes": ..., "Frequency": ..., }, "Retry": {"Max": ..., "Backoff": ..., } }`. **Note** that if either `Messages` or `Bytes` are nonzero, then a non-zero value for `Frequency` must be provided. Refer to [Pub/Sub sink configuration](/docs/stable/changefeed-sinks.md) for more details on using this option. | | `resolved` | [Duration string](https://pkg.go.dev/time#ParseDuration) | Emit [resolved timestamps](/docs/stable/changefeed-messages.md#resolved-messages) in a format dependent on the connected sink. Resolved timestamps do not emit until the changefeed job's progress has been checkpointed. Set a minimum amount of time that the changefeed's high-water mark (overall resolved timestamp) must advance by before another resolved timestamp is emitted. Example: `resolved='10s'`. This option will **only** emit a resolved timestamp if the timestamp has advanced (and by at least the optional duration, if set). If a duration is unspecified, all resolved timestamps are emitted as the high-water mark advances. **Note:** If you set `resolved` lower than `30s`, then you **must** also set [`min_checkpoint_frequency`](#min-checkpoint-frequency) to at minimum the same value as `resolved`, because `resolved` messages may be emitted less frequently than `min_checkpoint_frequency`, but cannot be emitted more frequently. Refer to [Resolved messages](/docs/stable/changefeed-messages.md#resolved-messages) for more detail. | | `schema_change_events` | `default` / `column_changes` | The type of schema change event that triggers the behavior specified by the `schema_change_policy` option:Default: `schema_change_events=default` | | `schema_change_policy` | `backfill` / `nobackfill` / `stop` | The behavior to take when an event specified by the `schema_change_events` option occurs:Default: `schema_change_policy=backfill` | | `split_column_families` | N/A | Use this option to create a changefeed on a table with multiple [column families](/docs/stable/column-families.md). The changefeed will emit messages for each of the table's column families. See [Changefeeds on tables with column families](/docs/stable/changefeeds-on-tables-with-column-families.md) for more usage detail. | | `topic_in_value` | [`BOOL`](/docs/stable/bool) | Set to include the topic in each emitted row update. This option is automatically set for [webhook sinks](/docs/stable/changefeed-sinks.md#webhook-sink). **Note:** `topic_in_value` is not compatible with changefeeds running in [`parquet` format](#format). | | `unordered` | N/A | Run a changefeed to [Google Cloud Pub/Sub](/docs/stable/changefeed-sinks.md) without specifying a region. You must include the `unordered` option with [`key_column`](#key-column) in your changefeed creation statement. You cannot use `unordered` with `resolved`, because resolved timestamps may not be correct in unordered mode. | | `updated` | N/A | Include updated timestamps with each row. If a `cursor` is provided, the "updated" timestamps will match the [MVCC](/docs/stable/architecture/storage-layer.md#mvcc) timestamps of the emitted rows, and there is no initial scan. If a `cursor` is not provided, the changefeed will perform an initial scan (as of the time the changefeed was created), and the "updated" timestamp for each change record emitted in the initial scan will be the timestamp of the initial scan. Similarly, when a [backfill is performed for a schema change](/docs/stable/changefeed-messages.md#schema-changes-with-column-backfill), the "updated" timestamp is set to the first timestamp for when the new schema is valid. | | `virtual_columns` | `STRING` | Changefeeds omit [virtual computed columns](/docs/stable/computed-columns.md) from emitted [messages](/docs/stable/changefeed-messages.md#ordering-and-delivery-guarantees) by default. To maintain the behavior of previous CockroachDB versions where the changefeed would emit [`NULL`](/docs/stable/null-handling) values for virtual computed columns, set `virtual_columns = "null"` when you start a changefeed. You may also define `virtual_columns = "omitted"`, though this is already the default behavior for v22.1+. If you do not set `"omitted"` on a table with virtual computed columns when you create a changefeed, you will receive a warning that changefeeds will filter out virtual computed values. **Default:** `"omitted"` | | `webhook_auth_header` | [`STRING`](/docs/stable/string) | Pass a value (password, token etc.) to the HTTP [Authorization header](https://developer.mozilla.org/docs/Web/HTTP/Headers/Authorization) with a webhook request for a "Basic" HTTP authentication scheme. Example: With a username of "user" and password of "pwd", add a colon between "user:pwd" and then base64 encode, which results in "dXNlcjpwd2Q=". `WITH webhook_auth_header='Basic dXNlcjpwd2Q='`. | | `webhook_client_timeout` | [`INTERVAL`](/docs/stable/interval) | If a response is not recorded from the sink within this timeframe, it will error and retry to connect. Note this must be a positive value. **Default:** `"3s"` | | `webhook_sink_config` | [`STRING`](/docs/stable/string) | Set fields to configure sink batching and retries. The schema is as follows: `{ "Flush": { "Messages": ..., "Bytes": ..., "Frequency": ..., }, "Retry": {"Max": ..., "Backoff": ..., } }`. **Note** that if either `Messages` or `Bytes` are nonzero, then a non-zero value for `Frequency` must be provided. See [Webhook sink configuration](/docs/stable/changefeed-sinks.md#webhook-sink-configuration) for more details on using this option. | > [!NOTE] > > Using the `format=avro`, `envelope=key_only`, and `updated` options together is rejected. `envelope=key_only` prevents any rows with updated fields from being emitted, which makes the `updated` option meaningless. ## Files The files emitted to a sink use the following naming conventions: * [General file format](#general-file-format) * [Resolved file format](#resolved-file-format) > [!NOTE] > > The timestamp format is `YYYYMMDDHHMMSSNNNNNNNNNLLLLLLLLLL`. ### 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](/docs/stable/changefeed-sinks.md#cloud-storage-sink), you can specify a partition format for your files using the [`partition_format`](#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](/docs/stable/create-and-configure-changefeeds#enable-rangefeeds). If you are working on a CockroachDB Standard or Basic 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](#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](/docs/stable/changefeed-sinks.md) 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`](/docs/stable/create-external-connection) page. We recommend limiting the number of changefeeds per cluster to 80. ### Create a changefeed connected to a sink ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} CREATE CHANGEFEED FOR TABLE table_name, table_name2, table_name3 INTO 'scheme://host:port' WITH updated, resolved; ``` You can connect a changefeed to the following sinks: * [Amazon MSK](#amazon-msk) [Amazon MSK](/docs/stable/changefeed-sinks.md#amazon-msk) * [Apache Pulsar](#apache-pulsar) (in Preview) [Apache Pulsar](/docs/stable/changefeed-sinks.md#apache-pulsar) (in Preview) * [Azure Event Hubs](#azure-event-hubs) [Azure Event Hubs](/docs/stable/changefeed-sinks.md#azure-event-hubs) * [Cloud Storage](#cloud-storage-sink) / HTTP [Cloud Storage](/docs/stable/changefeed-sinks.md#cloud-storage-sink) / HTTP * [Confluent Cloud](#confluent-cloud) [Confluent Cloud](/docs/stable/changefeed-sinks.md#confluent-cloud) * [Google Cloud Pub/Sub](#google-cloud-pub-sub) [Google Cloud Pub/Sub](/docs/stable/changefeed-sinks.md) * [Kafka](#kafka) [Kafka](/docs/stable/changefeed-sinks.md#kafka) * [Webhook](#webhook-sink) [Webhook](/docs/stable/changefeed-sinks.md#webhook-sink) ### Create a changefeed that filters and transforms change data [CDC queries](/docs/stable/cdc-queries.md#filter-columns) can filter and transform change data before emitting it to a sink or [a SQL client](#create-a-sinkless-changefeed). You can adapt a changefeed with CDC queries by including `SELECT` and `WHERE` clauses in your `CREATE` statement: ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} 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: ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} 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: ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} 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](/docs/stable/create-external-connection.md) 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](/docs/stable/create-changefeed#required-privileges) on a set of tables can only create changefeeds into [external connections](/docs/stable/create-external-connection.md). External connections support all changefeed sinks. ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} 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: ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} 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: ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} 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'); ``` ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} 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](/docs/stable/cluster-settings.md) 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`: ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} 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](/docs/stable/row-level-ttl.md) deletes only, refer to [Change Data Capture Queries](/docs/stable/cdc-queries.md#reference-ttl-in-a-cdc-query). ### Disallow schema changes on tables to improve changefeed performance Use the `schema_locked` [storage parameter](/docs/stable/with-storage-parameter.md) to disallow [schema changes](/docs/stable/online-schema-changes.md) on a watched table, which helps to decrease the latency between a write committing to a table and it emitting to the [changefeed's sink](/docs/stable/changefeed-sinks.md). You can lock the table before creating a changefeed or while a changefeed is running, which will enable the performance improvement for changefeeds watching the particular table. Enable `schema_locked` on the watched table with the [`ALTER TABLE`](/docs/stable/alter-table) statement: ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} ALTER TABLE watched_table SET (schema_locked = true); ``` While `schema_locked` is enabled on a table, attempted schema changes on the table will be rejected and an error returned. If you need to run a schema change on the locked table, unlock the table with `schema_locked = false`, complete the schema change, and then lock the table again with `schema_locked = true`. The changefeed will run as normal while `schema_locked = false`, but it will not benefit from the performance optimization. ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} ALTER TABLE watched_table SET (schema_locked = false); ``` ### Manage a changefeed For Enterprise changefeeds, use [`SHOW CHANGEFEED JOBS`](/docs/stable/show-jobs) to check the status of your changefeed jobs: ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} SHOW CHANGEFEED JOBS; ``` Use the following SQL statements to pause, resume, or cancel a changefeed. #### Pause a changefeed ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} PAUSE JOB job_id; ``` For more information, see [`PAUSE JOB`](/docs/stable/pause-job). #### Resume a paused changefeed ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} RESUME JOB job_id; ``` For more information, see [`RESUME JOB`](/docs/stable/resume-job). #### Cancel a changefeed ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} CANCEL JOB job_id; ``` For more information, see [`CANCEL JOB`](/docs/stable/cancel-job). #### Modify a changefeed To modify an Enterprise changefeed, [pause](/docs/stable/create-and-configure-changefeeds.md#pause) the job and then use: ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} ALTER CHANGEFEED job_id {ADD table DROP table SET option UNSET option}; ``` You can add new table targets, remove them, set new [changefeed options](/docs/stable/create-changefeed.md#options), and unset them. For more information, see [`ALTER CHANGEFEED`](/docs/stable/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: ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} 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`](/docs/stable/show-jobs#show-changefeed-jobs). To resume all running changefeeds: ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} 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](/docs/stable/create-and-configure-changefeeds#known-limitations), and you need to restart the changefeed. 1. Use [`SHOW CHANGEFEED JOB`](/docs/stable/show-jobs#show-changefeed-jobs) to find the [high-water timestamp](/docs/stable/monitor-and-debug-changefeeds.md#monitor-a-changefeed) for the ended changefeed: ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} 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. 1. Use the `high_water_timestamp` to start the new changefeed: ```sql theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}} CREATE CHANGEFEED FOR TABLE table_name, table_name2, table_name3 INTO 'scheme//host:port' WITH cursor = '