Flink Procedures

Fluss provides stored procedures to perform administrative and management operations through Flink SQL. All procedures are located in the sys namespace and can be invoked using the CALL statement.

Available Procedures

You can list all available procedures using:

Flink SQL

SHOW PROCEDURES;

Access Control Procedures

Fluss provides procedures to manage Access Control Lists (ACLs) for security and authorization. See the Security documentation for more details.

add_acl

Add an ACL entry to grant permissions to a principal.

Syntax:

CALL [catalog_name.]sys.add_acl(
  resource => 'STRING',
  permission => 'STRING', 
  principal => 'STRING',
  operation => 'STRING',
  host => 'STRING'  -- optional, defaults to '*'
)

Parameters:

• resource (required): The resource to grant permissions on. Can be 'CLUSTER' for cluster-level permissions or a specific resource name (e.g., database or table name).
• permission (required): The permission type to grant. Valid values are 'ALLOW' or 'DENY'.
• principal (required): The principal to grant permissions to, in the format 'Type:Name' (e.g., 'User:Alice').
• operation (required): The operation type to grant. Valid values include 'READ', 'WRITE', 'CREATE', 'DELETE', 'ALTER', 'DESCRIBE', 'CLUSTER_ACTION', 'IDEMPOTENT_WRITE'.
• host (optional): The host from which the principal can access the resource. Defaults to '*' (all hosts).

Example:

Flink SQL

-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
USE fluss_catalog;

-- Grant read permission to user Alice from any host
CALL sys.add_acl(
  resource => 'CLUSTER',
  permission => 'ALLOW',
  principal => 'User:Alice',
  operation => 'READ',
  host => '*'
);

-- Grant write permission to user Bob from a specific host
CALL sys.add_acl(
  resource => 'my_database.my_table',
  permission => 'ALLOW',
  principal => 'User:Bob',
  operation => 'WRITE',
  host => '192.168.1.100'
);

drop_acl

Remove an ACL entry to revoke permissions.

Syntax:

CALL [catalog_name.]sys.drop_acl(
  resource => 'STRING',
  permission => 'STRING',
  principal => 'STRING', 
  operation => 'STRING',
  host => 'STRING'  -- optional, defaults to '*'
)

Parameters:

All parameters accept the same values as add_acl. You can use 'ANY' as a wildcard value to match multiple entries for batch deletion.

Example:

Flink SQL

-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
USE fluss_catalog;

-- Remove a specific ACL entry
CALL sys.drop_acl(
  resource => 'CLUSTER',
  permission => 'ALLOW',
  principal => 'User:Alice',
  operation => 'READ',
  host => '*'
);

-- Remove all ACL entries for a specific user
CALL sys.drop_acl(
  resource => 'ANY',
  permission => 'ANY',
  principal => 'User:Alice',
  operation => 'ANY',
  host => 'ANY'
);

list_acl

List ACL entries matching the specified filters.

Syntax:

CALL [catalog_name.]sys.list_acl(
  resource => 'STRING',
  permission => 'STRING',  -- optional, defaults to 'ANY'
  principal => 'STRING',   -- optional, defaults to 'ANY'
  operation => 'STRING',   -- optional, defaults to 'ANY'
  host => 'STRING'         -- optional, defaults to 'ANY'
)

Parameters:

All parameters accept the same values as add_acl. Use 'ANY' as a wildcard to match all values for that parameter.

Returns: An array of strings, each representing an ACL entry in the format: resource="...";permission="...";principal="...";operation="...";host="..."

Example:

Flink SQL

-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
USE fluss_catalog;

-- List all ACL entries
CALL sys.list_acl(resource => 'ANY');

-- List all ACL entries for a specific user
CALL sys.list_acl(
  resource => 'ANY',
  principal => 'User:Alice'
);

-- List all read permissions
CALL sys.list_acl(
  resource => 'ANY',
  operation => 'READ'
);

Cluster Configuration Procedures

Fluss provides procedures to dynamically manage cluster configurations without requiring a server restart.

get_cluster_configs

Retrieve cluster configuration values.

Syntax:

-- Get multiple configurations
CALL [catalog_name.]sys.get_cluster_configs(config_keys => 'key1' [, 'key2', ...])

-- Get all cluster configurations
CALL [catalog_name.]sys.get_cluster_configs()

Parameters:

• config_keys (optional): The configuration keys to retrieve. If omitted, returns all cluster configurations.

Returns: A table with columns:

• config_key: The configuration key name
• config_value: The current value
• config_source: The source of the configuration (e.g., DYNAMIC_CONFIG, STATIC_CONFIG)

