ALTER Statements

ALTER statements are used to modified a registered table/view/function definition in the Catalog.

Flink SQL supports the following ALTER statements for now:

  • ALTER TABLE
  • ALTER VIEW
  • ALTER DATABASE
  • ALTER FUNCTION

Run an ALTER statement

Java

ALTER statements can be executed with the executeSql() method of the TableEnvironment. The executeSql() method returns ‘OK’ for a successful ALTER operation, otherwise will throw an exception.

The following examples show how to run an ALTER statement in TableEnvironment.

Scala

ALTER statements can be executed with the executeSql() method of the TableEnvironment. The executeSql() method returns ‘OK’ for a successful ALTER operation, otherwise will throw an exception.

The following examples show how to run an ALTER statement in TableEnvironment.

Python

ALTER statements can be executed with the execute_sql() method of the TableEnvironment. The execute_sql() method returns ‘OK’ for a successful ALTER operation, otherwise will throw an exception.

The following examples show how to run an ALTER statement in TableEnvironment.

SQL CLI

ALTER statements can be executed in SQL CLI.

The following examples show how to run an ALTER statement in SQL CLI.

Java

  1. EnvironmentSettings settings = EnvironmentSettings.newInstance()...
  2. TableEnvironment tableEnv = TableEnvironment.create(settings);
  4. // register a table named "Orders"
  5. tableEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)");
  7. // a string array: ["Orders"]
  8. String[] tables = tableEnv.listTables();
  9. // or tableEnv.executeSql("SHOW TABLES").print();
  11. // add a new column `order` to the first position
  12. tableEnv.executeSql("ALTER TABLE Orders ADD `order` INT COMMENT 'order identifier' FIRST");
  14. // add more columns, primary key and watermark
  15. tableEnv.executeSql("ALTER TABLE Orders ADD (ts TIMESTAMP(3), category STRING AFTER product, PRIMARY KEY(`order`) NOT ENFORCED, WATERMARK FOR ts AS ts - INTERVAL '1' HOUR)");
  17. // modify column type, column comment and watermark
  18. tableEnv.executeSql("ALTER TABLE Orders MODIFY (amount DOUBLE NOT NULL, category STRING COMMENT 'category identifier' AFTER `order`, WATERMARK FOR ts AS ts)");
  20. // drop watermark
  21. tableEnv.executeSql("ALTER TABLE Orders DROP WATERMARK");
  23. // drop column
  24. tableEnv.executeSql("ALTER TABLE Orders DROP (amount, ts, category)");
  26. // rename column
  27. tableEnv.executeSql("ALTER TABLE Orders RENAME `order` TO order_id");
  29. // rename "Orders" to "NewOrders"
  30. tableEnv.executeSql("ALTER TABLE Orders RENAME TO NewOrders");
  32. // a string array: ["NewOrders"]
  33. String[] tables = tableEnv.listTables();
  34. // or tableEnv.executeSql("SHOW TABLES").print();

Scala

  1. val tableEnv = TableEnvironment.create(...)
  3. // register a table named "Orders"
  4. tableEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)")
  6. // add a new column `order` to the first position
  7. tableEnv.executeSql("ALTER TABLE Orders ADD `order` INT COMMENT 'order identifier' FIRST")
  9. // add more columns, primary key and watermark
  10. tableEnv.executeSql("ALTER TABLE Orders ADD (ts TIMESTAMP(3), category STRING AFTER product, PRIMARY KEY(`order`) NOT ENFORCED, WATERMARK FOR ts AS ts - INTERVAL '1' HOUR)")
  12. // modify column type, column comment and watermark
  13. tableEnv.executeSql("ALTER TABLE Orders MODIFY (amount DOUBLE NOT NULL, category STRING COMMENT 'category identifier' AFTER `order`, WATERMARK FOR ts AS ts)")
  15. // drop watermark
  16. tableEnv.executeSql("ALTER TABLE Orders DROP WATERMARK")
  18. // drop column
  19. tableEnv.executeSql("ALTER TABLE Orders DROP (amount, ts, category)")
  21. // rename column
  22. tableEnv.executeSql("ALTER TABLE Orders RENAME `order` TO order_id")
  24. // a string array: ["Orders"]
  25. val tables = tableEnv.listTables()
  26. // or tableEnv.executeSql("SHOW TABLES").print()
  28. // rename "Orders" to "NewOrders"
  29. tableEnv.executeSql("ALTER TABLE Orders RENAME TO NewOrders")
  31. // a string array: ["NewOrders"]
  32. val tables = tableEnv.listTables()
  33. // or tableEnv.executeSql("SHOW TABLES").print()

