ALTER SINK

To modify the SQL definition of a sink, please refer to Alter a streaming job.

Syntax

ALTER SINK sink_name
    alter_option;

alter_option depends on the operation you want to perform on the sink. For all supported clauses, see the sections below.

Clause

`OWNER TO`

ALTER SINK sink_name
    OWNER TO new_user;

Parameter or clause	Description
OWNER TO	This clause changes the owner of the sink. This will cascadingly change all related internal-objects as well.
new_user	The new owner you want to assign to the sink.

-- Change the owner of the sink named "sink1" to user "user1"
ALTER SINK sink1 OWNER TO user1;

`SET SCHEMA`

ALTER SINK sink_name
    SET SCHEMA schema_name;

Parameter or clause	Description
SET SCHEMA	This clause moves the sink to a different schema.
schema_name	The name of the schema to which the sink will be moved.

-- Move the sink named "test_sink" to the schema named "test_schema"
ALTER SINK test_sink SET SCHEMA test_schema;

`SET PARALLELISM`

ALTER SINK sink_name
SET PARALLELISM { TO | = } parallelism_number;

Parameter or clause	Description
SET PARALLELISM	This clause controls the degree of parallelism for the targeted streaming job.
parallelism_number	Can be ADAPTIVE or a fixed number (e.g., 1, 2, 3). Setting it to ADAPTIVE expands the job’s parallelism to all available units, while a fixed number locks it at that value. Setting it to 0 is equivalent to ADAPTIVE. The maximum allowed value is determined by the `max_parallelism` of the job. For more information, see Configuring maximum parallelism.

-- Set the parallelism of the sink "s" to 4.
ALTER SINK s SET PARALLELISM = 4;

`SET BACKFILL_PARALLELISM`

ALTER SINK sink_name
    SET BACKFILL_PARALLELISM { TO | = } parallelism_value [ DEFERRED ];

This statement controls the degree of parallelism used during the backfill phase of a sink. It lets you tune resource usage for backfilling independently from the sink’s steady-state parallelism.

Parameter or clause	Description
SET BACKFILL_PARALLELISM	Sets the parallelism for the backfill phase of the streaming job.
parallelism_value	Can be `ADAPTIVE` or a fixed number (e.g., 1, 2, 3). Setting it to `ADAPTIVE` or `0` expands backfill parallelism to all available units. A fixed number locks it at that value.
DEFERRED	Optional. When specified, the change takes effect after the current backfill completes, without interrupting an ongoing backfill. If omitted, the change is applied immediately.

-- Set backfill parallelism to a fixed value
ALTER SINK s1 SET BACKFILL_PARALLELISM = 4;

-- Set backfill parallelism to adaptive
ALTER SINK s1 SET BACKFILL_PARALLELISM = ADAPTIVE;

-- Defer the parallelism change until the current backfill completes
ALTER SINK s1 SET BACKFILL_PARALLELISM = 4 DEFERRED;

`SET SINK_RATE_LIMIT`

ALTER SINK sink_name
    SET SINK_RATE_LIMIT { TO | = } { default | rate_limit_number };

Use this statement to modify the rate limit of a sink. For the specific value of SINK_RATE_LIMIT, refer to How to view runtime parameters.

Example

-- Alter the rate limit of a sink to default
ALTER SINK s1 SET SINK_RATE_LIMIT = default;

-- Alter the rate limit of a sink to 1000
ALTER SINK s1 SET SINK_RATE_LIMIT = 1000;

`SET BACKFILL_RATE_LIMIT`

ALTER SINK sink_name
    SET BACKFILL_RATE_LIMIT { TO | = } { default | rate_limit_number };

This statement controls the rate limit of a sink’s backfilling process, which reads the historical data of the upstream relation when the sink is created. It is separate from SINK_RATE_LIMIT, which throttles writes to the external system. For the specific value of BACKFILL_RATE_LIMIT, refer to How to view runtime parameters.

Examples

-- Pause the backfill
ALTER SINK s1 SET BACKFILL_RATE_LIMIT=0;

-- Set backfill rate limit to 1000
ALTER SINK s1 SET BACKFILL_RATE_LIMIT=1000;

-- Disable the backfill rate limit
ALTER SINK s1 SET BACKFILL_RATE_LIMIT=DEFAULT;

For how the backfill rate limit behaves and how to inspect it, see Rate limiting.

`SET RESOURCE_GROUP`

ALTER SINK sink_name
    SET RESOURCE_GROUP { TO | = } { DEFAULT | resource_group_name } [ DEFERRED ];

