Adding or removing a changefeed

ADD CHANGEFEED <name> WITH (<option> = <value>[, ...]): Adds a changefeed with the specified name and options.

Changefeed options

• MODE: Operation mode. Specifies what to write to a changefeed each time table data is altered.
  • KEYS_ONLY: Only the primary key components and change flag are written.
  • UPDATES: Updated column values that result from updates are written.
  • NEW_IMAGE: Any column values resulting from updates are written.
  • OLD_IMAGE: Any column values before updates are written.
  • NEW_AND_OLD_IMAGES: A combination of NEW_IMAGE and OLD_IMAGE modes. Any column values prior to and resulting from updates are written.
• FORMAT: Data write format.
  • JSON: Write data in JSON format.
  • DEBEZIUM_JSON: Write data in the Debezium-like JSON format.
• VIRTUAL_TIMESTAMPS: Enabling/disabling virtual timestamps. Disabled by default.
• BARRIERS_INTERVAL — barrier emission interval. The value type is Interval. Disabled by default.
• RETENTION_PERIOD: Record retention period. The value type is Interval and the default value is 24 hours (Interval('PT24H')).
• TOPIC_AUTO_PARTITIONING: Topic autopartitioning mode:
  • ENABLED – An autopartitioned topic will be created for this changefeed. The number of partitions in such a topic increases automatically as the table update rate increases. Topic autopartitioning parameters can be configured.
  • DISABLED – A topic without autopartitioning will be created for this changefeed. This is the default value.
• TOPIC_MIN_ACTIVE_PARTITIONS: The initial number of topic partitions. By default, the initial number of topic partitions is equal to the number of table partitions. For autopartitioned topics, the number of partitions increases automatically as the table update rate increases.
• INITIAL_SCAN: Enables/disables initial table scan. Disabled by default.

The code below adds a changefeed named updates_feed, where the values of updated table columns will be exported in JSON format:

ALTER TABLE `series` ADD CHANGEFEED `updates_feed` WITH (
    FORMAT = 'JSON',
    MODE = 'UPDATES'
);

Records in this changefeed will be stored for 24 hours (default value). The code in the following example will create a changefeed with a record retention period of 12 hours:

ALTER TABLE `series` ADD CHANGEFEED `updates_feed` WITH (
    FORMAT = 'JSON',
    MODE = 'UPDATES',
    RETENTION_PERIOD = Interval('PT12H')
);

The example of creating a changefeed with enabled virtual timestamps:

ALTER TABLE `series` ADD CHANGEFEED `updates_feed` WITH (
    FORMAT = 'JSON',
    MODE = 'UPDATES',
    VIRTUAL_TIMESTAMPS = TRUE
);

The example of creating a changefeed with virtual timestamps and barriers every 10 seconds:

ALTER TABLE `series` ADD CHANGEFEED `updates_feed` WITH (
    FORMAT = 'JSON',
    MODE = 'UPDATES',
    VIRTUAL_TIMESTAMPS = TRUE,
    BARRIERS_INTERVAL = Interval('PT10S')
);

Example of creating a changefeed with initial scan:

ALTER TABLE `series` ADD CHANGEFEED `updates_feed` WITH (
    FORMAT = 'JSON',
    MODE = 'UPDATES',
    INITIAL_SCAN = TRUE
);

Example of creating a changefeed with autopartitioning:

ALTER TABLE `series` ADD CHANGEFEED `updates_feed` WITH (
    FORMAT = 'JSON',
    MODE = 'UPDATES',
    TOPIC_AUTO_PARTITIONING = 'ENABLED',
    TOPIC_MIN_ACTIVE_PARTITIONS = 2
);

DROP CHANGEFEED: Deletes the changefeed with the specified name. The code below deletes the updates_feed changefeed:

ALTER TABLE `series` DROP CHANGEFEED `updates_feed`;