Python

  1. table_env = TableEnvironment.create(...)
  3. # a string array: ["Orders"]
  4. tables = table_env.list_tables()
  5. # or table_env.execute_sql("SHOW TABLES").print()
  7. # add a new column `order` to the first position
  8. table_env.execute_sql("ALTER TABLE Orders ADD `order` INT COMMENT 'order identifier' FIRST");
  10. # add more columns, primary key and watermark
  11. table_env.execute_sql("ALTER TABLE Orders ADD (ts TIMESTAMP(3), category STRING AFTER product, PRIMARY KEY(`order`) NOT ENFORCED, WATERMARK FOR ts AS ts - INTERVAL '1' HOUR)");
  13. # modify column type, column comment and watermark
  14. table_env.execute_sql("ALTER TABLE Orders MODIFY (amount DOUBLE NOT NULL, category STRING COMMENT 'category identifier' AFTER `order`, WATERMARK FOR ts AS ts)");
  16. # drop watermark
  17. table_env.execute_sql("ALTER TABLE Orders DROP WATERMARK");
  19. # drop column
  20. table_env.execute_sql("ALTER TABLE Orders DROP (amount, ts, category)");
  22. # rename column
  23. table_env.execute_sql("ALTER TABLE Orders RENAME `order` TO order_id");
  25. # rename "Orders" to "NewOrders"
  26. table_env.execute_sql("ALTER TABLE Orders RENAME TO NewOrders");
  28. # a string array: ["NewOrders"]
  29. tables = table_env.list_tables()
  30. # or table_env.execute_sql("SHOW TABLES").print()

