ALTER 语句

ALTER 语句用于修改一个已经在 Catalog 中注册的表、视图或函数定义。

Flink SQL 目前支持以下 ALTER 语句：

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

执行 ALTER 语句

Java

可以使用 TableEnvironment 中的 executeSql() 方法执行 ALTER 语句。 若 ALTER 操作执行成功，executeSql() 方法返回 ‘OK’，否则会抛出异常。

以下的例子展示了如何在 TableEnvironment 中执行一个 ALTER 语句。

Scala

可以使用 TableEnvironment 中的 executeSql() 方法执行 ALTER 语句。 若 ALTER 操作执行成功，executeSql() 方法返回 ‘OK’，否则会抛出异常。

以下的例子展示了如何在 TableEnvironment 中执行一个 ALTER 语句。

Python

可以使用 TableEnvironment 中的 execute_sql() 方法执行 ALTER 语句。 若 ALTER 操作执行成功，execute_sql() 方法返回 ‘OK’，否则会抛出异常。

以下的例子展示了如何在 TableEnvironment 中执行一个 ALTER 语句。

SQL CLI

可以在 SQL CLI 中执行 ALTER 语句。

以下的例子展示了如何在 SQL CLI 中执行一个 ALTER 语句。

Java

  1. TableEnvironment tableEnv = TableEnvironment.create(...);
  3. // 注册名为 “Orders” 的表
  4. tableEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)");
  6. // 字符串数组： ["Orders"]
  7. String[] tables = tableEnv.listTables();
  8. // or tableEnv.executeSql("SHOW TABLES").print();
  10. // 新增列 `order` 并置于第一位
  11. tableEnv.executeSql("ALTER TABLE Orders ADD `order` INT COMMENT 'order identifier' FIRST");
  13. // 新增更多列, 以及主键和 watermark
  14. 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)");
  16. // 修改列类型, 注释及 watermark 策略
  17. tableEnv.executeSql("ALTER TABLE Orders MODIFY (amount DOUBLE NOT NULL, category STRING COMMENT 'category identifier' AFTER `order`, WATERMARK FOR ts AS ts)");
  19. // 删除 watermark
  20. tableEnv.executeSql("ALTER TABLE Orders DROP WATERMARK");
  22. // 删除列
  23. tableEnv.executeSql("ALTER TABLE Orders DROP (amount, ts, category)");
  25. // 重命名列
  26. tableEnv.executeSql("ALTER TABLE Orders RENAME `order` TO order_id");
  28. // "Orders" 的表名改为 "NewOrders"
  29. tableEnv.executeSql("ALTER TABLE Orders RENAME TO NewOrders");
  31. // 字符串数组：["NewOrders"]
  32. String[] tables = tableEnv.listTables();
  33. // or tableEnv.executeSql("SHOW TABLES").print();

Scala

  1. val tableEnv = TableEnvironment.create(...)
  3. // 注册名为 “Orders” 的表
  4. tableEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)")
  6. // 新增列 `order` 并置于第一位
  7. tableEnv.executeSql("ALTER TABLE Orders ADD `order` INT COMMENT 'order identifier' FIRST")
  9. // 新增更多列, 以及主键和 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. // 修改列类型, 注释, 以及主键和 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. // 删除 watermark
  16. tableEnv.executeSql("ALTER TABLE Orders DROP WATERMARK")
  18. // 删除列
  19. tableEnv.executeSql("ALTER TABLE Orders DROP (amount, ts, category)")
  21. // 重命名列
  22. tableEnv.executeSql("ALTER TABLE Orders RENAME `order` TO order_id")
  24. // 字符串数组： ["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. // 字符串数组：["NewOrders"]
  32. val tables = tableEnv.listTables()
  33. // or tableEnv.executeSql("SHOW TABLES").print()