Example:

Flink SQL

-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
USE fluss_catalog;

-- Get a specific configuration
CALL sys.get_cluster_configs(
  config_keys => 'kv.rocksdb.shared-rate-limiter.bytes-per-sec'
);

-- Get multiple configuration
CALL sys.get_cluster_configs(
  config_keys => 'kv.rocksdb.shared-rate-limiter.bytes-per-sec', 'datalake.format'
);

-- Get all cluster configurations
CALL sys.get_cluster_configs();

set_cluster_configs

Set cluster configurations dynamically.

Syntax:

-- Set configuration values
CALL [catalog_name.]sys.set_cluster_configs(
  config_pairs => 'key1', 'value1' [, 'key2', 'value2' ...]
)

Parameters:

• config_pairs(required): For key-value pairs in configuration items, the number of parameters must be even.

Important Notes:

• Changes are validated before being applied and persisted in ZooKeeper
• Changes are automatically applied to all servers (Coordinator and TabletServers)
• Changes survive server restarts
• Not all configurations support dynamic changes. The server will reject invalid modifications

Example:

Flink SQL

-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
USE fluss_catalog;

-- Set RocksDB rate limiter
CALL sys.set_cluster_configs(
  config_pairs => 'kv.rocksdb.shared-rate-limiter.bytes-per-sec', '200MB'
);

-- Set RocksDB rate limiter and datalake format
CALL sys.set_cluster_configs(
  config_pairs => 'kv.rocksdb.shared-rate-limiter.bytes-per-sec', '200MB', 'datalake.format','paimon'
);

append_cluster_configs

Append values to list-type or map-type cluster configurations dynamically.

Syntax:

-- Append collection configuration values
CALL [catalog_name.]sys.append_cluster_configs(
  config_pairs => 'key1', 'value1' [, 'key2', 'value2' ...]
)

Parameters:

• config_pairs(required): For key-value pairs in configuration items, the number of parameters must be even.

Important Notes:

• APPEND operations are only supported for list-type or map-type configuration keys
• For map-type configurations, values must use the map-key:map-value format
• Appending a map entry whose key already exists is rejected

Example:

Flink SQL

-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
USE fluss_catalog;

-- Add a SASL Plain credential
CALL sys.append_cluster_configs(
  config_pairs => 'security.sasl.plain.credentials', 'bob:bob-secret'
);

subtract_cluster_configs

Subtract values from list-type or map-type cluster configurations dynamically.

Syntax:

-- Subtract collection configuration values
CALL [catalog_name.]sys.subtract_cluster_configs(
  config_pairs => 'key1', 'value1' [, 'key2', 'value2' ...]
)

Parameters:

• config_pairs(required): For key-value pairs in configuration items, the number of parameters must be even.

Important Notes:

• SUBTRACT operations are only supported for list-type or map-type configuration keys
• For list-type configurations, SUBTRACT removes the exact list value
• For map-type configurations, values must use the map-key:map-value format, but removal is matched only by map-key; the supplied map-value is ignored for matching
• Subtracting a list value or map key that does not exist is a no-op

Example:

Flink SQL

-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
USE fluss_catalog;

-- Remove user "bob" from the SASL Plain credentials map.
-- This removes the "bob" entry even if the stored secret is different.
CALL sys.subtract_cluster_configs(
  config_pairs => 'security.sasl.plain.credentials', 'bob:any-secret'
);

reset_cluster_configs

reset cluster configurations dynamically.

Syntax:

-- reset configuration values
CALL [catalog_name.]sys.reset_cluster_configs(config_keys => 'key1' [, 'key2', ...])

Parameters:

• config_keys(required): The configuration keys to reset.

Example:

Flink SQL

-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
USE fluss_catalog;

-- Reset a specific configuration
CALL sys.reset_cluster_configs(
  config_keys => 'kv.rocksdb.shared-rate-limiter.bytes-per-sec'
);

-- Reset RocksDB rate limiter and datalake format
CALL sys.reset_cluster_configs(
  config_keys => 'kv.rocksdb.shared-rate-limiter.bytes-per-sec', 'datalake.format'
);

Rebalance Procedures

Fluss provides procedures to rebalance buckets across the cluster based on workload. Rebalancing primarily occurs in the following scenarios: Offline existing tabletServers from the cluster, adding new tabletServers to the cluster, and routine adjustments for load imbalance.

add_server_tag