ALTER SINK sink_name
    RESET RESOURCE_GROUP [ DEFERRED ];

Use this statement to change the resource group used by the sink’s streaming job.

Parameter or clause	Description
SET RESOURCE_GROUP	Assigns the sink to a specific resource group.
RESET RESOURCE_GROUP	Resets the sink to the default resource group.
resource_group_name	The target resource group.
DEFERRED	Optional. Defers applying the resource group change.

-- Assign the sink to a resource group
ALTER SINK orders_sink SET RESOURCE_GROUP = high_priority;

-- Reset the sink to the default resource group
ALTER SINK orders_sink RESET RESOURCE_GROUP DEFERRED;

`RENAME TO`

ALTER SINK sink_name
    RENAME TO new_name;

Parameter or clause	Description
RENAME TO	This clause changes the name of the sink.
new_name	The new name of the sink.

-- Change the name of the sink named "sink0" to "sink1"
ALTER SINK sink0 RENAME TO sink1;

`SWAP WITH`

ALTER SINK name
SWAP WITH target_name;

Parameter	Description
name	The current name of the sink to swap.
target_name	The target name of the sink you want to swap with.

-- Swap the names of the log_sink sink and the error_sink sink.
ALTER SINK log_sink
SWAP WITH error_sink;

`CONNECTOR WITH`

Added in v2.5.0.

ALTER SINK sink_name CONNECTOR WITH (
  'property_name' = 'value',
  ...
);

Parameter	Description
CONNECTOR WITH	Allows you to update connector-specific properties for an existing sink without recreating it.
property_name	The properties vary by connector type: For ClickHouse, DeltaLake, and StarRocks, the supported parameter is `commit_checkpoint_interval`. For Iceberg, the supported parameter is `commit_checkpoint_interval`. For Kafka, the supported parameters include: `properties.allow.auto.create.topics`, `properties.batch.num.messages`, `properties.batch.size`, `properties.enable.idempotence`, `properties.max.in.flight.requests.per.connection`, `properties.message.max.bytes`, `properties.message.send.max.retries`, `properties.message.timeout.ms`, `properties.queue.buffering.max.kbytes`, `properties.queue.buffering.max.messages`, `properties.queue.buffering.max.ms`, `properties.reconnect.backoff.ms`, `properties.reconnect.backoff.max.ms`, `properties.request.required.acks`, `properties.retry.backoff.ms`, `properties.retry.backoff.max.ms`, `properties.receive.message.max.bytes`, `properties.socket.connection.setup.timeout.ms`. For OpenSearch, the supported parameters are `batch_num_messages`, `batch_size_kb`, and `concurrent_requests`. For Pulsar, the supported parameter is `properties.routing.mode`.

-- Enable idempotence for the sink
ALTER SINK my_kafka_sink CONNECTOR WITH (
  'properties.enable.idempotence' = 'true'
);

-- Update batching settings for an OpenSearch sink
ALTER SINK my_opensearch_sink CONNECTOR WITH (
  'batch_num_messages' = '500',
  'batch_size_kb' = '4096',
  'concurrent_requests' = '2'
);

-- Change the routing policy of a Pulsar sink
ALTER SINK my_pulsar_sink CONNECTOR WITH (
  'properties.routing.mode' = 'round_robin'
);

SQL reference

Python SDK

User-defined functions

Syntax

Clause

`OWNER TO`

`SET SCHEMA`

`SET PARALLELISM`

`SET BACKFILL_PARALLELISM`

`SET SINK_RATE_LIMIT`

`SET BACKFILL_RATE_LIMIT`

`SET RESOURCE_GROUP`

`RENAME TO`

`SWAP WITH`

`CONNECTOR WITH`

​Syntax

​Clause

​OWNER TO

​SET SCHEMA

​SET PARALLELISM

​SET BACKFILL_PARALLELISM

​SET SINK_RATE_LIMIT

​SET BACKFILL_RATE_LIMIT

​SET RESOURCE_GROUP

​RENAME TO

​SWAP WITH

​CONNECTOR WITH

Syntax

Clause

`OWNER TO`

`SET SCHEMA`

`SET PARALLELISM`

`SET BACKFILL_PARALLELISM`

`SET SINK_RATE_LIMIT`

`SET BACKFILL_RATE_LIMIT`

`SET RESOURCE_GROUP`

`RENAME TO`

`SWAP WITH`

`CONNECTOR WITH`