ALTER 문

ALTER 문 (ALTER Statements)

ALTER 문은 Catalog에 이미 등록된 테이블, 뷰, 함수의 정의 또는 카탈로그 자체의 정의를 수정하는 데 사용됩니다. Flink SQL은 현재 다음 ALTER 문을 지원합니다: ALTER TABLE, ALTER MATERIALIZED TABLE, ALTER VIEW, ALTER DATABASE, ALTER FUNCTION, ALTER CATALOG, ALTER MODEL.

출처: 문서

본문

ALTER 문은 Catalog에 이미 등록된 테이블, 뷰, 함수의 정의 또는 카탈로그 자체의 정의를 수정하는 데 사용됩니다.

Flink SQL은 현재 다음 ALTER 문을 지원합니다.

ALTER TABLE
ALTER MATERIALIZED TABLE
ALTER VIEW
ALTER DATABASE
ALTER FUNCTION
ALTER CATALOG
ALTER MODEL

ALTER 문 실행 (Run an ALTER statement)

ALTER 문은 TableEnvironment의 executeSql() 메서드로 실행할 수 있습니다. executeSql() 메서드는 성공적인 ALTER 연산에 대해 'OK'를 반환하고, 그렇지 않으면 예외를 던집니다. SQL CLI에서도 ALTER 문을 실행할 수 있습니다.

다음 예시는 TableEnvironment에서 ALTER 문을 실행하는 방법을 보여줍니다.

Java

EnvironmentSettings settings = EnvironmentSettings.newInstance()...
TableEnvironment tableEnv = TableEnvironment.create(settings);

// register a table named "Orders"
tableEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)");

// a string array: ["Orders"]
String[] tables = tableEnv.listTables();
// or tableEnv.executeSql("SHOW TABLES").print();

// add a new column `order` to the first position
tableEnv.executeSql("ALTER TABLE Orders ADD `order` INT COMMENT 'order identifier' FIRST");

// add more columns, primary key and 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)");

// modify column type, column comment and watermark
tableEnv.executeSql("ALTER TABLE Orders MODIFY (amount DOUBLE NOT NULL, category STRING COMMENT 'category identifier' AFTER `order`, WATERMARK FOR ts AS ts)");

// drop watermark
tableEnv.executeSql("ALTER TABLE Orders DROP WATERMARK");

// drop column
tableEnv.executeSql("ALTER TABLE Orders DROP (amount, ts, category)");

// rename column
tableEnv.executeSql("ALTER TABLE Orders RENAME `order` TO order_id");

// rename "Orders" to "NewOrders"
tableEnv.executeSql("ALTER TABLE Orders RENAME TO NewOrders");

// a string array: ["NewOrders"]
String[] tables = tableEnv.listTables();
// or tableEnv.executeSql("SHOW TABLES").print();

// register a catalog named "cat2"
tableEnv.executeSql("CREATE CATALOG cat2 WITH ('type'='generic_in_memory')");

// add a new property `default-database`
tableEnv.executeSql("ALTER CATALOG cat2 SET ('default-database'='db')");

Scala

val tableEnv = TableEnvironment.create(...)

// register a table named "Orders"
tableEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)")

// add a new column `order` to the first position
tableEnv.executeSql("ALTER TABLE Orders ADD `order` INT COMMENT 'order identifier' FIRST")

// add more columns, primary key and 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)")

// modify column type, column comment and watermark
tableEnv.executeSql("ALTER TABLE Orders MODIFY (amount DOUBLE NOT NULL, category STRING COMMENT 'category identifier' AFTER `order`, WATERMARK FOR ts AS ts)")

// drop watermark
tableEnv.executeSql("ALTER TABLE Orders DROP WATERMARK")

// drop column
tableEnv.executeSql("ALTER TABLE Orders DROP (amount, ts, category)")

// rename column
tableEnv.executeSql("ALTER TABLE Orders RENAME `order` TO order_id")

// a string array: ["Orders"]
val tables = tableEnv.listTables()
// or tableEnv.executeSql("SHOW TABLES").print()

// rename "Orders" to "NewOrders"
tableEnv.executeSql("ALTER TABLE Orders RENAME TO NewOrders")

// a string array: ["NewOrders"]
val tables = tableEnv.listTables()
// or tableEnv.executeSql("SHOW TABLES").print()

// register a catalog named "cat2"
tableEnv.executeSql("CREATE CATALOG cat2 WITH ('type'='generic_in_memory')")

// add a new property `default-database`
tableEnv.executeSql("ALTER CATALOG cat2 SET ('default-database'='db')")

Python

table_env = TableEnvironment.create(...)

# a string array: ["Orders"]
tables = table_env.list_tables()
# or table_env.execute_sql("SHOW TABLES").print()

# add a new column `order` to the first position
table_env.execute_sql("ALTER TABLE Orders ADD `order` INT COMMENT 'order identifier' FIRST")

# add more columns, primary key and 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)")

# modify column type, column comment and 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)")