SQL CLI

  1. Flink SQL> CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...);
  2. [INFO] Execute statement succeed.
  4. Flink SQL> ALTER TABLE Orders ADD `order` INT COMMENT 'order identifier' FIRST;
  5. [INFO] Execute statement succeed.
  7. Flink SQL> DESCRIBE Orders;
  8. +---------+--------+------+-----+--------+-----------+------------------+
  9. |    name |   type | null | key | extras | watermark |          comment |
  10. +---------+--------+------+-----+--------+-----------+------------------+
  11. |   order |    INT | TRUE |     |        |           | order identifier |
  12. |    user | BIGINT | TRUE |     |        |           |                  |
  13. | product | STRING | TRUE |     |        |           |                  |
  14. |  amount |    INT | TRUE |     |        |           |                  |
  15. +---------+--------+------+-----+--------+-----------+------------------+
  16. 4 rows in set
  18. Flink SQL> ALTER TABLE Orders ADD (ts TIMESTAMP(3), category STRING AFTER product, PRIMARY KEY(`order`) NOT ENFORCED, WATERMARK FOR ts AS ts - INTERVAL '1' HOUR);
  19. [INFO] Execute statement succeed. 
  21. Flink SQL> DESCRIBE Orders;
  22. +----------+------------------------+-------+------------+--------+--------------------------+------------------+
  23. |     name |                   type |  null |        key | extras |                watermark |          comment |
  24. +----------+------------------------+-------+------------+--------+--------------------------+------------------+
  25. |    order |                    INT | FALSE | PRI(order) |        |                          | order identifier |
  26. |     user |                 BIGINT |  TRUE |            |        |                          |                  |
  27. |  product |                 STRING |  TRUE |            |        |                          |                  |
  28. | category |                 STRING |  TRUE |            |        |                          |                  |
  29. |   amount |                    INT |  TRUE |            |        |                          |                  |
  30. |       ts | TIMESTAMP(3) *ROWTIME* |  TRUE |            |        | `ts` - INTERVAL '1' HOUR |                  |
  31. +----------+------------------------+-------+------------+--------+--------------------------+------------------+
  32. 6 rows in set
  34. Flink SQL> ALTER TABLE Orders MODIFY (amount DOUBLE NOT NULL, category STRING COMMENT 'category identifier' AFTER `order`, WATERMARK FOR ts AS ts);
  35. [INFO] Execute statement succeed. 
  37. Flink SQL> DESCRIBE Orders;
  38. +----------+------------------------+-------+------------+--------+-----------+---------------------+
  39. |     name |                   type |  null |        key | extras | watermark |             comment |
  40. +----------+------------------------+-------+------------+--------+-----------+---------------------+
  41. |    order |                    INT | FALSE | PRI(order) |        |           |    order identifier |
  42. | category |                 STRING |  TRUE |            |        |           | category identifier |
  43. |     user |                 BIGINT |  TRUE |            |        |           |                     |
  44. |  product |                 STRING |  TRUE |            |        |           |                     |
  45. |   amount |                 DOUBLE | FALSE |            |        |           |                     |
  46. |       ts | TIMESTAMP(3) *ROWTIME* |  TRUE |            |        |      `ts` |                     |
  47. +----------+------------------------+-------+------------+--------+-----------+---------------------+
  48. 6 rows in set
  50. Flink SQL> ALTER TABLE Orders DROP WATERMARK;
  51. [INFO] Execute statement succeed.
  53. Flink SQL> DESCRIBE Orders;
  54. +----------+--------------+-------+------------+--------+-----------+---------------------+
  55. |     name |         type |  null |        key | extras | watermark |             comment |
  56. +----------+--------------+-------+------------+--------+-----------+---------------------+
  57. |    order |          INT | FALSE | PRI(order) |        |           |    order identifier |
  58. | category |       STRING |  TRUE |            |        |           | category identifier |
  59. |     user |       BIGINT |  TRUE |            |        |           |                     |
  60. |  product |       STRING |  TRUE |            |        |           |                     |
  61. |   amount |       DOUBLE | FALSE |            |        |           |                     |
  62. |       ts | TIMESTAMP(3) |  TRUE |            |        |           |                     |
  63. +----------+--------------+-------+------------+--------+-----------+---------------------+
  64. 6 rows in set
  66. Flink SQL> ALTER TABLE Orders DROP (amount, ts, category);
  67. [INFO] Execute statement succeed.
  69. Flink SQL> DESCRIBE Orders;
  70. +---------+--------+-------+------------+--------+-----------+------------------+
  71. |    name |   type |  null |        key | extras | watermark |          comment |
  72. +---------+--------+-------+------------+--------+-----------+------------------+
  73. |   order |    INT | FALSE | PRI(order) |        |           | order identifier |
  74. |    user | BIGINT |  TRUE |            |        |           |                  |
  75. | product | STRING |  TRUE |            |        |           |                  |
  76. +---------+--------+-------+------------+--------+-----------+------------------+
  77. 3 rows in set
  79. Flink SQL> ALTER TABLE Orders RENAME `order` to `order_id`;
  80. [INFO] Execute statement succeed.
  82. Flink SQL> DESCRIBE Orders;
  83. +----------+--------+-------+---------------+--------+-----------+------------------+
  84. |     name |   type |  null |           key | extras | watermark |          comment |
  85. +----------+--------+-------+---------------+--------+-----------+------------------+
  86. | order_id |    INT | FALSE | PRI(order_id) |        |           | order identifier |
  87. |     user | BIGINT |  TRUE |               |        |           |                  |
  88. |  product | STRING |  TRUE |               |        |           |                  |
  89. +----------+--------+-------+---------------+--------+-----------+------------------+
  90. 3 rows in set
  92. Flink SQL> SHOW TABLES;
  93. +------------+
  94. | table name |
  95. +------------+
  96. |     Orders |
  97. +------------+
  98. 1 row in set
  100. Flink SQL> ALTER TABLE Orders RENAME TO NewOrders;
  101. [INFO] Execute statement succeed.
  103. Flink SQL> SHOW TABLES;
  104. +------------+
  105. | table name |
  106. +------------+
  107. |  NewOrders |
  108. +------------+
  109. 1 row in set

