ALTER 语句 #
ALTER 语句用于修改一个已经在 Catalog 中注册的表、视图或函数定义。
Flink SQL 目前支持以下 ALTER 语句:
- ALTER TABLE
- ALTER VIEW
- ALTER DATABASE
- ALTER FUNCTION
执行 ALTER 语句 #
可以使用 TableEnvironment
中的 executeSql()
方法执行 ALTER 语句。 若 ALTER 操作执行成功,executeSql()
方法返回 ‘OK’,否则会抛出异常。
以下的例子展示了如何在 TableEnvironment
中执行一个 ALTER 语句。
可以使用 TableEnvironment
中的 executeSql()
方法执行 ALTER 语句。 若 ALTER 操作执行成功,executeSql()
方法返回 ‘OK’,否则会抛出异常。
以下的例子展示了如何在 TableEnvironment
中执行一个 ALTER 语句。
可以使用 TableEnvironment
中的 execute_sql()
方法执行 ALTER 语句。 若 ALTER 操作执行成功,execute_sql()
方法返回 ‘OK’,否则会抛出异常。
以下的例子展示了如何在 TableEnvironment
中执行一个 ALTER 语句。
可以在 SQL CLI 中执行 ALTER 语句。
以下的例子展示了如何在 SQL CLI 中执行一个 ALTER 语句。
TableEnvironment tableEnv = TableEnvironment.create(...);
// 注册名为 “Orders” 的表
tableEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)");
// 字符串数组: ["Orders"]
String[] tables = tableEnv.listTables();
// or tableEnv.executeSql("SHOW TABLES").print();
// 新增列 `order` 并置于第一位
tableEnv.executeSql("ALTER TABLE Orders ADD `order` INT COMMENT 'order identifier' FIRST");
// 新增更多列, 以及主键和 watermark
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)");
// 修改列类型, 注释及 watermark 策略
tableEnv.executeSql("ALTER TABLE Orders MODIFY (amount DOUBLE NOT NULL, category STRING COMMENT 'category identifier' AFTER `order`, WATERMARK FOR ts AS ts)");
// 删除 watermark
tableEnv.executeSql("ALTER TABLE Orders DROP WATERMARK");
// 删除列
tableEnv.executeSql("ALTER TABLE Orders DROP (amount, ts, category)");
// 重命名列
tableEnv.executeSql("ALTER TABLE Orders RENAME `order` TO order_id");
// "Orders" 的表名改为 "NewOrders"
tableEnv.executeSql("ALTER TABLE Orders RENAME TO NewOrders");
// 字符串数组:["NewOrders"]
String[] tables = tableEnv.listTables();
// or tableEnv.executeSql("SHOW TABLES").print();
val tableEnv = TableEnvironment.create(...)
// 注册名为 “Orders” 的表
tableEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)")
// 新增列 `order` 并置于第一位
tableEnv.executeSql("ALTER TABLE Orders ADD `order` INT COMMENT 'order identifier' FIRST")
// 新增更多列, 以及主键和 watermark
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)")
// 修改列类型, 注释, 以及主键和 watermark
tableEnv.executeSql("ALTER TABLE Orders MODIFY (amount DOUBLE NOT NULL, category STRING COMMENT 'category identifier' AFTER `order`, WATERMARK FOR ts AS ts)")
// 删除 watermark
tableEnv.executeSql("ALTER TABLE Orders DROP WATERMARK")
// 删除列
tableEnv.executeSql("ALTER TABLE Orders DROP (amount, ts, category)")
// 重命名列
tableEnv.executeSql("ALTER TABLE Orders RENAME `order` TO order_id")
// 字符串数组: ["Orders"]
val tables = tableEnv.listTables()
// or tableEnv.executeSql("SHOW TABLES").print()
// rename "Orders" to "NewOrders"
tableEnv.executeSql("ALTER TABLE Orders RENAME TO NewOrders")
// 字符串数组:["NewOrders"]
val tables = tableEnv.listTables()
// or tableEnv.executeSql("SHOW TABLES").print()
table_env = TableEnvironment.create(...)
# 字符串数组: ["Orders"]
tables = table_env.list_tables()
# or table_env.execute_sql("SHOW TABLES").print()
# 新增列 `order` 并置于第一位
table_env.execute_sql("ALTER TABLE Orders ADD `order` INT COMMENT 'order identifier' FIRST");
# 新增更多列, 主键及 watermark
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)");
# 修改列类型, 列注释, 主键及 watermark
table_env.execute_sql("ALTER TABLE Orders MODIFY (amount DOUBLE NOT NULL, category STRING COMMENT 'category identifier' AFTER `order`, WATERMARK FOR ts AS ts)");
# 删除 watermark
table_env.execute_sql("ALTER TABLE Orders DROP WATERMARK");
# 删除列
table_env.execute_sql("ALTER TABLE Orders DROP (amount, ts, category)");
# 重命名列
table_env.execute_sql("ALTER TABLE Orders RENAME `order` TO order_id");
# 把 "Orders" 的表名改为 "NewOrders"
table_env.execute_sql("ALTER TABLE Orders RENAME TO NewOrders");
# 字符串数组:["NewOrders"]
tables = table_env.list_tables()
# or table_env.execute_sql("SHOW TABLES").print()
Flink SQL> CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...);
[INFO] Execute statement succeed.
Flink SQL> ALTER TABLE Orders ADD `order` INT COMMENT 'order identifier' FIRST;
[INFO] Execute statement succeed.
Flink SQL> DESCRIBE Orders;
+---------+--------+------+-----+--------+-----------+------------------+
| name | type | null | key | extras | watermark | comment |
+---------+--------+------+-----+--------+-----------+------------------+
| order | INT | TRUE | | | | order identifier |
| user | BIGINT | TRUE | | | | |
| product | STRING | TRUE | | | | |
| amount | INT | TRUE | | | | |
+---------+--------+------+-----+--------+-----------+------------------+
4 rows in set
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);
[INFO] Execute statement succeed.
Flink SQL> DESCRIBE Orders;
+----------+------------------------+-------+------------+--------+--------------------------+------------------+
| name | type | null | key | extras | watermark | comment |
+----------+------------------------+-------+------------+--------+--------------------------+------------------+
| order | INT | FALSE | PRI(order) | | | order identifier |
| user | BIGINT | TRUE | | | | |
| product | STRING | TRUE | | | | |
| category | STRING | TRUE | | | | |
| amount | INT | TRUE | | | | |
| ts | TIMESTAMP(3) *ROWTIME* | TRUE | | | `ts` - INTERVAL '1' HOUR | |
+----------+------------------------+-------+------------+--------+--------------------------+------------------+
6 rows in set
Flink SQL> ALTER TABLE Orders MODIFY (amount DOUBLE NOT NULL, category STRING COMMENT 'category identifier' AFTER `order`, WATERMARK FOR ts AS ts);
[INFO] Execute statement succeed.
Flink SQL> DESCRIBE Orders;
+----------+------------------------+-------+------------+--------+-----------+---------------------+
| name | type | null | key | extras | watermark | comment |
+----------+------------------------+-------+------------+--------+-----------+---------------------+
| order | INT | FALSE | PRI(order) | | | order identifier |
| category | STRING | TRUE | | | | category identifier |
| user | BIGINT | TRUE | | | | |
| product | STRING | TRUE | | | | |
| amount | DOUBLE | FALSE | | | | |
| ts | TIMESTAMP(3) *ROWTIME* | TRUE | | | `ts` | |
+----------+------------------------+-------+------------+--------+-----------+---------------------+
6 rows in set
Flink SQL> ALTER TABLE Orders DROP WATERMARK;
[INFO] Execute statement succeed.
Flink SQL> DESCRIBE Orders;
+----------+--------------+-------+------------+--------+-----------+---------------------+
| name | type | null | key | extras | watermark | comment |
+----------+--------------+-------+------------+--------+-----------+---------------------+
| order | INT | FALSE | PRI(order) | | | order identifier |
| category | STRING | TRUE | | | | category identifier |
| user | BIGINT | TRUE | | | | |
| product | STRING | TRUE | | | | |
| amount | DOUBLE | FALSE | | | | |
| ts | TIMESTAMP(3) | TRUE | | | | |
+----------+--------------+-------+------------+--------+-----------+---------------------+
6 rows in set
Flink SQL> ALTER TABLE Orders DROP (amount, ts, category);
[INFO] Execute statement succeed.
Flink SQL> DESCRIBE Orders;
+---------+--------+-------+------------+--------+-----------+------------------+
| name | type | null | key | extras | watermark | comment |
+---------+--------+-------+------------+--------+-----------+------------------+
| order | INT | FALSE | PRI(order) | | | order identifier |
| user | BIGINT | TRUE | | | | |
| product | STRING | TRUE | | | | |
+---------+--------+-------+------------+--------+-----------+------------------+
3 rows in set
Flink SQL> ALTER TABLE Orders RENAME `order` to `order_id`;
[INFO] Execute statement succeed.
Flink SQL> DESCRIBE Orders;
+----------+--------+-------+---------------+--------+-----------+------------------+
| name | type | null | key | extras | watermark | comment |
+----------+--------+-------+---------------+--------+-----------+------------------+
| order_id | INT | FALSE | PRI(order_id) | | | order identifier |
| user | BIGINT | TRUE | | | | |
| product | STRING | TRUE | | | | |
+----------+--------+-------+---------------+--------+-----------+------------------+
3 rows in set
Flink SQL> SHOW TABLES;
+------------+
| table name |
+------------+
| Orders |
+------------+
1 row in set
Flink SQL> ALTER TABLE Orders RENAME TO NewOrders;
[INFO] Execute statement succeed.
Flink SQL> SHOW TABLES;
+------------+
| table name |
+------------+
| NewOrders |
+------------+
1 row in set
ALTER TABLE #
当前支持的 ALTER TABLE 语法如下
ALTER TABLE [IF EXISTS] table_name {
ADD { <schema_component> | (<schema_component> [, ...]) | [IF NOT EXISTS] <partition_component> [<partition_component> ...]}
| MODIFY { <schema_component> | (<schema_component> [, ...]) }
| DROP {column_name | (column_name, column_name, ....) | PRIMARY KEY | CONSTRAINT constraint_name | WATERMARK | [IF EXISTS] <partition_component> [, ...]}
| RENAME old_column_name TO new_column_name
| RENAME TO new_table_name
| SET (key1=val1, ...)
| RESET (key1, ...)
}
<schema_component>:
{ <column_component> | <constraint_component> | <watermark_component> }
<column_component>:
column_name <column_definition> [FIRST | AFTER column_name]
<constraint_component>:
[CONSTRAINT constraint_name] PRIMARY KEY (column_name, ...) NOT ENFORCED
<watermark_component>:
WATERMARK FOR rowtime_column_name AS watermark_strategy_expression
<column_definition>:
{ <physical_column_definition> | <metadata_column_definition> | <computed_column_definition> } [COMMENT column_comment]
<physical_column_definition>:
column_type
<metadata_column_definition>:
column_type METADATA [ FROM metadata_key ] [ VIRTUAL ]
<computed_column_definition>:
AS computed_column_expression
<partition_component>:
PARTITION (key1=val1, key2=val2, ...) [WITH (key1=val1, key2=val2, ...)]
IF EXISTS
若表不存在,则不进行任何操作。
ADD #
使用 ADD
语句向已有表中增加 columns, constraints,watermark, partitions。
向表新增列时可通过 FIRST
or AFTER col_name
指定位置,不指定位置时默认追加在最后。
ADD
语句示例如下。
-- 新增一列
ALTER TABLE MyTable ADD category_id STRING COMMENT 'identifier of the category';
-- 新增列,主键和 watermark
ALTER TABLE MyTable ADD (
log_ts STRING COMMENT 'log timestamp string' FIRST,
ts AS TO_TIMESTAMP(log_ts) AFTER log_ts,
PRIMARY KEY (id) NOT ENFORCED,
WATERMARK FOR ts AS ts - INTERVAL '3' SECOND
);
-- 新增一个分区
ALTER TABLE MyTable ADD PARTITION (p1=1,p2='a') with ('k1'='v1');
-- 新增两个分区
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
语句示例如下。
-- modify a column type, comment and position
ALTER TABLE MyTable MODIFY measurement double COMMENT 'unit is bytes per second' AFTER `id`;
-- modify definition of column log_ts and ts, primary key, watermark. They must exist in table schema
ALTER TABLE MyTable MODIFY (
log_ts STRING COMMENT 'log timestamp string' AFTER `id`, -- reorder columns
ts AS TO_TIMESTAMP(log_ts) AFTER log_ts,
PRIMARY KEY (id) NOT ENFORCED,
WATERMARK FOR ts AS ts -- modify watermark strategy
);
注意 指定列为主键列时会隐式修改该列的 nullability 为 false。
DROP #
使用 DROP
语句删除列 、主键 、 分区或 watermark。
DROP
语句示例如下。
-- 删除一个列
ALTER TABLE MyTable DROP measurement;
-- 删除多个列
ALTER TABLE MyTable DROP (col1, col2, col3);
-- 删除主键
ALTER TABLE MyTable DROP PRIMARY KEY;
-- 删除一个分区
ALTER TABLE MyTable DROP PARTITION (`id` = 1);
-- 删除两个分区
ALTER TABLE MyTable DROP PARTITION (`id` = 1), PARTITION (`id` = 2);
-- 删除 watermark
ALTER TABLE MyTable DROP WATERMARK;
RENAME #
使用 RENAME
语句修改列名或表名。
RENAME
语句示例如下。
-- rename column
ALTER TABLE MyTable RENAME request_body TO payload;
-- rename table
ALTER TABLE MyTable RENAME TO MyTable2;
SET #
为指定的表设置一个或多个属性。若个别属性已经存在于表中,则使用新值覆盖旧值。
SET
语句示例如下。
-- set 'rows-per-second'
ALTER TABLE DataGenSource SET ('rows-per-second' = '10');
RESET #
为指定的表重置一个或多个属性。
RESET
语句示例如下。
-- reset 'rows-per-second' to the default value
ALTER TABLE DataGenSource RESET ('rows-per-second');
ALTER VIEW #
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.
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 #
ALTER DATABASE [catalog_name.]db_name SET (key1=val1, key2=val2, ...)
在数据库中设置一个或多个属性。若个别属性已经在数据库中设定,将会使用新值覆盖旧值。
ALTER FUNCTION #
ALTER [TEMPORARY|TEMPORARY SYSTEM] FUNCTION
[IF EXISTS] [catalog_name.][db_name.]function_name
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。