SET Statement¶

Important

Confluent Cloud for Apache Flink®️ is currently available for Preview. A Preview feature is a Confluent Cloud component that is being introduced to gain early feedback from developers. Preview features can be used for evaluation and non-production testing purposes or to provide feedback to Confluent. The warranty, SLA, and Support Services provisions of your agreement with Confluent do not apply to Preview features. Confluent may discontinue providing Preview releases of the Preview features at any time in Confluent’s sole discretion. Check out Getting Help for questions, feedback and requests.

For Flink SQL features and limitations in the preview program, see Notable Limitations in Public Preview.

Confluent Cloud for Apache Flink®️ enables setting Flink SQL shell properties to different values.

Syntax¶

SET 'key' = 'value';

Description¶

Modify or list the Flink SQL shell configuration.

If no key and value are specified, SET prints all of the properties that you have assigned for the session.

To reset a session property to its default value, use the RESET Statement.

Example¶

The following examples show how to run a SET statement in the Flink SQL shell.

SET 'table.local-time-zone' = 'America/Los_Angeles';

Your output should resemble:

Statement successfully submitted.
Statement phase is COMPLETED.
configuration updated successfully.

To list the current session settings, run the SET command with no parameters.

SET;

Your output should resemble:

 Statement successfully submitted.
 Statement phase is COMPLETED.
+-----------------------+--------------------------+
|          Key          |          Value           |
+-----------------------+--------------------------+
| catalog               | default (default)        |
| default_database      | <your_cluster> (default) |
| table.local-time-zone | America/Los_Angeles      |
+-----------------------+--------------------------+

Available SET Statements¶

These are the available SET statements in Confluent Cloud for Flink SQL.

Key	Default	Type	Description
client.results-timeout	600000	Long	Total amount of time in milliseconds to wait before timing out the request waiting for results to be ready.
client.statement-name	(None)	String	Give your Flink statement a meaningful name that can help you identify it more easily. Instead of 123e4567-e89b-12d3, this sets the statement name to the given value. The name resets itself after successful submission to avoid naming conflicts.
sql.current-catalog	(None)	String	Default catalog to use. Is also set by USE CATALOG [catalog_name]
sql.current-database	(None)	String	Default database to use. Is also set by USE [database_id]
sql.local-time-zone	Local device time zone	String	Specify the local time zone offset for Flink SQL. When converting to data types that don’t include a time zone (e.g. TIMESTAMP, TIME, or simply STRING), this time zone is used. The input of option is either a full name such as “America/Los_Angeles”, or a custom timezone id such as “GMT-08:00”.
sql.state-ttl	0 ms	Duration	Specifies a minimum time interval for how long idle state (i.e. state which was not updated), will be retained. State will never be cleared until it was idle for less than the minimum time, and will be cleared at some time after it was idle.
sql.tables.scan.bounded.mode	(None)	Enum	This specifies the bounded mode for Kafka consumer. The valid enumerations are: group-offsets: bounded by committed offsets of a specific consumer group. This is evaluated at the start of consumption from a given partition. latest-offset: bounded by latest offsets. This is evaluated at the start of consumption from a given partition. timestamp: bounded by a user-supplied timestamp. If this config option is not set, the default is an unbounded table. If timestamp is specified, config option sql.tables.scan.bounded.timestamp-millis is required to specify a specific bounded timestamp in milliseconds since January 1, 1970 00:00:00.000 GMT.
sql.tables.scan.bounded.timestamp-millis	(None)	Long	End at the specified epoch timestamp (milliseconds) used in case of ‘timestamp’ bounded mode.
sql.tables.scan.idle-timeout	0 ms	Duration	When a source do not receive any elements for the timeout time, it will be marked as temporarily idle. This allows downstream tasks to advance their watermarks without the need to wait for watermarks from this source while it is idle. Default value is 0 ms, which means detecting source idleness is not enabled.
sql.tables.scan.startup.mode	(None)	Enum	This specifies the startup mode for Kafka consumer for this specific statement. The valid enumerations are: group-offsets: start from comitted offsets of a specific consumer group. earliest-offset: start from the earliest offset possible. latest-offset: start from the latest offset. timestamp: start from user-supplied timestamp for each partition. If this config option is not set, the default is earliest-offset for this statement. This value can also be set on a table configuration instead of per statement. If timestamp is specified, config option sql.tables.scan.startup.timestamp-millis is required to specify a specific startup timestamp in milliseconds since January 1, 1970 00:00:00.000 GMT.
sql.tables.scan.startup.timestamp-millis	(None)	String	Start from the specified epoch timestamp (milliseconds) used in case of ‘timestamp’ startup mode.