ALTER TABLE

The following grammar gives an overview about the available syntax:

  1. ALTER TABLE [IF EXISTS] table_name {
  2.     ADD { <schema_component> | (<schema_component> [, ...]) | [IF NOT EXISTS] <partition_component> [<partition_component> ...]}
  3.   | MODIFY { <schema_component> | (<schema_component> [, ...]) }
  4.   | DROP {column_name | (column_name, column_name, ....) | PRIMARY KEY | CONSTRAINT constraint_name | WATERMARK | [IF EXISTS] <partition_component> [, ...]}
  5.   | RENAME old_column_name TO new_column_name
  6.   | RENAME TO new_table_name
  7.   | SET (key1=val1, ...)
  8.   | RESET (key1, ...)
  9. }
  11. <schema_component>:
  12.   { <column_component> | <constraint_component> | <watermark_component> }
  14. <column_component>:
  15.   column_name <column_definition> [FIRST | AFTER column_name]
  17. <constraint_component>:
  18.   [CONSTRAINT constraint_name] PRIMARY KEY (column_name, ...) NOT ENFORCED
  20. <watermark_component>:
  21.   WATERMARK FOR rowtime_column_name AS watermark_strategy_expression
  23. <column_definition>:
  24.   { <physical_column_definition> | <metadata_column_definition> | <computed_column_definition> } [COMMENT column_comment]
  26. <physical_column_definition>:
  27.   column_type
  29. <metadata_column_definition>:
  30.   column_type METADATA [ FROM metadata_key ] [ VIRTUAL ]
  32. <computed_column_definition>:
  33.   AS computed_column_expression
  35. <partition_component>:
  36.   PARTITION (key1=val1, key2=val2, ...) [WITH (key1=val1, key2=val2, ...)]

IF EXISTS

If the table does not exist, nothing happens.

ADD

Use ADD clause to add columns, constraints, watermark and partitions to an existing table.

To add a column at the specified position, use FIRST or AFTER col_name. By default, the column is appended at last.

The following examples illustrate the usage of the ADD statements.

  1. -- add a new column 
  2. ALTER TABLE MyTable ADD category_id STRING COMMENT 'identifier of the category';
  4. -- add columns, constraint, and watermark
  5. ALTER TABLE MyTable ADD (
  6.     log_ts STRING COMMENT 'log timestamp string' FIRST,
  7.     ts AS TO_TIMESTAMP(log_ts) AFTER log_ts,
  8.     PRIMARY KEY (id) NOT ENFORCED,
  9.     WATERMARK FOR ts AS ts - INTERVAL '3' SECOND
  10. );
  12. -- add a new partition 
  13. ALTER TABLE MyTable ADD PARTITION (p1=1,p2='a') with ('k1'='v1');
  15. -- add two new partitions
  16. ALTER TABLE MyTable ADD PARTITION (p1=1,p2='a') with ('k1'='v1') PARTITION (p1=1,p2='b') with ('k2'='v2');

Note Add a column to be primary key will change the column’s nullability to false implicitly.

MODIFY

Use MODIFY clause to change column’s position, type, comment or nullability, change primary key columns and watermark strategy to an existing table.

To modify an existent column to a new position, use FIRST or AFTER col_name. By default, the position remains unchanged.

