diff --git a/docs/content.zh/docs/dev/table/sql/show.md b/docs/content.zh/docs/dev/table/sql/show.md index 0054d665dfd244..0297cdd0b01623 100644 --- a/docs/content.zh/docs/dev/table/sql/show.md +++ b/docs/content.zh/docs/dev/table/sql/show.md @@ -28,7 +28,7 @@ under the License. -SHOW 语句用于列出所有的 catalog，或者列出当前 catalog 中所有的 database，或者列出当前 catalog 和当前 database 的所有表或视图，或者列出当前正在使用的 catalog 和 database, 或者列出当前 catalog 和当前 database 中所有的 function，包括：系统 function 和用户定义的 function，或者仅仅列出当前 catalog 和当前 database 中用户定义的 function，或者列出当前环境所有激活的 module，或者列出当前环境所有加载的 module 及激活状态。 +SHOW 语句用于列出所有的 catalog，或者列出当前 catalog 中所有的 database，或者列出当前 catalog 和当前 database 的所有表或视图，或者列出当前正在使用的 catalog 和 database, 或者列出创建指定表的语句，或者列出当前 catalog 和当前 database 中所有的 function，包括：系统 function 和用户定义的 function，或者仅仅列出当前 catalog 和当前 database 中用户定义的 function，或者列出当前环境所有激活的 module，或者列出当前环境所有加载的 module 及激活状态。 目前 Flink SQL 支持下列 SHOW 语句： - SHOW CATALOGS @@ -36,6 +36,7 @@ SHOW 语句用于列出所有的 catalog，或者列出当前 catalog 中所有 - SHOW DATABASES - SHOW CURRENT DATABASE - SHOW TABLES +- SHOW CREATE TABLE - SHOW VIEWS - SHOW FUNCTIONS - SHOW MODULES @@ -120,6 +121,15 @@ tEnv.executeSql("SHOW TABLES").print(); // | my_table | // +------------+ +// show create table +tEnv.executeSql("SHOW CREATE TABLE my_table").print(); +// CREATE TABLE `default_catalog`.`default_db`.`my_table` ( +// ... +// ) WITH ( +// ... +// ) + + // create a view tEnv.executeSql("CREATE VIEW my_view AS ..."); // show views @@ -201,6 +211,13 @@ tEnv.executeSql("SHOW TABLES").print() // | my_table | // +------------+ +// show create table +tEnv.executeSql("SHOW CREATE TABLE my_table").print() +// CREATE TABLE `default_catalog`.`default_db`.`my_table` ( +// ... +// ) WITH ( +// ... +// ) // create a view tEnv.executeSql("CREATE VIEW my_view AS ...") // show views @@ -281,6 +298,13 @@ table_env.execute_sql("SHOW TABLES").print() # +------------+ # | my_table | # +------------+ +# show create table +table_env.executeSql("SHOW CREATE TABLE my_table").print() +# CREATE TABLE `default_catalog`.`default_db`.`my_table` ( +# ... +# ) WITH ( +# ... +# ) # create a view table_env.execute_sql("CREATE VIEW my_view AS ...") @@ -346,6 +370,13 @@ Flink SQL> CREATE TABLE my_table (...) WITH (...); Flink SQL> SHOW TABLES; my_table +Flink SQL> SHOW CREATE TABLE my_table; +CREATE TABLE `default_catalog`.`default_db`.`my_table` ( + ... +) WITH ( + ... +) + Flink SQL> CREATE VIEW my_view AS ...; [INFO] View has been created. @@ -428,6 +459,16 @@ SHOW TABLES 展示当前 catalog 和当前 database 中所有的表。 +## SHOW CREATE TABLE + +```sql +SHOW CREATE TABLE [catalog_name.][db_name.]table_name +``` + +展示创建指定表的 create 语句。 + +Attention 目前 `SHOW CREATE TABLE` 只支持通过 Flink SQL DDL 创建的表。 + ## SHOW VIEWS ```sql diff --git a/docs/content/docs/dev/table/sql/show.md b/docs/content/docs/dev/table/sql/show.md index 5314281eca9fff..9ea474b893f516 100644 --- a/docs/content/docs/dev/table/sql/show.md +++ b/docs/content/docs/dev/table/sql/show.md @@ -26,7 +26,7 @@ under the License. # SHOW Statements -SHOW statements are used to list all catalogs, or list all databases in the current catalog, or list all tables/views in the current catalog and the current database, or show current catalog and database, or list all functions including system functions and user-defined functions in the current catalog and current database, or list only user-defined functions in the current catalog and current database, or list enabled module names, or list all loaded modules with enabled status in the current session. +SHOW statements are used to list all catalogs, or list all databases in the current catalog, or list all tables/views in the current catalog and the current database, or show current catalog and database, or show create statement for specified table, or list all functions including system functions and user-defined functions in the current catalog and current database, or list only user-defined functions in the current catalog and current database, or list enabled module names, or list all loaded modules with enabled status in the current session. Flink SQL supports the following SHOW statements for now: - SHOW CATALOGS @@ -34,6 +34,7 @@ Flink SQL supports the following SHOW statements for now: - SHOW DATABASES - SHOW CURRENT DATABASE - SHOW TABLES +- SHOW CREATE TABLE - SHOW VIEWS - SHOW FUNCTIONS - SHOW MODULES @@ -119,6 +120,15 @@ tEnv.executeSql("SHOW TABLES").print(); // | my_table | // +------------+ +// show create table +tEnv.executeSql("SHOW CREATE TABLE my_table").print(); +// CREATE TABLE `default_catalog`.`default_db`.`my_table` ( +// ... +// ) WITH ( +// ... +// ) + + // create a view tEnv.executeSql("CREATE VIEW my_view AS ..."); // show views @@ -200,6 +210,13 @@ tEnv.executeSql("SHOW TABLES").print() // | my_table | // +------------+ +// show create table +tEnv.executeSql("SHOW CREATE TABLE my_table").print() +// CREATE TABLE `default_catalog`.`default_db`.`my_table` ( +// ... +// ) WITH ( +// ... +// ) // create a view tEnv.executeSql("CREATE VIEW my_view AS ...") // show views @@ -280,6 +297,13 @@ table_env.execute_sql("SHOW TABLES").print() # +------------+ # | my_table | # +------------+ +# show create table +table_env.executeSql("SHOW CREATE TABLE my_table").print() +# CREATE TABLE `default_catalog`.`default_db`.`my_table` ( +# ... +# ) WITH ( +# ... +# ) # create a view table_env.execute_sql("CREATE VIEW my_view AS ...") @@ -345,6 +369,13 @@ Flink SQL> CREATE TABLE my_table (...) WITH (...); Flink SQL> SHOW TABLES; my_table +Flink SQL> SHOW CREATE TABLE my_table; +CREATE TABLE `default_catalog`.`default_db`.`my_table` ( + ... +) WITH ( + ... +) + Flink SQL> CREATE VIEW my_view AS ...; [INFO] View has been created. @@ -427,6 +458,17 @@ SHOW TABLES Show all tables in the current catalog and the current database. + +## SHOW CREATE TABLE + +```sql +SHOW CREATE TABLE +``` + +Show create table statement for specified table. + +Attention Currently `SHOW CREATE TABLE` only supports table that is created by Flink SQL DDL. + ## SHOW VIEWS ```sql diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java index a738c977d46e13..e3a526a9fd0910 100644 --- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java +++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java @@ -35,6 +35,7 @@ import org.apache.flink.table.operations.ModifyOperation; import org.apache.flink.table.operations.Operation; import org.apache.flink.table.operations.QueryOperation; +import org.apache.flink.table.operations.ShowCreateTableOperation; import org.apache.flink.table.operations.UnloadModuleOperation; import org.apache.flink.table.operations.UseOperation; import org.apache.flink.table.operations.command.ClearOperation; @@ -414,6 +415,9 @@ private void callOperation(Operation operation, ExecutionMode mode) { } else if (operation instanceof EndStatementSetOperation) { // END callEndStatementSet(); + } else if (operation instanceof ShowCreateTableOperation) { + // SHOW CREATE TABLE + callShowCreateTable((ShowCreateTableOperation) operation); } else { // fallback to default implementation executeOperation(operation); @@ -527,6 +531,14 @@ private void callInserts(List operations) { } public void callExplain(ExplainOperation operation) { + printRawContent(operation); + } + + public void callShowCreateTable(ShowCreateTableOperation operation) { + printRawContent(operation); + } + + public void printRawContent(Operation operation) { TableResult tableResult = executor.executeOperation(sessionId, operation); // show raw content instead of tableau style final String explanation = diff --git a/flink-table/flink-sql-client/src/test/resources/sql/table.q b/flink-table/flink-sql-client/src/test/resources/sql/table.q index b89c746b28e2ce..44957b82a17dd6 100644 --- a/flink-table/flink-sql-client/src/test/resources/sql/table.q +++ b/flink-table/flink-sql-client/src/test/resources/sql/table.q @@ -73,6 +73,22 @@ show tables; 1 row in set !ok +# test SHOW CREATE TABLE +show create table orders; +CREATE TABLE `default_catalog`.`default_database`.`orders` ( + `user` BIGINT NOT NULL, + `product` VARCHAR(32), + `amount` INT, + `ts` TIMESTAMP(3), + `ptime` AS PROCTIME(), + WATERMARK FOR `ts` AS `ts` - INTERVAL '1' SECOND, + CONSTRAINT `PK_3599338` PRIMARY KEY (`user`) NOT ENFORCED +) WITH ( + 'connector' = 'datagen' +) + +!ok + # ========================================================================== # test alter table # ========================================================================== @@ -117,6 +133,22 @@ desc orders2; 5 rows in set !ok +# test SHOW CREATE TABLE +show create table orders2; +CREATE TABLE `default_catalog`.`default_database`.`orders2` ( + `user` BIGINT NOT NULL, + `product` VARCHAR(32), + `amount` INT, + `ts` TIMESTAMP(3), + `ptime` AS PROCTIME(), + WATERMARK FOR `ts` AS `ts` - INTERVAL '1' SECOND, + CONSTRAINT `PK_3599338` PRIMARY KEY (`user`) NOT ENFORCED +) WITH ( + 'connector' = 'kafka' +) + +!ok + # ========================================================================== # test drop table # ========================================================================== @@ -165,6 +197,18 @@ show tables; 1 row in set !ok +# SHOW CREATE TABLE for temporary table +show create table tbl1; +CREATE TEMPORARY TABLE `default_catalog`.`default_database`.`tbl1` ( + `user` BIGINT NOT NULL, + `product` VARCHAR(32), + `amount` INT +) WITH ( + 'connector' = 'datagen' +) + +!ok + drop temporary table tbl1; [INFO] Execute statement succeed. !info diff --git a/flink-table/flink-sql-client/src/test/resources/sql/view.q b/flink-table/flink-sql-client/src/test/resources/sql/view.q index e8e9d624830f20..d2e115ceb4715e 100644 --- a/flink-table/flink-sql-client/src/test/resources/sql/view.q +++ b/flink-table/flink-sql-client/src/test/resources/sql/view.q @@ -72,6 +72,12 @@ show views; 2 rows in set !ok +# test SHOW CREATE TABLE for views +show create table v1; +[ERROR] Could not execute SQL statement. Reason: +org.apache.flink.table.api.TableException: SHOW CREATE TABLE does not support showing CREATE VIEW statement with identifier `default_catalog`.`default_database`.`v1`. +!error + # ==== test permanent view ===== # register a permanent view with the duplicate name with temporary view diff --git a/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd b/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd index 5ad878b684cf68..8134d906428af8 100644 --- a/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd +++ b/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd @@ -69,6 +69,7 @@ "org.apache.flink.sql.parser.dql.SqlShowFunctions" "org.apache.flink.sql.parser.dql.SqlShowModules" "org.apache.flink.sql.parser.dql.SqlShowTables" + "org.apache.flink.sql.parser.dql.SqlShowCreateTable" "org.apache.flink.sql.parser.dql.SqlShowViews" "org.apache.flink.sql.parser.dql.SqlRichDescribeTable" "org.apache.flink.sql.parser.dql.SqlUnloadModule" @@ -478,6 +479,7 @@ "SqlAlterFunction()" "SqlShowFunctions()" "SqlShowTables()" + "SqlShowCreateTable()" "SqlRichDescribeTable()" "SqlAlterTable()" "SqlShowModules()" diff --git a/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl b/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl index b4cfc34d8fbae7..81cf8fba592e1e 100644 --- a/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl +++ b/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl @@ -422,6 +422,22 @@ SqlShowTables SqlShowTables() : } } +/** +* Parse a "Show Create Table" query command. +*/ +SqlShowCreateTable SqlShowCreateTable() : +{ + SqlIdentifier tableName; + SqlParserPos pos; +} +{ + { pos = getPos();} + tableName = CompoundIdentifier() + { + return new SqlShowCreateTable(pos, tableName); + } +} + /** * DESCRIBE | DESC [ EXTENDED] [[catalogName.] dataBasesName].tableName sql call. * Here we add Rich in className to distinguish from calcite's original SqlDescribeTable. diff --git a/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/dql/SqlShowCreateTable.java b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/dql/SqlShowCreateTable.java new file mode 100644 index 00000000000000..447a044c7e37ce --- /dev/null +++ b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/dql/SqlShowCreateTable.java @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.sql.parser.dql; + +import org.apache.calcite.sql.SqlCall; +import org.apache.calcite.sql.SqlIdentifier; +import org.apache.calcite.sql.SqlKind; +import org.apache.calcite.sql.SqlNode; +import org.apache.calcite.sql.SqlOperator; +import org.apache.calcite.sql.SqlSpecialOperator; +import org.apache.calcite.sql.SqlWriter; +import org.apache.calcite.sql.parser.SqlParserPos; + +import java.util.Collections; +import java.util.List; + +/** SHOW CREATE TABLE sql call. */ +public class SqlShowCreateTable extends SqlCall { + + public static final SqlSpecialOperator OPERATOR = + new SqlSpecialOperator("SHOW CREATE TABLE", SqlKind.OTHER_DDL); + private final SqlIdentifier tableName; + + public SqlShowCreateTable(SqlParserPos pos, SqlIdentifier tableName) { + super(pos); + this.tableName = tableName; + } + + public SqlIdentifier getTableName() { + return tableName; + } + + public String[] getFullTableName() { + return tableName.names.toArray(new String[0]); + } + + @Override + public SqlOperator getOperator() { + return OPERATOR; + } + + @Override + public List getOperandList() { + return Collections.singletonList(tableName); + } + + @Override + public void unparse(SqlWriter writer, int leftPrec, int rightPrec) { + writer.keyword("SHOW CREATE TABLE"); + tableName.unparse(writer, leftPrec, rightPrec); + } +} diff --git a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java index a222321f3ac97b..53dc861a5efc34 100644 --- a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java +++ b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java @@ -191,6 +191,12 @@ public void testShowTables() { sql("show tables").ok("SHOW TABLES"); } + @Test + public void testShowCreateTable() { + sql("show create table tbl").ok("SHOW CREATE TABLE `TBL`"); + sql("show create table catalog1.db1.tbl").ok("SHOW CREATE TABLE `CATALOG1`.`DB1`.`TBL`"); + } + @Test public void testDescribeTable() { sql("describe tbl").ok("DESCRIBE `TBL`"); diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java index 1259cdf5e289e8..19720212ad12ca 100644 --- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java @@ -54,6 +54,8 @@ import org.apache.flink.table.catalog.ObjectIdentifier; import org.apache.flink.table.catalog.ObjectPath; import org.apache.flink.table.catalog.QueryOperationCatalogView; +import org.apache.flink.table.catalog.ResolvedCatalogBaseTable; +import org.apache.flink.table.catalog.ResolvedCatalogTable; import org.apache.flink.table.catalog.ResolvedSchema; import org.apache.flink.table.catalog.UnresolvedIdentifier; import org.apache.flink.table.catalog.WatermarkSpec; @@ -96,6 +98,7 @@ import org.apache.flink.table.operations.QueryOperation; import org.apache.flink.table.operations.SelectSinkOperation; import org.apache.flink.table.operations.ShowCatalogsOperation; +import org.apache.flink.table.operations.ShowCreateTableOperation; import org.apache.flink.table.operations.ShowCurrentCatalogOperation; import org.apache.flink.table.operations.ShowCurrentDatabaseOperation; import org.apache.flink.table.operations.ShowDatabasesOperation; @@ -144,11 +147,14 @@ import org.apache.flink.table.types.AbstractDataType; import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.utils.EncodingUtils; import org.apache.flink.table.utils.PrintUtils; import org.apache.flink.table.utils.TableSchemaUtils; import org.apache.flink.types.Row; import org.apache.flink.util.Preconditions; +import org.apache.commons.lang3.StringUtils; + import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -1121,6 +1127,33 @@ public TableResult executeInternal(Operation operation) { return TableResultImpl.TABLE_RESULT_OK; } else if (operation instanceof ShowCatalogsOperation) { return buildShowResult("catalog name", listCatalogs()); + } else if (operation instanceof ShowCreateTableOperation) { + ShowCreateTableOperation showCreateTableOperation = + (ShowCreateTableOperation) operation; + Optional result = + catalogManager.getTable(showCreateTableOperation.getTableIdentifier()); + if (result.isPresent()) { + return TableResultImpl.builder() + .resultKind(ResultKind.SUCCESS_WITH_CONTENT) + .schema(ResolvedSchema.of(Column.physical("result", DataTypes.STRING()))) + .data( + Collections.singletonList( + Row.of( + buildShowCreateTableRow( + result.get().getResolvedTable(), + showCreateTableOperation + .getTableIdentifier(), + result.get().isTemporary())))) + .setPrintStyle(TableResultImpl.PrintStyle.rawContent()) + .build(); + } else { + throw new ValidationException( + String.format( + "Could not execute SHOW CREATE TABLE. Table with identifier %s does not exist.", + showCreateTableOperation + .getTableIdentifier() + .asSerializableString())); + } } else if (operation instanceof ShowCurrentCatalogOperation) { return buildShowResult( "current catalog name", new String[] {catalogManager.getCurrentCatalog()}); @@ -1293,6 +1326,112 @@ private TableResult buildShowResult(String columnName, String[] objects) { Arrays.stream(objects).map((c) -> new String[] {c}).toArray(String[][]::new)); } + private String buildShowCreateTableRow( + ResolvedCatalogBaseTable table, + ObjectIdentifier tableIdentifier, + boolean isTemporary) { + final String printIndent = " "; + CatalogBaseTable.TableKind kind = table.getTableKind(); + if (kind == CatalogBaseTable.TableKind.VIEW) { + throw new TableException( + String.format( + "SHOW CREATE TABLE does not support showing CREATE VIEW statement with identifier %s.", + tableIdentifier.asSerializableString())); + } + StringBuilder sb = + new StringBuilder( + String.format( + "CREATE %sTABLE %s (\n", + isTemporary ? "TEMPORARY " : "", + tableIdentifier.asSerializableString())); + ResolvedSchema schema = table.getResolvedSchema(); + // append columns + sb.append( + schema.getColumns().stream() + .map(column -> String.format("%s%s", printIndent, getColumnString(column))) + .collect(Collectors.joining(",\n"))); + // append watermark spec + if (!schema.getWatermarkSpecs().isEmpty()) { + sb.append(",\n"); + sb.append( + schema.getWatermarkSpecs().stream() + .map( + watermarkSpec -> + String.format( + "%sWATERMARK FOR %s AS %s", + printIndent, + EncodingUtils.escapeIdentifier( + watermarkSpec.getRowtimeAttribute()), + watermarkSpec + .getWatermarkExpression() + .asSerializableString())) + .collect(Collectors.joining("\n"))); + } + // append constraint + if (schema.getPrimaryKey().isPresent()) { + sb.append(",\n"); + sb.append(String.format("%s%s", printIndent, schema.getPrimaryKey().get())); + } + sb.append("\n) "); + // append comment + String comment = table.getComment(); + if (StringUtils.isNotEmpty(comment)) { + sb.append(String.format("COMMENT '%s'\n", comment)); + } + // append partitions + ResolvedCatalogTable catalogTable = (ResolvedCatalogTable) table; + if (catalogTable.isPartitioned()) { + sb.append("PARTITIONED BY (") + .append( + catalogTable.getPartitionKeys().stream() + .map(EncodingUtils::escapeIdentifier) + .collect(Collectors.joining(", "))) + .append(")\n"); + } + // append `with` properties + Map options = table.getOptions(); + sb.append("WITH (\n") + .append( + options.entrySet().stream() + .map( + entry -> + String.format( + "%s'%s' = '%s'", + printIndent, + entry.getKey(), + entry.getValue())) + .collect(Collectors.joining(",\n"))) + .append("\n)\n"); + return sb.toString(); + } + + private String getColumnString(Column column) { + final StringBuilder sb = new StringBuilder(); + sb.append(EncodingUtils.escapeIdentifier(column.getName())); + sb.append(" "); + // skip data type for computed column + if (column instanceof Column.ComputedColumn) { + sb.append( + column.explainExtras() + .orElseThrow( + () -> + new TableException( + String.format( + "Column expression can not be null for computed column '%s'", + column.getName())))); + } else { + sb.append(column.getDataType().getLogicalType().asSerializableString()); + column.explainExtras() + .ifPresent( + e -> { + sb.append(" "); + sb.append(e); + }); + } + // TODO: Print the column comment until FLINK-18958 is fixed + return sb.toString(); + } + private TableResult buildShowFullModulesResult(ModuleEntry[] moduleEntries) { Object[][] rows = Arrays.stream(moduleEntries) diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowCreateTableOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowCreateTableOperation.java new file mode 100644 index 00000000000000..47f6b3eed7ddf0 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowCreateTableOperation.java @@ -0,0 +1,40 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.operations; + +import org.apache.flink.table.catalog.ObjectIdentifier; + +/** Operation to describe a SHOW CREATE TABLE statement. */ +public class ShowCreateTableOperation implements ShowOperation { + + private final ObjectIdentifier tableIdentifier; + + public ShowCreateTableOperation(ObjectIdentifier sqlIdentifier) { + this.tableIdentifier = sqlIdentifier; + } + + public ObjectIdentifier getTableIdentifier() { + return tableIdentifier; + } + + @Override + public String asSummaryString() { + return String.format("SHOW CREATE TABLE %s", tableIdentifier.asSummaryString()); + } +} diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/UniqueConstraint.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/UniqueConstraint.java index a60f0d2f831bed..722a51d3b9f168 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/UniqueConstraint.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/UniqueConstraint.java @@ -56,6 +56,17 @@ public List getColumns() { return columns; } + private String getTypeString() { + switch (getType()) { + case PRIMARY_KEY: + return "PRIMARY KEY"; + case UNIQUE_KEY: + return "UNIQUE"; + default: + throw new IllegalStateException("Unknown key type: " + getType()); + } + } + @Override public ConstraintType getType() { return type; @@ -67,27 +78,15 @@ public ConstraintType getType() { * 
      * CONSTRAINT [constraint-name] [constraint-type] ([constraint-definition])
      *
