Consumers and Query Service

Manage stored consumer progress and start a lookup query service.

See Procedures for Flink version requirements, argument conventions, and catalog selection.

reset_consumer

To reset or delete consumer. Arguments:

• table: the target table identifier. Cannot be empty.

• consumerId: consumer to be reset or deleted.

• nextSnapshotId (Long): the new next snapshot id of the consumer.

Syntax

-- Use named argument
CALL [catalog.]sys.reset_consumer(
    `table` => 'identifier',
    consumer_id => 'consumerId',
    next_snapshot_id => 'nextSnapshotId'
);

-- Use indexed argument
-- reset the new next snapshot id in the consumer
CALL [catalog.]sys.reset_consumer('identifier', 'consumerId', nextSnapshotId);

-- delete consumer
CALL [catalog.]sys.reset_consumer('identifier', 'consumerId');

Example

CALL sys.reset_consumer(
    `table` => 'default.T',
    consumer_id => 'myid',
    next_snapshot_id => cast(10 as bigint)
);

clear_consumers

To reset or delete consumer. Arguments:

• table: the target table identifier. Cannot be empty.

• includingConsumers: consumers to be cleared.

• excludingConsumers: consumers which not to be cleared.

Syntax

-- Use named argument
CALL [catalog.]sys.clear_consumers(
    `table` => 'identifier',
    including_consumers => 'includingConsumers',
    excluding_consumers => 'excludingConsumers'
);

-- Use indexed argument
-- clear all consumers in the table
CALL [catalog.]sys.clear_consumers('identifier');

-- clear some consumers in the table (accept regular expression)
CALL [catalog.]sys.clear_consumers('identifier', 'includingConsumers');

-- exclude some consumers (accept regular expression)
CALL [catalog.]sys.clear_consumers('identifier', 'includingConsumers', 'excludingConsumers');

Example

CALL sys.clear_consumers(`table` => 'default.T');

CALL sys.clear_consumers(`table` => 'default.T', including_consumers => 'myid.*');

CALL sys.clear_consumers(table => 'default.T', including_consumers => '', excluding_consumers => 'myid1.*');

CALL sys.clear_consumers(
    table => 'default.T',
    including_consumers => 'myid.*',
    excluding_consumers => 'myid1.*'
);

query_service

Start a query service for a table. Arguments:

• table: the target table identifier.

• parallelism: the query service parallelism.

Syntax

CALL [catalog.]sys.query_service(`table` => 'identifier', parallelism => parallelism);

Example

CALL sys.query_service(`table` => 'default.T', parallelism => 4);