ALTER 문

ALTER 문 (ALTER Statements)

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

출처: 문서

본문

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

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

ALTER 문 실행 (Run an ALTER statement)

Java / Scala:

ALTER 문은 TableEnvironment의 executeSql() 메서드로 실행할 수 있습니다. executeSql() 메서드는 성공적인 ALTER 작업에 대해 'OK'를 반환하고, 그렇지 않으면 예외를 던집니다. 다음 예제는 TableEnvironment에서 ALTER 문을 실행하는 방법을 보여줍니다.

Java:

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

// "Orders"라는 테이블 등록
tableEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)");

// 문자열 배열: ["Orders"]
String[] tables = tableEnv.listTables();
// 또는 tableEnv.executeSql("SHOW TABLES").print();

// 첫 번째 위치에 새 컬럼 `order` 추가
tableEnv.executeSql("ALTER TABLE Orders ADD `order` INT COMMENT 'order identifier' FIRST");

// 컬럼, primary key, 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();
// 또는 tableEnv.executeSql("SHOW TABLES").print();

// "cat2"라는 catalog 등록
tableEnv.executeSql("CREATE CATALOG cat2 WITH ('type'='generic_in_memory')");

// 새 속성 `default-database` 추가
tableEnv.executeSql("ALTER CATALOG cat2 SET ('default-database'='db')");

Scala:

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")

// 컬럼, primary key, 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()
// 또는 tableEnv.executeSql("SHOW TABLES").print()

// "Orders"를 "NewOrders"로 이름 변경
tableEnv.executeSql("ALTER TABLE Orders RENAME TO NewOrders")

// 문자열 배열: ["NewOrders"]
val tables = tableEnv.listTables()
// 또는 tableEnv.executeSql("SHOW TABLES").print()

// "cat2"라는 catalog 등록
tableEnv.executeSql("CREATE CATALOG cat2 WITH ('type'='generic_in_memory')")

// 새 속성 `default-database` 추가
tableEnv.executeSql("ALTER CATALOG cat2 SET ('default-database'='db')")

Python:

table_env = TableEnvironment.create(...)

# 문자열 배열: ["Orders"]
tables = table_env.list_tables()
# 또는 table_env.execute_sql("SHOW TABLES").print()

# 첫 번째 위치에 새 컬럼 `order` 추가
table_env.execute_sql("ALTER TABLE Orders ADD `order` INT COMMENT 'order identifier' FIRST")

# 컬럼, primary key, 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()
# 또는 table_env.execute_sql("SHOW TABLES").print()

# "cat2"라는 catalog 등록
table_env.execute_sql("CREATE CATALOG cat2 WITH ('type'='generic_in_memory')")

# 새 속성 `default-database` 추가
table_env.execute_sql("ALTER CATALOG cat2 SET ('default-database'='db')")

SQL CLI:

ALTER 문은 SQL CLI에서 실행할 수 있습니다. 다음은 SQL CLI에서 ALTER 문을 실행하는 예제입니다(the full transcript is kept from the original doc):

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> 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> 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> ALTER TABLE Orders DROP WATERMARK;
[INFO] Execute statement succeeded.

Flink SQL> ALTER TABLE Orders DROP (amount, ts, category);
[INFO] Execute statement succeeded.

Flink SQL> ALTER TABLE Orders RENAME `order` to `order_id`;
[INFO] Execute statement succeeded.

Flink SQL> ALTER TABLE Orders RENAME TO NewOrders;
[INFO] Execute statement succeeded.

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.

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 절을 사용해 기존 테이블에 컬럼, 제약 조건, watermark, 파티션, distribution을 추가합니다.

지정된 위치에 컬럼을 추가하려면 FIRST 또는 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');

-- uid에 해시를 사용해 4개 버킷으로 새 distribution 추가
ALTER TABLE MyTable ADD DISTRIBUTION BY HASH(uid) INTO 4 BUCKETS;

-- uid에 4개 버킷으로 새 distribution 추가
ALTER TABLE MyTable ADD DISTRIBUTION BY (uid) INTO 4 BUCKETS;

-- uid에 새 distribution 추가
ALTER TABLE MyTable ADD DISTRIBUTION BY (uid);

-- 4개 버킷으로 새 distribution 추가
ALTER TABLE MyTable ADD DISTRIBUTION INTO 4 BUCKETS;