Python

  1. table_env = TableEnvironment.create(...)
  3. # 字符串数组： ["Orders"]
  4. tables = table_env.list_tables()
  5. # or table_env.execute_sql("SHOW TABLES").print()
  7. # 新增列 `order` 并置于第一位
  8. table_env.execute_sql("ALTER TABLE Orders ADD `order` INT COMMENT 'order identifier' FIRST");
  10. # 新增更多列, 主键及 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. # 修改列类型, 列注释, 主键及 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. # 删除 watermark
  17. table_env.execute_sql("ALTER TABLE Orders DROP WATERMARK");
  19. # 删除列
  20. table_env.execute_sql("ALTER TABLE Orders DROP (amount, ts, category)");
  22. # 重命名列
  23. table_env.execute_sql("ALTER TABLE Orders RENAME `order` TO order_id");
  25. # 把 "Orders" 的表名改为 "NewOrders"
  26. table_env.execute_sql("ALTER TABLE Orders RENAME TO NewOrders");
  28. # 字符串数组：["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

当前支持的 ALTER TABLE 语法如下

  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

若表不存在，则不进行任何操作。

ADD

使用 ADD 语句向已有表中增加 columns， constraints，watermark, partitions。

向表新增列时可通过 FIRST or AFTER col_name 指定位置，不指定位置时默认追加在最后。

ADD 语句示例如下。

  1. -- 新增一列
  2. ALTER TABLE MyTable ADD category_id STRING COMMENT 'identifier of the category';
  4. -- 新增列，主键和 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. -- 新增一个分区 
  13. ALTER TABLE MyTable ADD PARTITION (p1=1,p2='a') with ('k1'='v1');
  15. -- 新增两个分区
  16. ALTER TABLE MyTable ADD PARTITION (p1=1,p2='a') with ('k1'='v1') PARTITION (p1=1,p2='b') with ('k2'='v2');

注意 指定列为主键列时会隐式修改该列的 nullability 为 false。

MODIFY

使用 MODIFY 语句修改列的位置 、类型 、注释 、nullability，主键或 watermark。

可使用 FIRST 或 AFTER col_name 将已有列移动至指定位置，不指定时默认保持位置不变。

MODIFY 语句示例如下。

  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. );

注意 指定列为主键列时会隐式修改该列的 nullability 为 false。

DROP

使用 DROP 语句删除列 、主键 、 分区或 watermark。

DROP 语句示例如下。

  1. -- 删除一个列
  2. ALTER TABLE MyTable DROP measurement;
  4. -- 删除多个列
  5. ALTER TABLE MyTable DROP (col1, col2, col3);
  7. -- 删除主键
  8. ALTER TABLE MyTable DROP PRIMARY KEY;
  10. -- 删除一个分区
  11. ALTER TABLE MyTable DROP PARTITION (`id` = 1);
  13. -- 删除两个分区
  14. ALTER TABLE MyTable DROP PARTITION (`id` = 1), PARTITION (`id` = 2);
  16. -- 删除 watermark
  17. ALTER TABLE MyTable DROP WATERMARK;

RENAME

使用 RENAME 语句修改列名或表名。

RENAME 语句示例如下。

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

SET

为指定的表设置一个或多个属性。若个别属性已经存在于表中，则使用新值覆盖旧值。

SET 语句示例如下。

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

RESET

为指定的表重置一个或多个属性。

RESET 语句示例如下。

  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, ...)

在数据库中设置一个或多个属性。若个别属性已经在数据库中设定，将会使用新值覆盖旧值。

ALTER FUNCTION

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

修改一个有 catalog 和数据库命名空间的 catalog function ，需要指定一个新的 identifier ，可指定 language tag 。若函数不存在，删除会抛出异常。

如果 language tag 是 JAVA 或者 SCALA ，则 identifier 是 UDF 实现类的全限定名。关于 JAVA/SCALA UDF 的实现，请参考 自定义函数。

如果 language tag 是 PYTHON ， 则 identifier 是 UDF 对象的全限定名，例如 pyflink.table.tests.test_udf.add。关于 PYTHON UDF 的实现，请参考 Python UDFs。

TEMPORARY

修改一个有 catalog 和数据库命名空间的临时 catalog function ，并覆盖原有的 catalog function 。

TEMPORARY SYSTEM

修改一个没有数据库命名空间的临时系统 catalog function ，并覆盖系统内置的函数。

IF EXISTS

若函数不存在，则不进行任何操作。

LANGUAGE JAVA|SCALA|PYTHON

Language tag 用于指定 Flink runtime 如何执行这个函数。目前，只支持 JAVA，SCALA 和 PYTHON，且函数的默认语言为 JAVA。