-     * E.g CONSTRAINT pk PRIMARY KEY (f0, f1) NOT ENFORCED
+     * E.g CONSTRAINT pk PRIMARY KEY (`f0`, `f1`) NOT ENFORCED
      *
*/ @Override public final String asSummaryString() { - final String typeString; - switch (getType()) { - case PRIMARY_KEY: - typeString = "PRIMARY KEY"; - break; - case UNIQUE_KEY: - typeString = "UNIQUE"; - break; - default: - throw new IllegalStateException("Unknown key type: " + getType()); - } - return String.format( "CONSTRAINT %s %s (%s)%s", EncodingUtils.escapeIdentifier(getName()), - typeString, + getTypeString(), columns.stream() .map(EncodingUtils::escapeIdentifier) .collect(Collectors.joining(", ")), diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/WatermarkSpec.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/WatermarkSpec.java index 5179684e91d797..f1c6342223ae60 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/WatermarkSpec.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/WatermarkSpec.java @@ -68,7 +68,7 @@ public ResolvedExpression getWatermarkExpression() { public String asSummaryString() { return "WATERMARK FOR " - + String.join(".", EncodingUtils.escapeIdentifier(rowtimeAttribute)) + + EncodingUtils.escapeIdentifier(rowtimeAttribute) + ": " + watermarkExpression.getOutputDataType() + " AS " diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/operations/SqlToOperationConverter.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/operations/SqlToOperationConverter.java index 538b9094e1c6c3..8d55d63adb408e 100644 --- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/operations/SqlToOperationConverter.java +++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/operations/SqlToOperationConverter.java @@ -55,6 +55,7 @@ import org.apache.flink.sql.parser.dql.SqlRichDescribeTable; import org.apache.flink.sql.parser.dql.SqlRichExplain; import org.apache.flink.sql.parser.dql.SqlShowCatalogs; +import org.apache.flink.sql.parser.dql.SqlShowCreateTable; import org.apache.flink.sql.parser.dql.SqlShowCurrentCatalog; import org.apache.flink.sql.parser.dql.SqlShowCurrentDatabase; import org.apache.flink.sql.parser.dql.SqlShowDatabases; @@ -94,6 +95,7 @@ import org.apache.flink.table.operations.LoadModuleOperation; import org.apache.flink.table.operations.Operation; import org.apache.flink.table.operations.ShowCatalogsOperation; +import org.apache.flink.table.operations.ShowCreateTableOperation; import org.apache.flink.table.operations.ShowCurrentCatalogOperation; import org.apache.flink.table.operations.ShowCurrentDatabaseOperation; import org.apache.flink.table.operations.ShowDatabasesOperation; @@ -256,6 +258,8 @@ public static Optional convert( return Optional.of(converter.convertDropFunction((SqlDropFunction) validated)); } else if (validated instanceof SqlAlterFunction) { return Optional.of(converter.convertAlterFunction((SqlAlterFunction) validated)); + } else if (validated instanceof SqlShowCreateTable) { + return Optional.of(converter.convertShowCreateTable((SqlShowCreateTable) validated)); } else if (validated instanceof SqlShowFunctions) { return Optional.of(converter.convertShowFunctions((SqlShowFunctions) validated)); } else if (validated instanceof SqlShowPartitions) { @@ -781,6 +785,14 @@ private Operation convertShowTables(SqlShowTables sqlShowTables) { return new ShowTablesOperation(); } + /** Convert SHOW CREATE TABLE statement. */ + private Operation convertShowCreateTable(SqlShowCreateTable sqlShowCreateTable) { + UnresolvedIdentifier unresolvedIdentifier = + UnresolvedIdentifier.of(sqlShowCreateTable.getFullTableName()); + ObjectIdentifier identifier = catalogManager.qualifyIdentifier(unresolvedIdentifier); + return new ShowCreateTableOperation(identifier); + } + /** Convert SHOW FUNCTIONS statement. */ private Operation convertShowFunctions(SqlShowFunctions sqlShowFunctions) { return new ShowFunctionsOperation( diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/catalog/CatalogTableITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/catalog/CatalogTableITCase.scala index da8a69c9f1b684..4f122e0cc4ca02 100644 --- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/catalog/CatalogTableITCase.scala +++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/catalog/CatalogTableITCase.scala @@ -20,7 +20,7 @@ package org.apache.flink.table.planner.catalog import org.apache.flink.table.api.config.{ExecutionConfigOptions, TableConfigOptions} import org.apache.flink.table.api.internal.TableEnvironmentImpl -import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment, ValidationException} +import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment, TableException, ValidationException} import org.apache.flink.table.catalog.{CatalogDatabaseImpl, CatalogFunctionImpl, GenericInMemoryCatalog, ObjectPath} import org.apache.flink.table.planner.expressions.utils.Func0 import org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory @@ -29,7 +29,7 @@ import org.apache.flink.table.planner.utils.DateTimeTestUtil.localDateTime import org.apache.flink.test.util.AbstractTestBase import org.apache.flink.types.Row import org.apache.flink.util.FileUtils -import org.junit.Assert.{assertEquals, assertTrue, fail} +import org.junit.Assert.{assertEquals, fail} import org.junit.rules.ExpectedException import org.junit.runner.RunWith import org.junit.runners.Parameterized @@ -39,7 +39,6 @@ import java.io.File import java.math.{BigDecimal => JBigDecimal} import java.net.URI import java.util - import scala.collection.JavaConversions._ /** Test cases for catalog table. */ @@ -985,6 +984,91 @@ class CatalogTableITCase(isStreamingMode: Boolean) extends AbstractTestBase { assertEquals(false, tableSchema2.getPrimaryKey.isPresent) } + @Test + def testCreateTableAndShowCreateTable(): Unit = { + val executedDDL = + """ + |create temporary table TBL1 ( + | a bigint not null, + | h string, + | g as 2*(a+1), + | b string not null, + | c bigint metadata virtual, + | e row, + | f as myfunc(a), + | ts1 timestamp(3), + | ts2 timestamp_ltz(3) metadata from 'timestamp', + | `__source__` varchar(255), + | proc as proctime(), + | watermark for ts1 as cast(timestampadd(hour, 8, ts1) as timestamp(3)), + | constraint test_constraint primary key (a, b) not enforced + |) comment 'test show create table statement' + |partitioned by (b,h) + |with ( + | 'connector' = 'kafka', + | 'kafka.topic' = 'log.test' + |) + |""".stripMargin + + val expectedDDL = + """ |CREATE TEMPORARY TABLE `default_catalog`.`default_database`.`TBL1` ( + | `a` BIGINT NOT NULL, + | `h` VARCHAR(2147483647), + | `g` AS 2 * (`a` + 1), + | `b` VARCHAR(2147483647) NOT NULL, + | `c` BIGINT METADATA VIRTUAL, + | `e` ROW<`name` VARCHAR(2147483647), `age` INT, `flag` BOOLEAN>, + | `f` AS `default_catalog`.`default_database`.`myfunc`(`a`), + | `ts1` TIMESTAMP(3), + | `ts2` TIMESTAMP(3) WITH LOCAL TIME ZONE METADATA FROM 'timestamp', + | `__source__` VARCHAR(255), + | `proc` AS PROCTIME(), + | WATERMARK FOR `ts1` AS CAST(TIMESTAMPADD(HOUR, 8, `ts1`) AS TIMESTAMP(3)), + | CONSTRAINT `test_constraint` PRIMARY KEY (`a`, `b`) NOT ENFORCED + |) COMMENT 'test show create table statement' + |PARTITIONED BY (`b`, `h`) + |WITH ( + | 'connector' = 'kafka', + | 'kafka.topic' = 'log.test' + |) + |""".stripMargin + tableEnv.executeSql(executedDDL) + val row = tableEnv.executeSql("SHOW CREATE TABLE `TBL1`").collect().next() + assertEquals(expectedDDL, row.getField(0)) + + expectedEx.expect(classOf[ValidationException]) + expectedEx.expectMessage( + "Could not execute SHOW CREATE TABLE. " + + "Table with identifier `default_catalog`.`default_database`.`tmp` does not exist.") + tableEnv.executeSql("SHOW CREATE TABLE `tmp`") + } + + @Test + def testCreateViewAndShowCreateTable(): Unit = { + val createTableDDL = + """ |create table `source` ( + | `id` bigint not null, + | `group` string not null, + | `score` double + |) with ( + | 'connector' = 'source-only' + |) + |""".stripMargin + val createViewDDL = + """ |create view `tmp` as + |select `group`, avg(`score`) as avg_score + |from `source` + |group by `group` + |""".stripMargin + tableEnv.executeSql(createTableDDL) + tableEnv.executeSql(createViewDDL) + expectedEx.expect(classOf[TableException]) + expectedEx.expectMessage( + "SHOW CREATE TABLE does not support showing CREATE VIEW statement with " + + "identifier `default_catalog`.`default_database`.`tmp`.") + tableEnv.executeSql("SHOW CREATE TABLE `tmp`") + } + @Test def testUseCatalogAndShowCurrentCatalog(): Unit = { tableEnv.registerCatalog("cat1", new GenericInMemoryCatalog("cat1"))