[FLINK-16384][table sql/client] Support SHOW CREATE TABLE statement

docs/content.zh/docs/dev/table/sql/show.md

@@ -28,14 +28,15 @@ 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
 - SHOW CURRENT 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 语句。
+
+<span class="label label-danger">Attention</span> 目前 `SHOW CREATE TABLE` 只支持通过 Flink SQL DDL 创建的表。
+
 ## SHOW VIEWS
 
 ```sql

docs/content/docs/dev/table/sql/show.md

@@ -26,14 +26,15 @@ 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
 - SHOW CURRENT CATALOG
 - 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.
+
+<span class="label label-danger">Attention</span> Currently `SHOW CREATE TABLE` only supports table that is created by Flink SQL DDL. 
+
 ## SHOW VIEWS
 
 ```sql

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<ModifyOperation> 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 =

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

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

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

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;
+}
+{
+    <SHOW> <CREATE> <TABLE> { 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.