# drop watermark
table_env.execute_sql("ALTER TABLE Orders DROP WATERMARK")

# drop column
table_env.execute_sql("ALTER TABLE Orders DROP (amount, ts, category)")

# rename column
table_env.execute_sql("ALTER TABLE Orders RENAME `order` TO order_id")

# rename "Orders" to "NewOrders"
table_env.execute_sql("ALTER TABLE Orders RENAME TO NewOrders")

# a string array: ["NewOrders"]
tables = table_env.list_tables()
# or table_env.execute_sql("SHOW TABLES").print()

# register a catalog named "cat2"
table_env.execute_sql("CREATE CATALOG cat2 WITH ('type'='generic_in_memory')")

# add a new property `default-database`
table_env.execute_sql("ALTER CATALOG cat2 SET ('default-database'='db')")

SQL CLI

Flink SQL> CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...);
[INFO] Execute statement succeeded.

Flink SQL> ALTER TABLE Orders ADD `order` INT COMMENT 'order identifier' FIRST;
[INFO] Execute statement succeeded.

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 succeeded.

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 succeeded.

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 succeeded.

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 succeeded.

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 succeeded.

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 succeeded.

Flink SQL> SHOW TABLES;
+------------+
| table name |
+------------+
|  NewOrders |
+------------+
1 row in set

Flink SQL> CREATE CATALOG cat2 WITH ('type'='generic_in_memory');
[INFO] Execute statement succeeded.

Flink SQL> ALTER CATALOG cat2 SET ('default-database'='db_new');
[INFO] Execute statement succeeded.

Flink SQL> DESC CATALOG EXTENDED cat2;
+-------------------------+-------------------+
|               info name |        info value |
+-------------------------+-------------------+
|                    name |              cat2 |
|                    type | generic_in_memory |
|                 comment |                   |
| option:default-database |            db_new |
+-------------------------+-------------------+
4 rows in set

ALTER TABLE

다음 문법은 사용 가능한 구문에 대한 개요를 제공합니다.

ALTER TABLE [IF EXISTS] table_name {
    ADD { <schema_component> | (<schema_component> [, ...]) | [IF NOT EXISTS] <partition_component> [<partition_component> ...] | <distribution> }
  | MODIFY { <schema_component> | (<schema_component> [, ...]) | <distribution> }
  | DROP {column_name | (column_name, column_name, ....) | PRIMARY KEY | CONSTRAINT constraint_name | WATERMARK | [IF EXISTS] <partition_component> [, ...] | DISTRIBUTION }
  | 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, ...)]

<distribution>:
{
    DISTRIBUTION BY [ { HASH | RANGE } ] (bucket_column_name1, bucket_column_name2, ...) ] [INTO n BUCKETS]
  | DISTRIBUTION INTO n BUCKETS
}

IF EXISTS — 테이블이 존재하지 않으면 아무 일도 일어나지 않습니다.

ADD

ADD 절을 사용해 기존 테이블에 컬럼, 제약 조건, 워터마크, 파티션, distribution을 추가합니다.

지정된 위치에 컬럼을 추가하려면 FIRST 또는 AFTER col_name을 사용하세요. 기본적으로 컬럼은 마지막에 추가됩니다.

다음 예시는 ADD 문의 사용법을 보여줍니다.

-- add a new column
ALTER TABLE MyTable ADD category_id STRING COMMENT 'identifier of the category';

-- add columns, constraint, and 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
);

-- add a new partition
ALTER TABLE MyTable ADD PARTITION (p1=1,p2='a') with ('k1'='v1');

-- add two new partitions
ALTER TABLE MyTable ADD PARTITION (p1=1,p2='a') with ('k1'='v1') PARTITION (p1=1,p2='b') with ('k2'='v2');

-- add new distribution using a hash on uid into 4 buckets
ALTER TABLE MyTable ADD DISTRIBUTION BY HASH(uid) INTO 4 BUCKETS;

-- add new distribution on uid into 4 buckets
ALTER TABLE MyTable ADD DISTRIBUTION BY (uid) INTO 4 BUCKETS;

-- add new distribution on uid.
ALTER TABLE MyTable ADD DISTRIBUTION BY (uid);

-- add new distribution into 4 buckets
ALTER TABLE MyTable ADD DISTRIBUTION INTO 4 BUCKETS;

참고: 기본 키가 되는 컬럼을 추가하면 해당 컬럼의 nullability가 암시적으로 false로 변경됩니다.

MODIFY

MODIFY 절을 사용해 기존 테이블의 컬럼 위치, 타입, 주석 또는 nullability를 변경하고, 기본 키 컬럼과 워터마크 전략을 변경합니다.

기존 컬럼을 새 위치로 수정하려면 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 절을 사용해 기존 테이블에서 컬럼, 기본 키, 파티션, 워터마크 전략을 제거합니다.

다음 예시는 DROP 문의 사용법을 보여줍니다.

-- drop a column
ALTER TABLE MyTable DROP measurement;