Add server tag to TabletServers in the cluster. For example, adding tabletServer-0 with PERMANENT_OFFLINE tag indicates that tabletServer-0 is about to be permanently decommissioned, and during the next rebalance, all buckets on this node need to be migrated away.

Syntax:

CALL [catalog_name.]sys.add_server_tag(
  tabletServers => 'STRING',
  serverTag => 'STRING'
)

Parameters:

• tabletServers (required): The TabletServer IDs to add tag to. Can be a single server ID (e.g., '0') or multiple IDs separated by commas (e.g., '0,1,2').
• serverTag (required): The tag to add to the TabletServers. Valid values are:
  • 'PERMANENT_OFFLINE': Indicates the TabletServer is permanently offline and will be decommissioned. All buckets on this server will be migrated during the next rebalance.
  • 'TEMPORARY_OFFLINE': Indicates the TabletServer is temporarily offline (e.g., for upgrading). Buckets may be temporarily migrated but can return after the server comes back online.

Returns: An array with a single element 'success' if the operation completes successfully.

Example:

Flink SQL

-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
USE fluss_catalog;

-- Add PERMANENT_OFFLINE tag to a single TabletServer
CALL sys.add_server_tag('0', 'PERMANENT_OFFLINE');

-- Add TEMPORARY_OFFLINE tag to multiple TabletServers
CALL sys.add_server_tag('1,2,3', 'TEMPORARY_OFFLINE');

remove_server_tag

Remove server tag from TabletServers in the cluster. This operation is typically used when a previously tagged TabletServer is ready to return to normal service, or to cancel a planned offline operation.

Syntax:

CALL [catalog_name.]sys.remove_server_tag(
  tabletServers => 'STRING',
  serverTag => 'STRING'
)

Parameters:

• tabletServers (required): The TabletServer IDs to remove tag from. Can be a single server ID (e.g., '0') or multiple IDs separated by commas (e.g., '0,1,2').
• serverTag (required): The tag to remove from the TabletServers. Valid values are:
  • 'PERMANENT_OFFLINE': Remove the permanent offline tag from the TabletServer.
  • 'TEMPORARY_OFFLINE': Remove the temporary offline tag from the TabletServer.

Returns: An array with a single element 'success' if the operation completes successfully.

Example:

Flink SQL

-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
USE fluss_catalog;

-- Remove PERMANENT_OFFLINE tag from a single TabletServer
CALL sys.remove_server_tag('0', 'PERMANENT_OFFLINE');

-- Remove TEMPORARY_OFFLINE tag from multiple TabletServers
CALL sys.remove_server_tag('1,2,3', 'TEMPORARY_OFFLINE');

add_server_tag_by_rack

Add a server tag to TabletServers that belong to the specified racks when the Coordinator processes the request. This is a convenience operation that resolves the current rack membership and tags the matched TabletServer IDs in one operation.

Syntax:

CALL [catalog_name.]sys.add_server_tag_by_rack(
  racks => 'STRING',
  serverTag => 'STRING'
)

Parameters:

• racks (required): The rack identifiers to target. Can be a single rack (e.g., 'rack-0') or multiple racks separated by commas (e.g., 'rack-0,rack-1').
• serverTag (required): The tag to add to the matched TabletServers. Valid values are:
  • 'PERMANENT_OFFLINE': Indicates the TabletServer is permanently offline and will be decommissioned. All buckets on this server will be migrated during the next rebalance.
  • 'TEMPORARY_OFFLINE': Indicates the TabletServer is temporarily offline (e.g., for upgrading). Buckets may be temporarily migrated but can return after the server comes back online.

Returns: An array with a single element 'success' if the operation completes successfully.

Important Notes:

• Only TabletServers registered in a requested rack when the Coordinator processes the operation are tagged. If no TabletServer matches, the operation succeeds without making changes.
• The operation does not create a persistent rack policy. TabletServers that register in the rack later do not inherit the tag.
• The caller must have ALTER permission on the cluster even when no TabletServer matches.
• If any matched TabletServer already has a different tag, the operation fails and none of the tags take effect.
• Before decommissioning a physical rack, cordon it in the external scheduler so that no new TabletServer can be scheduled there. This procedure is not a hard fence and does not move existing replicas. After adding PERMANENT_OFFLINE, trigger a rebalance and verify that the rack has no remaining assignments before stopping its machines.

Example:

Flink SQL

-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
USE fluss_catalog;

-- Add PERMANENT_OFFLINE tag to all TabletServers in rack-0
CALL sys.add_server_tag_by_rack('rack-0', 'PERMANENT_OFFLINE');

