REPLACE SINK

​

Syntax

REPLACE SINK sink_name
FROM sink_from
WITH (
   connector = 'connector_name',
   connector_parameter = 'value',
   [ snapshot = 'false' ],...
)
[ FORMAT data_format ENCODE data_encode [ (
    format_parameter = 'value'
) ] ];

​

Parameters

​

Behavior and limitations

• REPLACE SINK replaces an existing sink as a cut-over operation. After replacement, the sink starts from the cut-over barrier and only forwards new incremental changes from the new upstream.
• Explicit privileges granted on the existing sink are not migrated to the replacement sink. Re-grant these privileges after replacement.
• Use REPLACE SINK, not CREATE OR REPLACE SINK.
• REPLACE SINK currently supports FROM only. REPLACE SINK ... AS SELECT ... is not supported.
• REPLACE SINK does not support IF NOT EXISTS.
• An existing exactly-once sink cannot be replaced.
• Sinks created with CREATE SINK ... INTO cannot be replaced.
• Sinks with automatic schema change enabled cannot be replaced, and the replacement definition cannot set auto.schema.change to true.
• REPLACE SINK does not support the since_timestamp option.

​

Example

REPLACE SINK orders_sink
FROM orders_v2
WITH (
  connector = 'kafka',
  properties.bootstrap.server = 'localhost:9092',
  topic = 'orders',
  primary_key = 'id',
  snapshot = 'false'
)
FORMAT UPSERT ENCODE JSON;

​

See also