-- drop columns
ALTER TABLE MyTable DROP (col1, col2, col3);

-- drop primary key
ALTER TABLE MyTable DROP PRIMARY KEY;

-- drop a partition
ALTER TABLE MyTable DROP PARTITION (`id` = 1);

-- drop two partitions
ALTER TABLE MyTable DROP PARTITION (`id` = 1), PARTITION (`id` = 2);

-- drop a watermark
ALTER TABLE MyTable DROP WATERMARK;

-- drop distribution
ALTER TABLE MyTable DROP DISTRIBUTION;

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 MATERIALIZED TABLE

Materialized tables에 대한 전용 페이지를 참조하세요.

ALTER VIEW

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

주어진 뷰를 같은 카탈로그와 데이터베이스 내에서 새 이름으로 바꿉니다.

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

주어진 뷰를 정의하는 기본 쿼리를 새 쿼리로 변경합니다.

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]

새 식별자와 선택적 언어 태그로 카탈로그 함수를 변경합니다. 함수가 카탈로그에 존재하지 않으면 예외가 발생합니다.

언어 태그가 JAVA/SCALA이면 식별자는 UDF의 전체 classpath입니다. Java/Scala UDF 구현에 대해서는 User-defined Functions를 참조하세요.

언어 태그가 PYTHON이면 식별자는 UDF의 정규화된 이름입니다. 예: pyflink.table.tests.test_udf.add. Python UDF 구현에 대해서는 Python UDFs를 참조하세요.

TEMPORARY — 카탈로그와 데이터베이스 네임스페이스를 가지며 카탈로그 함수를 오버라이드하는 임시 카탈로그 함수를 변경합니다.

TEMPORARY SYSTEM — 네임스페이스가 없고 내장 함수를 오버라이드하는 임시 시스템 함수를 변경합니다.

IF EXISTS — 함수가 존재하지 않으면 아무 일도 일어나지 않습니다.

LANGUAGE JAVA|SCALA|PYTHON — Flink 런타임이 함수를 실행하는 방법을 지시하는 언어 태그. 현재 JAVA, SCALA, PYTHON만 지원되며, 함수의 기본 언어는 JAVA입니다.

ALTER CATALOG

ALTER CATALOG catalog_name
    SET (key1=val1, ...)
  | RESET (key1, ...)
  | COMMENT 'comment'

SET

지정된 카탈로그에 하나 이상의 속성을 설정합니다. 특정 속성이 카탈로그에 이미 설정되어 있으면 새 값으로 이전 값을 덮어씁니다.

다음 예시는 SET 문의 사용법을 보여줍니다.

-- set 'default-database'
ALTER CATALOG cat2 SET ('default-database'='db');

RESET

지정된 카탈로그에서 하나 이상의 속성을 기본값으로 재설정합니다.

다음 예시는 RESET 문의 사용법을 보여줍니다.

-- reset 'default-database'
ALTER CATALOG cat2 RESET ('default-database');

COMMENT

지정된 카탈로그에 주석을 설정합니다. 주석이 카탈로그에 이미 설정되어 있으면 새 값으로 이전 값을 덮어씁니다.

다음 예시는 COMMENT 문의 사용법을 보여줍니다.

ALTER CATALOG cat2 COMMENT 'comment for catalog ''cat2''';

ALTER MODEL

ALTER MODEL [IF EXISTS] [catalog_name.][db_name.]model_name
    SET (key1=val1, ...)
  | RESET (key1, ...)
  | RENAME TO new_model_name

SET

지정된 모델에 하나 이상의 속성을 설정합니다. 특정 속성이 모델에 이미 설정되어 있으면 새 값으로 이전 값을 덮어씁니다.

다음 예시는 SET 문의 사용법을 보여줍니다.

-- set model properties
ALTER MODEL MyModel SET ('model-version'='2.0', 'batch-size'='32');

IF EXISTS — 모델이 존재하지 않으면 아무 일도 일어나지 않습니다.

RESET

지정된 모델에서 하나 이상의 속성을 기본값으로 재설정합니다.

다음 예시는 RESET 문의 사용법을 보여줍니다.

-- reset model properties
ALTER MODEL MyModel RESET ('model-version', 'batch-size');

RENAME TO

주어진 모델을 같은 카탈로그와 데이터베이스 내에서 새 이름으로 바꿉니다.

다음 예시는 RENAME TO 문의 사용법을 보여줍니다.

-- rename model
ALTER MODEL MyModel RENAME TO NewModel;

ALTER 문

ALTER 문 (ALTER Statements)

본문

ALTER 문 실행 (Run an ALTER statement)

ALTER TABLE

ADD

MODIFY

DROP

RENAME

SET

RESET

ALTER MATERIALIZED TABLE

ALTER VIEW

ALTER DATABASE

ALTER FUNCTION

ALTER CATALOG

SET

RESET

COMMENT

ALTER MODEL

SET

RESET

RENAME TO

더 알아보기 (Learn more)