-- Add TEMPORARY_OFFLINE tag to all TabletServers in rack-0 and rack-1
CALL sys.add_server_tag_by_rack('rack-0,rack-1', 'TEMPORARY_OFFLINE');

remove_server_tag_by_rack

Remove a server tag from TabletServers that belong to the specified racks when the Coordinator processes the request.

Syntax:

CALL [catalog_name.]sys.remove_server_tag_by_rack(
  racks => 'STRING',
  serverTag => 'STRING'
)

Parameters:

• racks (required): The rack identifiers to target. Can be a single rack (e.g., 'rack-0') or multiple racks separated by commas (e.g., 'rack-0,rack-1').
• serverTag (required): The tag to remove from the matched TabletServers. Valid values are:
  • 'PERMANENT_OFFLINE': Remove the permanent offline tag from the TabletServer.
  • 'TEMPORARY_OFFLINE': Remove the temporary offline tag from the TabletServer.

Returns: An array with a single element 'success' if the operation completes successfully.

Important Notes:

• Only TabletServers registered in a requested rack when the Coordinator processes the operation are considered. If no TabletServer matches, the operation succeeds without making changes.
• The caller must have ALTER permission on the cluster even when no TabletServer matches.
• If any matched TabletServer has a different tag, the operation fails and none of the tags are removed.
• A tagged TabletServer that is offline or has moved to another rack is not selected by its former rack. Remove its tag explicitly with remove_server_tag and its TabletServer ID.

Example:

Flink SQL

-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
USE fluss_catalog;

-- Remove PERMANENT_OFFLINE tag from all TabletServers in rack-0
CALL sys.remove_server_tag_by_rack('rack-0', 'PERMANENT_OFFLINE');

-- Remove TEMPORARY_OFFLINE tag from all TabletServers in rack-0 and rack-1
CALL sys.remove_server_tag_by_rack('rack-0,rack-1', 'TEMPORARY_OFFLINE');

rebalance

Trigger a rebalance operation to redistribute buckets across TabletServers in the cluster. This procedure helps balance workload based on specified goals, such as distributing replicas or leaders evenly across the cluster.

Syntax:

CALL [catalog_name.]sys.rebalance(
  priorityGoals => 'STRING'
)

Parameters:

• priorityGoals (required): The rebalance goals to achieve, specified as goal types. Can be a single goal (e.g., 'REPLICA_DISTRIBUTION') or multiple goals separated by commas (e.g., 'RACK_AWARE,REPLICA_DISTRIBUTION,LEADER_DISTRIBUTION'). Valid goal types are:
  • 'RACK_AWARE': Ensures replicas of the same bucket are distributed across different racks. This goal should be placed first when rack awareness is required to prevent non-rack-balanced assignments.
  • 'REPLICA_DISTRIBUTION': Generates replica movement tasks to ensure the number of replicas on each TabletServer is near balanced.
  • 'LEADER_DISTRIBUTION': Generates leadership movement and leader replica movement tasks to ensure the number of leader replicas on each TabletServer is near balanced.

Returns: An array with a single element containing the rebalance ID (e.g., 'rebalance-12345'), which can be used to track or cancel the rebalance operation.

Important Notes:

• Multiple goals can be specified in priority order. The system will attempt to achieve goals in the order specified.
• When rack awareness is required, place RACK_AWARE as the first goal to ensure subsequent goals respect rack constraints.
• Rebalance operations run asynchronously in the background. Use the returned rebalance ID to monitor progress.
• The rebalance operation respects server tags set by add_server_tag and add_server_tag_by_rack. For example, servers marked with PERMANENT_OFFLINE will have their buckets migrated away.

Example:

Flink SQL

-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
USE fluss_catalog;

-- Trigger rebalance with rack-aware goal (recommended for multi-rack deployments)
CALL sys.rebalance('RACK_AWARE,REPLICA_DISTRIBUTION');

-- Trigger rebalance with replica distribution goal only
CALL sys.rebalance('REPLICA_DISTRIBUTION');

-- Trigger rebalance with multiple goals in priority order
CALL sys.rebalance('RACK_AWARE,REPLICA_DISTRIBUTION,LEADER_DISTRIBUTION');

list_rebalance

Query the progress and status of a rebalance operation. This procedure allows you to monitor ongoing or completed rebalance operations to track their progress and view detailed information about bucket movements.

Syntax:

-- List the most recent rebalance progress
CALL [catalog_name.]sys.list_rebalance()

-- List a specific rebalance progress by ID
CALL [catalog_name.]sys.list_rebalance(
  rebalanceId => 'STRING'
)