참고: primary key가 될 컬럼을 추가하면 컬럼의 nullability가 암시적으로 false로 변경됩니다.

MODIFY

MODIFY 절을 사용해 기존 테이블의 컬럼 위치, 타입, 주석, nullability를 변경하고, primary key 컬럼과 watermark 전략을 변경합니다.

기존 컬럼을 새 위치로 수정하려면 FIRST 또는 AFTER col_name을 사용합니다. 기본적으로 위치는 변경되지 않습니다.

다음 예제는 MODIFY 문의 사용법을 보여줍니다:

-- 컬럼 타입, 주석, 위치 수정
ALTER TABLE MyTable MODIFY measurement double COMMENT 'unit is bytes per second' AFTER `id`;

-- 컬럼 log_ts와 ts, primary key, watermark의 정의 수정. 테이블 스키마에 존재해야 합니다.
ALTER TABLE MyTable MODIFY (
    log_ts STRING COMMENT 'log timestamp string' AFTER `id`,  -- 컬럼 순서 변경
    ts AS TO_TIMESTAMP(log_ts) AFTER log_ts,
    PRIMARY KEY (id) NOT ENFORCED,
    WATERMARK FOR ts AS ts -- watermark 전략 수정
);

참고: primary key가 되도록 컬럼을 수정하면 컬럼의 nullability가 암시적으로 false로 변경됩니다.

DROP

DROP 절을 사용해 기존 테이블의 컬럼, primary key, 파티션, watermark 전략을 제거합니다.

다음 예제는 DROP 문의 사용법을 보여줍니다:

-- 컬럼 삭제
ALTER TABLE MyTable DROP measurement;

-- 컬럼 여러 개 삭제
ALTER TABLE MyTable DROP (col1, col2, col3);

-- primary key 삭제
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;

-- distribution 삭제
ALTER TABLE MyTable DROP DISTRIBUTION;

RENAME

RENAME 절을 사용해 컬럼이나 기존 테이블의 이름을 변경합니다.

다음 예제는 RENAME 문의 사용법을 보여줍니다:

-- 컬럼 이름 변경
ALTER TABLE MyTable RENAME request_body TO payload;

-- 테이블 이름 변경
ALTER TABLE MyTable RENAME TO MyTable2;

SET

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

-- 'rows-per-second' 설정
ALTER TABLE DataGenSource SET ('rows-per-second' = '10');

RESET

하나 이상의 속성을 기본값으로 재설정합니다.

-- 'rows-per-second'를 기본값으로 재설정
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

주어진 뷰를 같은 catalog와 database 내에서 새 이름으로 이름 변경합니다.

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

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

ALTER FUNCTION

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

새 identifier와 선택적 language 태그로 catalog 함수를 변경합니다. 함수가 catalog에 존재하지 않으면 예외가 발생합니다.

language 태그가 JAVA/SCALA이면 identifier는 UDF의 전체 클래스 경로입니다. Java/Scala UDF 구현에 대한 자세한 내용은 User-defined Functions를 참고하세요.

language 태그가 PYTHON이면 identifier는 UDF의 완전한 이름입니다. 예: pyflink.table.tests.test_udf.add. Python UDF 구현에 대한 자세한 내용은 Python UDFs를 참고하세요.

TEMPORARY

catalog와 database 네임스페이스를 가지며 catalog 함수를 덮어쓰는 임시 catalog 함수를 변경합니다.

TEMPORARY SYSTEM

네임스페이스가 없고 내장 함수를 덮어쓰는 임시 시스템 함수를 변경합니다.

IF EXISTS

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

LANGUAGE JAVA|SCALA|PYTHON

Flink 런타임에게 함수를 어떻게 실행할지 지시하는 language 태그입니다. 현재 JAVA, SCALA, PYTHON만 지원되며 함수의 기본 언어는 JAVA입니다.

ALTER CATALOG

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

SET

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

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

RESET

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

-- 'default-database' 재설정
ALTER CATALOG cat2 RESET ('default-database');

COMMENT

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

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

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

-- model 속성 설정
ALTER MODEL MyModel SET ('model-version'='2.0', 'batch-size'='32');

IF EXISTS

model이 존재하지 않으면 아무 일도 일어나지 않습니다.

RESET

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

-- model 속성 재설정
ALTER MODEL MyModel RESET ('model-version', 'batch-size');

RENAME TO

주어진 model을 같은 catalog와 database 내에서 새 이름으로 이름 변경합니다.

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