You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/07/12 13:22:21 UTC
[GitHub] [flink] Tartarus0zm opened a new pull request, #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Tartarus0zm opened a new pull request, #20252:
URL: https://github.com/apache/flink/pull/20252
## What is the purpose of the change
Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
## Brief change log
Modify parserImpls.ftl to add AS query clause support for CREATE TABLE statements
## Verifying this change
add UT
* CreateTableAsSelectTest
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (not documented)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
Tartarus0zm commented on PR #20252:
URL: https://github.com/apache/flink/pull/20252#issuecomment-1182705141
@wuchong @godfreyhe @lsyldliu @luoyuxia @beyond1920 please take a look, if you have time, thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
flinkbot commented on PR #20252:
URL: https://github.com/apache/flink/pull/20252#issuecomment-1181763374
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ae1ae28deafb0b9514a9711b73ad4f136dec9fc2",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "ae1ae28deafb0b9514a9711b73ad4f136dec9fc2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ae1ae28deafb0b9514a9711b73ad4f136dec9fc2 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on a diff in pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
Tartarus0zm commented on code in PR #20252:
URL: https://github.com/apache/flink/pull/20252#discussion_r925578347
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTable.java:
##########
@@ -76,6 +76,8 @@ public class SqlCreateTable extends SqlCreate implements ExtendedSqlNode {
private final boolean isTemporary;
+ private final SqlNode query;
Review Comment:
good catch!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lsyldliu commented on a diff in pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
lsyldliu commented on code in PR #20252:
URL: https://github.com/apache/flink/pull/20252#discussion_r927563895
##########
flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl:
##########
@@ -1151,9 +1152,13 @@ SqlCreate SqlCreateTable(Span s, boolean replace, boolean isTemporary) :
[
<LIKE>
tableLike = SqlTableLike(getPos())
+ |
Review Comment:
```suggestion
{
return new SqlCreateTableLike(startPos.plus(getPos()),
tableName,
columnList,
constraints,
propertyList,
partitionColumns,
watermark,
comment,
tableLike,
isTemporary,
ifNotExists);
}
|
<AS>
asQuery = OrderedQueryOrExpr(ExprContext.ACCEPT_QUERY)
{
return new SqlCreateTableAs(startPos.plus(getPos()),
tableName,
columnList,
constraints,
propertyList,
partitionColumns,
watermark,
comment,
asQuery,
isTemporary,
ifNotExists);
}
]
{
return new SqlCreateTable(startPos.plus(getPos()),
tableName,
columnList,
constraints,
propertyList,
partitionColumns,
watermark,
comment,
isTemporary,
ifNotExists);
}
```
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+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 org.apache.calcite.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+/**
+ * {@link SqlNode} to describe the CREATE TABLE AS syntax. The CTAS syntax is different from CREATE
+ * TABLE syntax, which also create the pipeline to sync the data from the source to the derived
+ * table.
+ *
+ * <p>Example: A DDL like the one below for creating a `derived_table`
+ *
+ * <pre>{@code
+ * CREATE TABLE base_table (
+ * id BIGINT,
+ * name STRING,
+ * tstmp TIMESTAMP,
+ * PRIMARY KEY(id)
+ * ) WITH (
+ * ‘connector’ = ‘kafka’,
+ * ‘connector.starting-offset’: ‘12345’,
+ * ‘format’ = ‘json’
+ * )
+ *
+ * CREATE TABLE derived_table
+ * WITH (
+ * 'connector' = 'jdbc',
+ * 'url' = 'http://localhost:10000',
+ * 'table-name' = 'syncedTable'
+ * )
+ * AS SELECT * FROM base_table_1;
+ * }</pre>
+ */
+public class SqlCreateTableAs extends SqlCreateTable {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS", SqlKind.CREATE_TABLE);
+
+ private final SqlNode asQuery;
+
+ public SqlCreateTableAs(
+ SqlParserPos pos,
+ SqlIdentifier tableName,
+ SqlNodeList columnList,
+ List<SqlTableConstraint> tableConstraints,
+ SqlNodeList propertyList,
+ SqlNodeList partitionKeyList,
+ @Nullable SqlWatermark watermark,
+ @Nullable SqlCharStringLiteral comment,
+ SqlNode asQuery,
+ boolean isTemporary,
+ boolean ifNotExists) {
+ super(
+ OPERATOR,
+ pos,
+ tableName,
+ columnList,
+ tableConstraints,
+ propertyList,
+ partitionKeyList,
+ watermark,
+ comment,
+ isTemporary,
+ ifNotExists);
+ this.asQuery = asQuery;
+ }
+
+ @Override
+ public @Nonnull SqlOperator getOperator() {
+ return OPERATOR;
+ }
+
+ @Override
+ public @Nonnull List<SqlNode> getOperandList() {
+ return ImmutableNullableList.<SqlNode>builder()
+ .addAll(super.getOperandList())
+ .add(asQuery)
+ .build();
+ }
+
+ @Override
+ public void validate() throws SqlValidateException {
+ super.validate();
+ if (isTemporary()) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not support to create temporary table yet.");
+ }
+
+ if (getColumnList().size() > 0) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not support explicitly specifying columns yet.");
Review Comment:
```suggestion
"CREATE TABLE AS SELECT syntax does not support to specify explicit columns yet.");
```
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableLike.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+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 org.apache.calcite.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+/** CREATE TABLE LIKE DDL sql call. */
Review Comment:
Referring to `SqlCreateTableAs`, give more description.
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+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 org.apache.calcite.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+/**
+ * {@link SqlNode} to describe the CREATE TABLE AS syntax. The CTAS syntax is different from CREATE
+ * TABLE syntax, which also create the pipeline to sync the data from the source to the derived
+ * table.
+ *
+ * <p>Example: A DDL like the one below for creating a `derived_table`
+ *
+ * <pre>{@code
+ * CREATE TABLE base_table (
+ * id BIGINT,
+ * name STRING,
+ * tstmp TIMESTAMP,
+ * PRIMARY KEY(id)
+ * ) WITH (
+ * ‘connector’ = ‘kafka’,
+ * ‘connector.starting-offset’: ‘12345’,
+ * ‘format’ = ‘json’
+ * )
+ *
+ * CREATE TABLE derived_table
+ * WITH (
+ * 'connector' = 'jdbc',
+ * 'url' = 'http://localhost:10000',
+ * 'table-name' = 'syncedTable'
+ * )
+ * AS SELECT * FROM base_table_1;
Review Comment:
```suggestion
* AS SELECT * FROM base_table;
```
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+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 org.apache.calcite.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+/**
+ * {@link SqlNode} to describe the CREATE TABLE AS syntax. The CTAS syntax is different from CREATE
+ * TABLE syntax, which also create the pipeline to sync the data from the source to the derived
+ * table.
+ *
+ * <p>Example: A DDL like the one below for creating a `derived_table`
+ *
+ * <pre>{@code
+ * CREATE TABLE base_table (
+ * id BIGINT,
+ * name STRING,
+ * tstmp TIMESTAMP,
+ * PRIMARY KEY(id)
+ * ) WITH (
+ * ‘connector’ = ‘kafka’,
+ * ‘connector.starting-offset’: ‘12345’,
+ * ‘format’ = ‘json’
+ * )
+ *
+ * CREATE TABLE derived_table
+ * WITH (
+ * 'connector' = 'jdbc',
+ * 'url' = 'http://localhost:10000',
+ * 'table-name' = 'syncedTable'
+ * )
+ * AS SELECT * FROM base_table_1;
+ * }</pre>
+ */
+public class SqlCreateTableAs extends SqlCreateTable {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS", SqlKind.CREATE_TABLE);
+
+ private final SqlNode asQuery;
+
+ public SqlCreateTableAs(
+ SqlParserPos pos,
+ SqlIdentifier tableName,
+ SqlNodeList columnList,
+ List<SqlTableConstraint> tableConstraints,
+ SqlNodeList propertyList,
+ SqlNodeList partitionKeyList,
+ @Nullable SqlWatermark watermark,
+ @Nullable SqlCharStringLiteral comment,
+ SqlNode asQuery,
+ boolean isTemporary,
+ boolean ifNotExists) {
+ super(
+ OPERATOR,
+ pos,
+ tableName,
+ columnList,
+ tableConstraints,
+ propertyList,
+ partitionKeyList,
+ watermark,
+ comment,
+ isTemporary,
+ ifNotExists);
+ this.asQuery = asQuery;
+ }
+
+ @Override
+ public @Nonnull SqlOperator getOperator() {
+ return OPERATOR;
+ }
+
+ @Override
+ public @Nonnull List<SqlNode> getOperandList() {
+ return ImmutableNullableList.<SqlNode>builder()
+ .addAll(super.getOperandList())
+ .add(asQuery)
+ .build();
+ }
+
+ @Override
+ public void validate() throws SqlValidateException {
+ super.validate();
+ if (isTemporary()) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not support to create temporary table yet.");
+ }
+
+ if (getColumnList().size() > 0) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not support explicitly specifying columns yet.");
+ }
+
+ if (getWatermark().isPresent()) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not support explicitly specifying watermark yet.");
Review Comment:
```suggestion
"CREATE TABLE AS SELECT syntax does not support to specify explicit watermark yet.");
```
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+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 org.apache.calcite.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+/**
+ * {@link SqlNode} to describe the CREATE TABLE AS syntax. The CTAS syntax is different from CREATE
+ * TABLE syntax, which also create the pipeline to sync the data from the source to the derived
+ * table.
+ *
+ * <p>Example: A DDL like the one below for creating a `derived_table`
+ *
+ * <pre>{@code
+ * CREATE TABLE base_table (
+ * id BIGINT,
+ * name STRING,
+ * tstmp TIMESTAMP,
+ * PRIMARY KEY(id)
+ * ) WITH (
+ * ‘connector’ = ‘kafka’,
+ * ‘connector.starting-offset’: ‘12345’,
+ * ‘format’ = ‘json’
+ * )
+ *
+ * CREATE TABLE derived_table
+ * WITH (
+ * 'connector' = 'jdbc',
+ * 'url' = 'http://localhost:10000',
+ * 'table-name' = 'syncedTable'
+ * )
+ * AS SELECT * FROM base_table_1;
+ * }</pre>
+ */
+public class SqlCreateTableAs extends SqlCreateTable {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS", SqlKind.CREATE_TABLE);
+
+ private final SqlNode asQuery;
+
+ public SqlCreateTableAs(
+ SqlParserPos pos,
+ SqlIdentifier tableName,
+ SqlNodeList columnList,
+ List<SqlTableConstraint> tableConstraints,
+ SqlNodeList propertyList,
+ SqlNodeList partitionKeyList,
+ @Nullable SqlWatermark watermark,
+ @Nullable SqlCharStringLiteral comment,
+ SqlNode asQuery,
+ boolean isTemporary,
+ boolean ifNotExists) {
+ super(
+ OPERATOR,
+ pos,
+ tableName,
+ columnList,
+ tableConstraints,
+ propertyList,
+ partitionKeyList,
+ watermark,
+ comment,
+ isTemporary,
+ ifNotExists);
+ this.asQuery = asQuery;
+ }
+
+ @Override
+ public @Nonnull SqlOperator getOperator() {
+ return OPERATOR;
+ }
+
+ @Override
+ public @Nonnull List<SqlNode> getOperandList() {
+ return ImmutableNullableList.<SqlNode>builder()
+ .addAll(super.getOperandList())
+ .add(asQuery)
+ .build();
+ }
+
+ @Override
+ public void validate() throws SqlValidateException {
+ super.validate();
+ if (isTemporary()) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not support to create temporary table yet.");
+ }
+
+ if (getColumnList().size() > 0) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not support explicitly specifying columns yet.");
+ }
+
+ if (getWatermark().isPresent()) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not support explicitly specifying watermark yet.");
+ }
+ // TODO flink dialect supports dynamic partition
+ if (getPartitionKeyList().size() > 0) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not support creating partitioned table yet.");
Review Comment:
```suggestion
"CREATE TABLE AS SELECT syntax does not support to create partitioned table yet.");
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] beyond1920 commented on a diff in pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
beyond1920 commented on code in PR #20252:
URL: https://github.com/apache/flink/pull/20252#discussion_r925746124
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ExtendedSqlNode;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCreate;
+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.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** CREATE TABLE AS SELECT DDL sql call. */
+public class SqlCreateTableAs extends SqlCreate implements ExtendedSqlNode {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS SELECT", SqlKind.OTHER_DDL);
+
+ private final SqlCreateTable sqlCreateTable;
+ private final SqlNode asQuery;
+
+ public SqlCreateTableAs(SqlCreateTable sqlCreateTable, SqlNode asQuery) {
+ super(
+ OPERATOR,
+ sqlCreateTable.getParserPosition(),
+ sqlCreateTable.getReplace(),
+ sqlCreateTable.isIfNotExists());
+ this.sqlCreateTable = sqlCreateTable;
Review Comment:
It's better to check the input sqlCreateTable does not have tableLike.
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ExtendedSqlNode;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCreate;
+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.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** CREATE TABLE AS SELECT DDL sql call. */
+public class SqlCreateTableAs extends SqlCreate implements ExtendedSqlNode {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS SELECT", SqlKind.OTHER_DDL);
Review Comment:
Why use `OTHER_DDL` instead of `CREATE_TABLE`?
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ExtendedSqlNode;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCreate;
+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.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** CREATE TABLE AS SELECT DDL sql call. */
+public class SqlCreateTableAs extends SqlCreate implements ExtendedSqlNode {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS SELECT", SqlKind.OTHER_DDL);
+
+ private final SqlCreateTable sqlCreateTable;
+ private final SqlNode asQuery;
+
+ public SqlCreateTableAs(SqlCreateTable sqlCreateTable, SqlNode asQuery) {
+ super(
+ OPERATOR,
+ sqlCreateTable.getParserPosition(),
+ sqlCreateTable.getReplace(),
+ sqlCreateTable.isIfNotExists());
+ this.sqlCreateTable = sqlCreateTable;
+ this.asQuery = asQuery;
+ }
+
+ @Override
+ public @Nonnull SqlOperator getOperator() {
+ return OPERATOR;
+ }
+
+ @Override
+ public @Nonnull List<SqlNode> getOperandList() {
+ SqlNode[] sqlNodes = new SqlNode[sqlCreateTable.getOperandList().size() + 1];
+ List<SqlNode> operandList = sqlCreateTable.getOperandList();
+ for (int i = 0; i < operandList.size(); i++) {
+ sqlNodes[i] = operandList.get(i);
+ }
+ sqlNodes[sqlNodes.length - 1] = asQuery;
+ return ImmutableNullableList.copyOf(sqlNodes);
+ }
+
+ @Override
+ public void validate() throws SqlValidateException {
+ sqlCreateTable.validate();
+ if (sqlCreateTable.isTemporary()) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not yet support creating temporary table.");
+ }
+
+ if (sqlCreateTable.getColumnList().size() > 0) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not yet support explicitly specifying columns.");
+ }
+
+ if (sqlCreateTable.getWatermark().isPresent()) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not yet support explicitly specifying watermark.");
+ }
+ // TODO flink dialect supports dynamic partition
+ if (sqlCreateTable.getPartitionKeyList().size() > 0) {
+ throw new SqlValidateException(
Review Comment:
Ditto
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ExtendedSqlNode;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCreate;
+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.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** CREATE TABLE AS SELECT DDL sql call. */
+public class SqlCreateTableAs extends SqlCreate implements ExtendedSqlNode {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS SELECT", SqlKind.OTHER_DDL);
+
+ private final SqlCreateTable sqlCreateTable;
+ private final SqlNode asQuery;
+
+ public SqlCreateTableAs(SqlCreateTable sqlCreateTable, SqlNode asQuery) {
+ super(
+ OPERATOR,
+ sqlCreateTable.getParserPosition(),
+ sqlCreateTable.getReplace(),
+ sqlCreateTable.isIfNotExists());
+ this.sqlCreateTable = sqlCreateTable;
+ this.asQuery = asQuery;
+ }
+
+ @Override
+ public @Nonnull SqlOperator getOperator() {
+ return OPERATOR;
+ }
+
+ @Override
+ public @Nonnull List<SqlNode> getOperandList() {
+ SqlNode[] sqlNodes = new SqlNode[sqlCreateTable.getOperandList().size() + 1];
+ List<SqlNode> operandList = sqlCreateTable.getOperandList();
+ for (int i = 0; i < operandList.size(); i++) {
+ sqlNodes[i] = operandList.get(i);
+ }
+ sqlNodes[sqlNodes.length - 1] = asQuery;
+ return ImmutableNullableList.copyOf(sqlNodes);
+ }
+
+ @Override
+ public void validate() throws SqlValidateException {
+ sqlCreateTable.validate();
+ if (sqlCreateTable.isTemporary()) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not yet support creating temporary table.");
+ }
+
+ if (sqlCreateTable.getColumnList().size() > 0) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not yet support explicitly specifying columns.");
Review Comment:
Ditto
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ExtendedSqlNode;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCreate;
+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.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** CREATE TABLE AS SELECT DDL sql call. */
+public class SqlCreateTableAs extends SqlCreate implements ExtendedSqlNode {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS SELECT", SqlKind.OTHER_DDL);
+
+ private final SqlCreateTable sqlCreateTable;
+ private final SqlNode asQuery;
+
+ public SqlCreateTableAs(SqlCreateTable sqlCreateTable, SqlNode asQuery) {
+ super(
+ OPERATOR,
+ sqlCreateTable.getParserPosition(),
+ sqlCreateTable.getReplace(),
+ sqlCreateTable.isIfNotExists());
+ this.sqlCreateTable = sqlCreateTable;
+ this.asQuery = asQuery;
+ }
+
+ @Override
+ public @Nonnull SqlOperator getOperator() {
+ return OPERATOR;
+ }
+
+ @Override
+ public @Nonnull List<SqlNode> getOperandList() {
+ SqlNode[] sqlNodes = new SqlNode[sqlCreateTable.getOperandList().size() + 1];
+ List<SqlNode> operandList = sqlCreateTable.getOperandList();
+ for (int i = 0; i < operandList.size(); i++) {
+ sqlNodes[i] = operandList.get(i);
+ }
+ sqlNodes[sqlNodes.length - 1] = asQuery;
+ return ImmutableNullableList.copyOf(sqlNodes);
+ }
+
+ @Override
+ public void validate() throws SqlValidateException {
+ sqlCreateTable.validate();
+ if (sqlCreateTable.isTemporary()) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not yet support creating temporary table.");
+ }
+
+ if (sqlCreateTable.getColumnList().size() > 0) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not yet support explicitly specifying columns.");
+ }
+
+ if (sqlCreateTable.getWatermark().isPresent()) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not yet support explicitly specifying watermark.");
Review Comment:
Ditto
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ExtendedSqlNode;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCreate;
+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.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** CREATE TABLE AS SELECT DDL sql call. */
+public class SqlCreateTableAs extends SqlCreate implements ExtendedSqlNode {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS SELECT", SqlKind.OTHER_DDL);
+
+ private final SqlCreateTable sqlCreateTable;
+ private final SqlNode asQuery;
+
+ public SqlCreateTableAs(SqlCreateTable sqlCreateTable, SqlNode asQuery) {
+ super(
+ OPERATOR,
+ sqlCreateTable.getParserPosition(),
+ sqlCreateTable.getReplace(),
+ sqlCreateTable.isIfNotExists());
+ this.sqlCreateTable = sqlCreateTable;
+ this.asQuery = asQuery;
+ }
+
+ @Override
+ public @Nonnull SqlOperator getOperator() {
+ return OPERATOR;
+ }
+
+ @Override
+ public @Nonnull List<SqlNode> getOperandList() {
+ SqlNode[] sqlNodes = new SqlNode[sqlCreateTable.getOperandList().size() + 1];
Review Comment:
```suggestion
return ImmutableNullableList.<SqlNode>builder() .addAll(sqlCreateTable.getOperandList()) .add(asQuery) .build();
```
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ExtendedSqlNode;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCreate;
+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.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** CREATE TABLE AS SELECT DDL sql call. */
+public class SqlCreateTableAs extends SqlCreate implements ExtendedSqlNode {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS SELECT", SqlKind.OTHER_DDL);
+
+ private final SqlCreateTable sqlCreateTable;
+ private final SqlNode asQuery;
+
+ public SqlCreateTableAs(SqlCreateTable sqlCreateTable, SqlNode asQuery) {
+ super(
+ OPERATOR,
+ sqlCreateTable.getParserPosition(),
+ sqlCreateTable.getReplace(),
+ sqlCreateTable.isIfNotExists());
+ this.sqlCreateTable = sqlCreateTable;
+ this.asQuery = asQuery;
+ }
+
+ @Override
+ public @Nonnull SqlOperator getOperator() {
+ return OPERATOR;
+ }
+
+ @Override
+ public @Nonnull List<SqlNode> getOperandList() {
+ SqlNode[] sqlNodes = new SqlNode[sqlCreateTable.getOperandList().size() + 1];
+ List<SqlNode> operandList = sqlCreateTable.getOperandList();
+ for (int i = 0; i < operandList.size(); i++) {
+ sqlNodes[i] = operandList.get(i);
+ }
+ sqlNodes[sqlNodes.length - 1] = asQuery;
+ return ImmutableNullableList.copyOf(sqlNodes);
+ }
+
+ @Override
+ public void validate() throws SqlValidateException {
+ sqlCreateTable.validate();
+ if (sqlCreateTable.isTemporary()) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not yet support creating temporary table.");
+ }
+
+ if (sqlCreateTable.getColumnList().size() > 0) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not yet support explicitly specifying columns.");
+ }
+
+ if (sqlCreateTable.getWatermark().isPresent()) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not yet support explicitly specifying watermark.");
+ }
+ // TODO flink dialect supports dynamic partition
+ if (sqlCreateTable.getPartitionKeyList().size() > 0) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not yet support creating partitioned table.");
+ }
+ List<SqlTableConstraint> constraints =
+ sqlCreateTable.getFullConstraints().stream()
+ .filter(SqlTableConstraint::isPrimaryKey)
+ .collect(Collectors.toList());
+ if (!constraints.isEmpty()) {
Review Comment:
```suggestion
if (sqlCreateTable.getFullConstraints().stream().anyMatch(SqlTableConstraint::isPrimaryKey)) {
...}
```
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ExtendedSqlNode;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCreate;
+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.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** CREATE TABLE AS SELECT DDL sql call. */
+public class SqlCreateTableAs extends SqlCreate implements ExtendedSqlNode {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS SELECT", SqlKind.OTHER_DDL);
+
+ private final SqlCreateTable sqlCreateTable;
+ private final SqlNode asQuery;
+
+ public SqlCreateTableAs(SqlCreateTable sqlCreateTable, SqlNode asQuery) {
+ super(
+ OPERATOR,
+ sqlCreateTable.getParserPosition(),
+ sqlCreateTable.getReplace(),
+ sqlCreateTable.isIfNotExists());
+ this.sqlCreateTable = sqlCreateTable;
+ this.asQuery = asQuery;
+ }
+
+ @Override
+ public @Nonnull SqlOperator getOperator() {
+ return OPERATOR;
+ }
+
+ @Override
+ public @Nonnull List<SqlNode> getOperandList() {
+ SqlNode[] sqlNodes = new SqlNode[sqlCreateTable.getOperandList().size() + 1];
+ List<SqlNode> operandList = sqlCreateTable.getOperandList();
+ for (int i = 0; i < operandList.size(); i++) {
+ sqlNodes[i] = operandList.get(i);
+ }
+ sqlNodes[sqlNodes.length - 1] = asQuery;
+ return ImmutableNullableList.copyOf(sqlNodes);
+ }
+
+ @Override
+ public void validate() throws SqlValidateException {
+ sqlCreateTable.validate();
+ if (sqlCreateTable.isTemporary()) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not yet support creating temporary table.");
Review Comment:
```suggestion
"CREATE TABLE AS SELECT syntax does not support creating temporary table yet."
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] beyond1920 commented on a diff in pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
beyond1920 commented on code in PR #20252:
URL: https://github.com/apache/flink/pull/20252#discussion_r929938421
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableLike.java:
##########
@@ -40,10 +40,10 @@
import static java.util.Objects.requireNonNull;
/**
- * {@link SqlNode} to describe the CREATE TABLE LIKE syntax. CREATE TABLE LIKE syntax is the same as
- * CREATE TABLE syntax, both are just metadata operations.
+ * {@link SqlNode} to describe the CREATE TABLE LIKE syntax. CREATE TABLE LIKE syntax is similar as
+ * CREATE TABLE syntax, besides it has LIKE sub-clause to inherits property of an existed table.
Review Comment:
to inherit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on a diff in pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
Tartarus0zm commented on code in PR #20252:
URL: https://github.com/apache/flink/pull/20252#discussion_r925577545
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTable.java:
##########
@@ -189,6 +198,15 @@ public void validate() throws SqlValidateException {
if (tableLike != null) {
tableLike.validate();
}
+
+ if (query != null
+ && (columnList.size() > 0
+ || constraints.size() > 0
Review Comment:
flink dialect does not supports dynamic partition, so CTAS also does not support Partition Key
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] beyond1920 commented on a diff in pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
beyond1920 commented on code in PR #20252:
URL: https://github.com/apache/flink/pull/20252#discussion_r926518137
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ExtendedSqlNode;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCreate;
+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.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** CREATE TABLE AS SELECT DDL sql call. */
+public class SqlCreateTableAs extends SqlCreate implements ExtendedSqlNode {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS SELECT", SqlKind.OTHER_DDL);
+
+ private final SqlCreateTable sqlCreateTable;
+ private final SqlNode asQuery;
+
+ public SqlCreateTableAs(SqlCreateTable sqlCreateTable, SqlNode asQuery) {
+ super(
+ OPERATOR,
+ sqlCreateTable.getParserPosition(),
+ sqlCreateTable.getReplace(),
+ sqlCreateTable.isIfNotExists());
+ this.sqlCreateTable = sqlCreateTable;
Review Comment:
Since Like and AS are all optional clauses for a create table query, it's wierd that `LIKE` is a property of `SqlCreateTable`, but `AS` is a property of subclass.
It's better to introduce `SqlCreateTable` as base class. And introduce `SqlCreateTableLike` and `SqlCreateTableAs` to handle `LIKE` and `AS` clauses.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] beyond1920 commented on a diff in pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
beyond1920 commented on code in PR #20252:
URL: https://github.com/apache/flink/pull/20252#discussion_r929887645
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableLike.java:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+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 org.apache.calcite.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * {@link SqlNode} to describe the CREATE TABLE LIKE syntax. CREATE TABLE LIKE syntax is the same as
+ * CREATE TABLE syntax, both are just metadata operations.
Review Comment:
CREATE TABLE LIKE syntax is similar as CREATE TABLE syntax, besides it has LIKE sub clause to inherits property of an existed table.
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableLike.java:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+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 org.apache.calcite.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * {@link SqlNode} to describe the CREATE TABLE LIKE syntax. CREATE TABLE LIKE syntax is the same as
+ * CREATE TABLE syntax, both are just metadata operations.
+ *
+ * <p>Example: A DDL like the one below for creating a `derived_table`
Review Comment:
Example:
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTable.java:
##########
@@ -85,7 +83,6 @@ public SqlCreateTable(
SqlNodeList partitionKeyList,
@Nullable SqlWatermark watermark,
@Nullable SqlCharStringLiteral comment,
- @Nullable SqlTableLike tableLike,
boolean isTemporary,
boolean ifNotExists) {
super(OPERATOR, pos, false, ifNotExists);
Review Comment:
how about call the other constructor to create a new instance?
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+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 org.apache.calcite.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * {@link SqlNode} to describe the CREATE TABLE AS syntax. The CTAS syntax is different from CREATE
+ * TABLE syntax, which also create the pipeline to sync the data from the source to the derived
+ * table.
+ *
+ * <p>Example: A DDL like the one below for creating a `derived_table`
+ *
+ * <pre>{@code
+ * CREATE TABLE base_table (
+ * id BIGINT,
+ * name STRING,
+ * tstmp TIMESTAMP,
+ * PRIMARY KEY(id)
+ * ) WITH (
+ * ‘connector’ = ‘kafka’,
+ * ‘connector.starting-offset’: ‘12345’,
+ * ‘format’ = ‘json’
+ * )
+ *
+ * CREATE TABLE derived_table
+ * WITH (
+ * 'connector' = 'jdbc',
+ * 'url' = 'http://localhost:10000',
+ * 'table-name' = 'syncedTable'
+ * )
+ * AS SELECT * FROM base_table;
+ * }</pre>
+ */
+public class SqlCreateTableAs extends SqlCreateTable {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS", SqlKind.CREATE_TABLE);
+
+ private final SqlNode asQuery;
+
+ public SqlCreateTableAs(
+ SqlParserPos pos,
+ SqlIdentifier tableName,
+ SqlNodeList columnList,
+ List<SqlTableConstraint> tableConstraints,
+ SqlNodeList propertyList,
+ SqlNodeList partitionKeyList,
+ @Nullable SqlWatermark watermark,
+ @Nullable SqlCharStringLiteral comment,
+ SqlNode asQuery,
+ boolean isTemporary,
+ boolean ifNotExists) {
+ super(
+ OPERATOR,
+ pos,
+ tableName,
+ columnList,
+ tableConstraints,
+ propertyList,
+ partitionKeyList,
+ watermark,
+ comment,
+ isTemporary,
+ ifNotExists);
+ this.asQuery = requireNonNull(asQuery, "asQuery should not be null");
Review Comment:
```suggestion
this.asQuery = requireNonNull(asQuery, "As clause is required for CREATE TABLE AS SELECT DDL");
```
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableLike.java:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+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 org.apache.calcite.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * {@link SqlNode} to describe the CREATE TABLE LIKE syntax. CREATE TABLE LIKE syntax is the same as
+ * CREATE TABLE syntax, both are just metadata operations.
+ *
+ * <p>Example: A DDL like the one below for creating a `derived_table`
+ *
+ * <pre>{@code
+ * CREATE TABLE base_table (
+ * id BIGINT,
+ * name STRING,
+ * tstmp TIMESTAMP,
+ * PRIMARY KEY(id)
+ * ) WITH (
+ * ‘connector’ = ‘kafka’,
+ * ‘connector.starting-offset’: ‘12345’,
+ * ‘format’ = ‘json’
+ * )
+ *
+ * CREATE TABLE derived_table (
+ * a int
+ * )
+ * WITH (
+ * 'connector' = 'jdbc',
+ * 'url' = 'http://localhost:10000',
+ * 'table-name' = 'derivedTable'
+ * )
+ * LIKE base_table;
+ * }</pre>
+ */
+public class SqlCreateTableLike extends SqlCreateTable {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE LIKE", SqlKind.CREATE_TABLE);
+
+ private final SqlTableLike tableLike;
+
+ public SqlCreateTableLike(
+ SqlParserPos pos,
+ SqlIdentifier tableName,
+ SqlNodeList columnList,
+ List<SqlTableConstraint> tableConstraints,
+ SqlNodeList propertyList,
+ SqlNodeList partitionKeyList,
+ @Nullable SqlWatermark watermark,
+ @Nullable SqlCharStringLiteral comment,
+ SqlTableLike tableLike,
+ boolean isTemporary,
+ boolean ifNotExists) {
+ super(
+ OPERATOR,
+ pos,
+ tableName,
+ columnList,
+ tableConstraints,
+ propertyList,
+ partitionKeyList,
+ watermark,
+ comment,
+ isTemporary,
+ ifNotExists);
+ this.tableLike = requireNonNull(tableLike, "tableLike should not be null");
Review Comment:
```suggestion
this.tableLike = requireNonNull(tableLike, "LIKE clause is required for CREATE TABLE LIKE DDL");
```
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+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 org.apache.calcite.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * {@link SqlNode} to describe the CREATE TABLE AS syntax. The CTAS syntax is different from CREATE
+ * TABLE syntax, which also create the pipeline to sync the data from the source to the derived
+ * table.
+ *
+ * <p>Example: A DDL like the one below for creating a `derived_table`
Review Comment:
```suggestion
* <p>Example:
```
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+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 org.apache.calcite.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * {@link SqlNode} to describe the CREATE TABLE AS syntax. The CTAS syntax is different from CREATE
+ * TABLE syntax, which also create the pipeline to sync the data from the source to the derived
+ * table.
Review Comment:
```suggestion
* . The CTAS would create a pipeline to compute the result of the given query and insert data into the derived table.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lsyldliu commented on a diff in pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
lsyldliu commented on code in PR #20252:
URL: https://github.com/apache/flink/pull/20252#discussion_r927222651
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ExtendedSqlNode;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCreate;
+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.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** CREATE TABLE AS SELECT DDL sql call. */
+public class SqlCreateTableAs extends SqlCreate implements ExtendedSqlNode {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS SELECT", SqlKind.OTHER_DDL);
+
+ private final SqlCreateTable sqlCreateTable;
+ private final SqlNode asQuery;
+
+ public SqlCreateTableAs(SqlCreateTable sqlCreateTable, SqlNode asQuery) {
+ super(
+ OPERATOR,
+ sqlCreateTable.getParserPosition(),
+ sqlCreateTable.getReplace(),
+ sqlCreateTable.isIfNotExists());
+ this.sqlCreateTable = sqlCreateTable;
+ this.asQuery = asQuery;
+ }
+
+ @Override
+ public @Nonnull SqlOperator getOperator() {
+ return OPERATOR;
+ }
+
+ @Override
+ public @Nonnull List<SqlNode> getOperandList() {
Review Comment:
The following code is enough:
`ImmutableNullableList.of(createTable, asQuery);`
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ExtendedSqlNode;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCreate;
+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.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** CREATE TABLE AS SELECT DDL sql call. */
Review Comment:
add more annotation
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ExtendedSqlNode;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCreate;
+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.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** CREATE TABLE AS SELECT DDL sql call. */
+public class SqlCreateTableAs extends SqlCreate implements ExtendedSqlNode {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS SELECT", SqlKind.OTHER_DDL);
+
+ private final SqlCreateTable sqlCreateTable;
+ private final SqlNode asQuery;
+
+ public SqlCreateTableAs(SqlCreateTable sqlCreateTable, SqlNode asQuery) {
+ super(
+ OPERATOR,
+ sqlCreateTable.getParserPosition(),
+ sqlCreateTable.getReplace(),
+ sqlCreateTable.isIfNotExists());
+ this.sqlCreateTable = sqlCreateTable;
Review Comment:
I think maybe we can try to refactor this, it seems to introduce `SqlCreateTableLike` and `SqlCreateTableAs ` as subclass more clear, but maybe it is a huge work.
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ExtendedSqlNode;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCreate;
+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.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** CREATE TABLE AS SELECT DDL sql call. */
Review Comment:
```suggestion
/**
* SqlNode to describe the CREATE TABLE AS syntax. The CTAS syntax is different from CREATE TABLE
* syntax, which also create the pipeline to sync the data from the source to the derived table.
*
* <p>Example: A DDL like the one below for creating a `derived_table`
*
* <pre>{@code
* CREATE TABLE base_table (
* id BIGINT,
* name STRING,
* tstmp TIMESTAMP,
* PRIMARY KEY(id)
* ) WITH (
* ‘connector’ = ‘kafka’,
* ‘connector.starting-offset’: ‘12345’,
* ‘format’ = ‘json’
* )
*
* CREATE TABLE derived_table
* WITH (
* 'connector' = 'jdbc',
* 'url' = 'http://localhost:10000',
* 'table-name' = 'syncedTable'
* )
* AS SELECT * FROM base_table_1;
* }</pre>
*/```
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ExtendedSqlNode;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCreate;
+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.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** CREATE TABLE AS SELECT DDL sql call. */
+public class SqlCreateTableAs extends SqlCreate implements ExtendedSqlNode {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS SELECT", SqlKind.OTHER_DDL);
+
+ private final SqlCreateTable sqlCreateTable;
+ private final SqlNode asQuery;
+
+ public SqlCreateTableAs(SqlCreateTable sqlCreateTable, SqlNode asQuery) {
+ super(
+ OPERATOR,
+ sqlCreateTable.getParserPosition(),
+ sqlCreateTable.getReplace(),
+ sqlCreateTable.isIfNotExists());
+ this.sqlCreateTable = sqlCreateTable;
+ this.asQuery = asQuery;
+ }
+
+ @Override
+ public @Nonnull SqlOperator getOperator() {
+ return OPERATOR;
+ }
+
+ @Override
+ public @Nonnull List<SqlNode> getOperandList() {
+ SqlNode[] sqlNodes = new SqlNode[sqlCreateTable.getOperandList().size() + 1];
+ List<SqlNode> operandList = sqlCreateTable.getOperandList();
+ for (int i = 0; i < operandList.size(); i++) {
+ sqlNodes[i] = operandList.get(i);
+ }
+ sqlNodes[sqlNodes.length - 1] = asQuery;
+ return ImmutableNullableList.copyOf(sqlNodes);
+ }
+
+ @Override
+ public void validate() throws SqlValidateException {
+ sqlCreateTable.validate();
+ if (sqlCreateTable.isTemporary()) {
+ throw new SqlValidateException(
+ getParserPosition(),
+ "CREATE TABLE AS SELECT syntax does not yet support creating temporary table.");
Review Comment:
support to create or support creating?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wuchong merged pull request #20252: [FLINK-28463][sql-parser] Supports CREATE TABLE AS SELECT syntax
Posted by GitBox <gi...@apache.org>.
wuchong merged PR #20252:
URL: https://github.com/apache/flink/pull/20252
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lsyldliu commented on a diff in pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
lsyldliu commented on code in PR #20252:
URL: https://github.com/apache/flink/pull/20252#discussion_r922646290
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTable.java:
##########
@@ -76,6 +76,8 @@ public class SqlCreateTable extends SqlCreate implements ExtendedSqlNode {
private final boolean isTemporary;
+ private final SqlNode query;
Review Comment:
What do you think about introduce a new SqlNode `SqlCreateTableAs` which extends the `SqlCreate` and implement `ExtendedSqlNode`? In this node, we refer to the `SqlCreateTable` and `asQuery` as member, so we can separate the two different function node, it seems that more clear.
Moreover, maybe we need hint in ctas syntax, such as `CREATE TABLE ... AS SELECT ... /*+ `OPTIONS`('key' = 'val') */`
##########
flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/CreateTableAsSelectTest.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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;
+
+import org.apache.flink.sql.parser.ddl.SqlCreateTable;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+import org.apache.flink.sql.parser.impl.FlinkSqlParserImpl;
+
+import org.apache.calcite.config.Lex;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.parser.SqlParseException;
+import org.apache.calcite.sql.parser.SqlParser;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests for parsing and validating CREATE TABLE AS SELECT(CTAS) clause in {@link SqlCreateTable}.
+ */
+class CreateTableAsSelectTest {
+
+ @Test
+ void testNoOptions() throws Exception {
+ SqlNode actualNode = createFlinkParser("CREATE TABLE t AS SELECT * FROM b").parseStmt();
+
+ assertThat(actualNode.toString()).isEqualTo("CREATE TABLE `t`\nAS\nSELECT *\nFROM `b`");
+ }
+
+ @Test
+ void testWithOptions() throws Exception {
+ SqlNode actualNode =
+ createFlinkParser("CREATE TABLE t WITH ('test' = 'zm') AS SELECT * FROM b")
+ .parseStmt();
+
+ assertThat(actualNode.toString())
+ .isEqualTo("CREATE TABLE `t` WITH (\n 'test' = 'zm'\n)\nAS\nSELECT *\nFROM `b`");
+ }
+
+ @Test
+ void testWithColumns() throws Exception {
Review Comment:
Test case as following:
1. testCreateTableAsSelectWithExplicitColumns
2. testCreateTableAsSelectWithWatermark
3. testCreateTableAsSelectWithConstraints
4. testCreateTableAsSelectWithPartitionKey
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTable.java:
##########
@@ -189,6 +198,15 @@ public void validate() throws SqlValidateException {
if (tableLike != null) {
tableLike.validate();
}
+
+ if (query != null
+ && (columnList.size() > 0
Review Comment:
I think we should validate these in separated if branch. In each case, we should give a clear error message.
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTable.java:
##########
@@ -189,6 +198,15 @@ public void validate() throws SqlValidateException {
if (tableLike != null) {
tableLike.validate();
}
+
+ if (query != null
+ && (columnList.size() > 0
+ || constraints.size() > 0
Review Comment:
Here exists following case we don't support:
1. the target table is temporary table
2. Specify the columns explicitly
3. Specify the watermark explicitly
4. Specify the constraint explicitly
Regarding to partition key, we think we should support it, see [spark](https://spark.apache.org/docs/latest/sql-ref-syntax-ddl-create-table-datasource.html) also support it.
##########
flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/CreateTableAsSelectTest.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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;
+
+import org.apache.flink.sql.parser.ddl.SqlCreateTable;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+import org.apache.flink.sql.parser.impl.FlinkSqlParserImpl;
+
+import org.apache.calcite.config.Lex;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.parser.SqlParseException;
+import org.apache.calcite.sql.parser.SqlParser;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests for parsing and validating CREATE TABLE AS SELECT(CTAS) clause in {@link SqlCreateTable}.
+ */
+class CreateTableAsSelectTest {
Review Comment:
I think we can move these tests to `FlinkSqlParserImplTest`, then we can reuse the code. BTW, to the illegal test, we can use the `sql(sql).fails`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
luoyuxia commented on code in PR #20252:
URL: https://github.com/apache/flink/pull/20252#discussion_r920919363
##########
flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl:
##########
@@ -1152,6 +1153,10 @@ SqlCreate SqlCreateTable(Span s, boolean replace, boolean isTemporary) :
<LIKE>
tableLike = SqlTableLike(getPos())
]
+ [
Review Comment:
As `LIKE` and `AS` won't appear at same time in our sytax, I think it should be somethink like `[ [LIKE] | [AS] ]`
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTable.java:
##########
@@ -186,9 +195,24 @@ public void validate() throws SqlValidateException {
}
}
+ if (query != null && tableLike != null) {
+ throw new SqlValidateException(
+ pos,
+ "CREATE TABLE AS SELECT and CREATE TABLE LIKE syntax cannot be used together.");
+ }
+
if (tableLike != null) {
tableLike.validate();
}
+
+ if (query != null
+ && (columnList.size() > 0
+ || constraints.size() > 0
+ || partitionKeyList.size() > 0)) {
+ throw new SqlValidateException(
+ pos,
+ "CREATE TABLE AS SELECT syntax does not yet support Column/Partition/Constraints settings.");
Review Comment:
may be some more clear message like
`CREATE TABLE AS SELECT syntax does not yet support to specific Column/Partition/Constraints`
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTable.java:
##########
@@ -302,6 +326,11 @@ public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
writer.newlineAndIndent();
this.tableLike.unparse(writer, leftPrec, rightPrec);
}
+
+ if (this.query != null) {
+ writer.newlineAndIndent();
+ this.query.unparse(writer, 0, 0);
Review Comment:
Why not follow the above code to pass `leftPrec` and `rightPrec` to the `unparse` method?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on a diff in pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
Tartarus0zm commented on code in PR #20252:
URL: https://github.com/apache/flink/pull/20252#discussion_r926251262
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ExtendedSqlNode;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCreate;
+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.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** CREATE TABLE AS SELECT DDL sql call. */
+public class SqlCreateTableAs extends SqlCreate implements ExtendedSqlNode {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS SELECT", SqlKind.OTHER_DDL);
+
+ private final SqlCreateTable sqlCreateTable;
+ private final SqlNode asQuery;
+
+ public SqlCreateTableAs(SqlCreateTable sqlCreateTable, SqlNode asQuery) {
+ super(
+ OPERATOR,
+ sqlCreateTable.getParserPosition(),
+ sqlCreateTable.getReplace(),
+ sqlCreateTable.isIfNotExists());
+ this.sqlCreateTable = sqlCreateTable;
Review Comment:
SQL syntax has defined that `AS` and `LIKE` can only appear one, theoretically `SqlCreateTableAs` will not appear `SqlCreateTable#tableLike` is not null.
see UT: `FlinkSqlParserImplTest#testCreateTableAsSelectWithCreateTableLike`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on a diff in pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
Tartarus0zm commented on code in PR #20252:
URL: https://github.com/apache/flink/pull/20252#discussion_r927546446
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ExtendedSqlNode;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCreate;
+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.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** CREATE TABLE AS SELECT DDL sql call. */
+public class SqlCreateTableAs extends SqlCreate implements ExtendedSqlNode {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS SELECT", SqlKind.OTHER_DDL);
+
+ private final SqlCreateTable sqlCreateTable;
+ private final SqlNode asQuery;
+
+ public SqlCreateTableAs(SqlCreateTable sqlCreateTable, SqlNode asQuery) {
+ super(
+ OPERATOR,
+ sqlCreateTable.getParserPosition(),
+ sqlCreateTable.getReplace(),
+ sqlCreateTable.isIfNotExists());
+ this.sqlCreateTable = sqlCreateTable;
+ this.asQuery = asQuery;
+ }
+
+ @Override
+ public @Nonnull SqlOperator getOperator() {
+ return OPERATOR;
+ }
+
+ @Override
+ public @Nonnull List<SqlNode> getOperandList() {
Review Comment:
introduce `SqlCreateTableLike` and `SqlCreateTableAs` as subclass,
so referring to the recommendations of @beyond1920
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on a diff in pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
Tartarus0zm commented on code in PR #20252:
URL: https://github.com/apache/flink/pull/20252#discussion_r927388393
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateTableAs.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.sql.parser.ExtendedSqlNode;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+import org.apache.flink.sql.parser.error.SqlValidateException;
+
+import org.apache.calcite.sql.SqlCreate;
+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.util.ImmutableNullableList;
+
+import javax.annotation.Nonnull;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** CREATE TABLE AS SELECT DDL sql call. */
+public class SqlCreateTableAs extends SqlCreate implements ExtendedSqlNode {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE TABLE AS SELECT", SqlKind.OTHER_DDL);
+
+ private final SqlCreateTable sqlCreateTable;
+ private final SqlNode asQuery;
+
+ public SqlCreateTableAs(SqlCreateTable sqlCreateTable, SqlNode asQuery) {
+ super(
+ OPERATOR,
+ sqlCreateTable.getParserPosition(),
+ sqlCreateTable.getReplace(),
+ sqlCreateTable.isIfNotExists());
+ this.sqlCreateTable = sqlCreateTable;
Review Comment:
@beyond1920 @lsyldliu thanks for your advice. I will try to introduce `SqlCreateTableLike` and `SqlCreateTableAs`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
Tartarus0zm commented on PR #20252:
URL: https://github.com/apache/flink/pull/20252#issuecomment-1192460451
@beyond1920 @lsyldliu Thank you for your advice, I have updated the pr, please take a look again!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
Tartarus0zm commented on PR #20252:
URL: https://github.com/apache/flink/pull/20252#issuecomment-1196203437
@flinkbo run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
Tartarus0zm commented on PR #20252:
URL: https://github.com/apache/flink/pull/20252#issuecomment-1196221001
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
Tartarus0zm commented on PR #20252:
URL: https://github.com/apache/flink/pull/20252#issuecomment-1195435889
@beyond1920 thanks for your review! I have updated this pr, please take a look again!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Tartarus0zm commented on pull request #20252: [FLINK-28463][flink-sql-parser] Flink dialect supports CREATE TABLE AS SELECT(CTAS) syntax
Posted by GitBox <gi...@apache.org>.
Tartarus0zm commented on PR #20252:
URL: https://github.com/apache/flink/pull/20252#issuecomment-1185134587
@luoyuxia thanks for your comments, I have addressed the comments left. please take a look again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org