The following examples illustrate the usage of the MODIFY statements.

  1. -- modify a column type, comment and position
  2. ALTER TABLE MyTable MODIFY measurement double COMMENT 'unit is bytes per second' AFTER `id`;
  4. -- modify definition of column log_ts and ts, primary key, watermark. They must exist in table schema
  5. ALTER TABLE MyTable MODIFY (
  6.     log_ts STRING COMMENT 'log timestamp string' AFTER `id`,  -- reorder columns
  7.     ts AS TO_TIMESTAMP(log_ts) AFTER log_ts,
  8.     PRIMARY KEY (id) NOT ENFORCED,
  9.     WATERMARK FOR ts AS ts -- modify watermark strategy
  10. );

Note Modify a column to be primary key will change the column’s nullability to false implicitly.

DROP

Use the DROP clause to drop columns, primary key, partitions, and watermark strategy to an existing table.

The following examples illustrate the usage of the DROP statements.

  1. -- drop a column
  2. ALTER TABLE MyTable DROP measurement;
  4. -- drop columns
  5. ALTER TABLE MyTable DROP (col1, col2, col3);
  7. -- drop primary key
  8. ALTER TABLE MyTable DROP PRIMARY KEY;
  10. -- drop a partition
  11. ALTER TABLE MyTable DROP PARTITION (`id` = 1);
  13. -- drop two partitions
  14. ALTER TABLE MyTable DROP PARTITION (`id` = 1), PARTITION (`id` = 2);
  16. -- drop a watermark
  17. ALTER TABLE MyTable DROP WATERMARK;

RENAME

Use RENAME clause to rename column or an existing table.

The following examples illustrate the usage of the RENAME statements.

  1. -- rename column
  2. ALTER TABLE MyTable RENAME request_body TO payload;
  4. -- rename table
  5. ALTER TABLE MyTable RENAME TO MyTable2;

SET

Set one or more properties in the specified table. If a particular property is already set in the table, override the old value with the new one.

The following examples illustrate the usage of the SET statements.

  1. -- set 'rows-per-second'
  2. ALTER TABLE DataGenSource SET ('rows-per-second' = '10');

RESET

Reset one or more properties to its default value.

The following examples illustrate the usage of the RESET statements.

  1. -- reset 'rows-per-second' to the default value
  2. ALTER TABLE DataGenSource RESET ('rows-per-second');

ALTER VIEW

  1. ALTER VIEW [catalog_name.][db_name.]view_name RENAME TO new_view_name

Renames a given view to a new name within the same catalog and database.

  1. ALTER VIEW [catalog_name.][db_name.]view_name AS new_query_expression

Changes the underlying query defining the given view to a new query.

ALTER DATABASE

  1. ALTER DATABASE [catalog_name.]db_name SET (key1=val1, key2=val2, ...)

Set one or more properties in the specified database. If a particular property is already set in the database, override the old value with the new one.

ALTER FUNCTION

  1. ALTER [TEMPORARY|TEMPORARY SYSTEM] FUNCTION 
  2.   [IF EXISTS] [catalog_name.][db_name.]function_name 
  3.   AS identifier [LANGUAGE JAVA|SCALA|PYTHON]

Alter a catalog function with the new identifier and optional language tag. If a function doesn’t exist in the catalog, an exception is thrown.

If the language tag is JAVA/SCALA, the identifier is the full classpath of the UDF. For the implementation of Java/Scala UDF, please refer to User-defined Functions for more details.

If the language tag is PYTHON, the identifier is the fully qualified name of the UDF, e.g. pyflink.table.tests.test_udf.add. For the implementation of Python UDF, please refer to Python UDFs for more details.

TEMPORARY

Alter temporary catalog function that has catalog and database namespaces and overrides catalog functions.

TEMPORARY SYSTEM

Alter temporary system function that has no namespace and overrides built-in functions

IF EXISTS

If the function doesn’t exist, nothing happens.

LANGUAGE JAVA|SCALA|PYTHON

Language tag to instruct flink runtime how to execute the function. Currently only JAVA, SCALA and PYTHON are supported, the default language for a function is JAVA.