Parameters:

• rebalanceId (optional): The rebalance ID to query. If omitted, returns the progress of the most recent rebalance operation. The rebalance ID is returned when calling the rebalance procedure.

Returns: An array of strings containing:

• Rebalance ID: The unique identifier of the rebalance operation
• Rebalance total status: The overall status of the rebalance. Possible values are:
  • NOT_STARTED: The rebalance has been created but not yet started
  • REBALANCING: The rebalance is currently in progress
  • COMPLETED: The rebalance has successfully completed
  • FAILED: The rebalance has failed
  • CANCELED: The rebalance has been canceled
• Rebalance progress: The completion percentage (e.g., 75.5%)
• Rebalance detail progress for bucket: Detailed progress information for each bucket being moved

If no rebalance is found, returns empty line.

Example:

Flink SQL

-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
USE fluss_catalog;

-- List the most recent rebalance progress
CALL sys.list_rebalance();

-- List a specific rebalance progress by ID
CALL sys.list_rebalance('rebalance-12345');

cancel_rebalance

Cancel an ongoing rebalance operation. This procedure allows you to stop a rebalance that is in progress, which is useful when you need to halt bucket redistribution due to operational requirements or unexpected issues.

Syntax:

-- Cancel the most recent rebalance operation
CALL [catalog_name.]sys.cancel_rebalance()

-- Cancel a specific rebalance operation by ID
CALL [catalog_name.]sys.cancel_rebalance(
  rebalanceId => 'STRING'
)

Parameters:

• rebalanceId (optional): The rebalance ID to cancel. If omitted, cancels the most recent rebalance operation. The rebalance ID is returned when calling the rebalance procedure.

Returns: An array with a single element 'success' if the operation completes successfully.

Important Notes:

• Only rebalance operations in NOT_STARTED or REBALANCING status can be canceled.
• Canceling a rebalance will stop bucket movements, but already completed bucket migrations will not be rolled back.
• After cancellation, the rebalance status will change to CANCELED.
• You can verify the cancellation by calling list_rebalance to check the status.

Example:

Flink SQL

-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
USE fluss_catalog;

-- Cancel the most recent rebalance operation
CALL sys.cancel_rebalance();

-- Cancel a specific rebalance operation by ID
CALL sys.cancel_rebalance('rebalance-12345');

Partition Procedures

Fluss provides procedures to inspect partition-level metadata of partitioned tables.

list_partition_infos

List the partition id, partition name and bucket count of every partition of a partitioned table.

For a partitioned table, ALTER TABLE ... SET ('bucket.num' = N) only changes the bucket count for newly created partitions; existing partitions keep their original bucket counts. The table-level 'bucket.num' shown by SHOW CREATE TABLE is only the default for new partitions, so use this procedure to see the actual bucket count of each partition.

Syntax:

-- List the partition id, partition name and bucket count of each partition
CALL [catalog_name.]sys.list_partition_infos(
  db => 'STRING',
  table_name => 'STRING'
)

Parameters:

• db (required): The database name of the table.
• table_name (required): The table name.

Returns: One row per partition, containing:

• partition_id: The unique identifier of the partition in the cluster.
• partition_name: The partition name, e.g. 2024-01 for a daily partition.
• bucket_count: The bucket count of the partition.

Example:

Flink SQL

-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
USE fluss_catalog;

-- List the partitions of my_db.orders with their bucket counts
CALL sys.list_partition_infos('my_db', 'orders');

An example result after ALTER TABLE my_db.orders SET ('bucket.num' = '8') on a table created with 4 buckets:

+I[1001, 2024-01, 4]
+I[1002, 2024-02, 4]
+I[1003, 2024-03, 8]

kv snapshot lease

Fluss provides procedures to manage KV snapshot leases, allowing you to drop leased kv snapshots.

drop_kv_snapshot_lease

Drop KV snapshots leased under a specified leaseId. This is typically used for handle the scenario of lease remnants. After an abnormal job termination (e.g., crash or forced cancellation), the registered lease may not be released automatically and could require manual cleanup.

Syntax:

CALL [catalog_name.]sys.drop_kv_snapshot_lease(
  leaseId => 'STRING'
)

Parameters:

• leaseId (required): The lease identifier of the KV snapshots to release. This should match the lease ID used when acquiring the KV snapshots.

Returns: An array with a single element 'success' if the operation completes successfully.

Example:

Flink SQL

-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
USE fluss_catalog;

-- Drop KV snapshots leased under the given leaseId
CALL sys.drop_kv_snapshot_lease('test-lease-id');