You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "hackergin (via GitHub)" <gi...@apache.org> on 2024/04/24 08:25:27 UTC
[PR] [FLINK-35190][table] Support create materialized table syntax [flink]
hackergin opened a new pull request, #24707:
URL: https://github.com/apache/flink/pull/24707
## What is the purpose of the change
* Support create materialized table syntax*
## Brief change log
- *Add create materialized table syntax*
## Verifying this change
* Add some unit test in FlinkSqlParserImplTest for testing the new added syntax.
## 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? ( Will be documented in a separated pr)
--
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
Re: [PR] [FLINK-35190][table] Support create materialized table syntax [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #24707:
URL: https://github.com/apache/flink/pull/24707#issuecomment-2074388424
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d011acc1d2021ce713e08e066cef9927fece5922",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d011acc1d2021ce713e08e066cef9927fece5922",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d011acc1d2021ce713e08e066cef9927fece5922 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
Re: [PR] [FLINK-35190][table] Support create materialized table syntax [flink]
Posted by "hackergin (via GitHub)" <gi...@apache.org>.
hackergin commented on code in PR #24707:
URL: https://github.com/apache/flink/pull/24707#discussion_r1579759071
##########
flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableStatementParserTest.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Execution;
+
+import static org.junit.jupiter.api.parallel.ExecutionMode.CONCURRENT;
+
+/** Sql parser test for materialized related syntax. * */
+@Execution(CONCURRENT)
+public class MaterializedTableStatementParserTest extends TestFlinkSqlParserBase {
Review Comment:
Currently, there is no need to extend TestFlinkSqlParserBase. T his will be considered in FLINK-35230 later. https://issues.apache.org/jira/browse/FLINK-35230
--
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
Re: [PR] [FLINK-35190][table] Support create materialized table syntax [flink]
Posted by "lsyldliu (via GitHub)" <gi...@apache.org>.
lsyldliu commented on code in PR #24707:
URL: https://github.com/apache/flink/pull/24707#discussion_r1579421407
##########
flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/TestFlinkSqlParserBase.java:
##########
@@ -0,0 +1,47 @@
+package org.apache.flink.sql.parser;
+
+import org.apache.flink.sql.parser.impl.FlinkSqlParserImpl;
+
+import org.apache.calcite.sql.parser.SqlParserFixture;
+import org.apache.calcite.sql.parser.SqlParserTest;
+import org.junit.jupiter.api.extension.ConditionEvaluationResult;
+import org.junit.jupiter.api.extension.ExecutionCondition;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.ExtensionContext;
+
+import java.lang.reflect.Method;
+import java.util.Optional;
+
+/**
+ * Abstract base class for testing the Flink SQL parser implementation.
+ *
+ * <p>This class extends the {@link SqlParserTest} class and overrides its fixture method to use the
+ * {@link FlinkSqlParserImpl} factory for parsing SQL statements. It also includes an extension to
+ * disable all inherited test methods from {@link SqlParserTest} and only execute specific tests
+ * defined in this class and its subclass.
+ */
+@ExtendWith(TestFlinkSqlParserBase.DisableSqlParserTestExtension.class)
+public abstract class TestFlinkSqlParserBase extends SqlParserTest {
Review Comment:
It would be better to rename to FlinkSqlParserTestBase
##########
flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableStatementParserTest.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Execution;
+
+import static org.junit.jupiter.api.parallel.ExecutionMode.CONCURRENT;
+
+/** Sql parser test for materialized related syntax. * */
+@Execution(CONCURRENT)
+public class MaterializedTableStatementParserTest extends TestFlinkSqlParserBase {
Review Comment:
BTW, FlinkSqlParserImplTest doesn't need to extends TestFlinkSqlParserBase?
##########
flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl:
##########
@@ -2189,6 +2267,8 @@ SqlCreate SqlCreateExtended(Span s, boolean replace) :
|
create = SqlCreateTable(s, replace, isTemporary)
|
+ create = SqlCreateMaterializedTable(s)
Review Comment:
What is the behavior when create temporary materialized 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
Re: [PR] [FLINK-35190][table] Support create materialized table syntax [flink]
Posted by "lsyldliu (via GitHub)" <gi...@apache.org>.
lsyldliu closed pull request #24707: [FLINK-35190][table] Support create materialized table syntax
URL: https://github.com/apache/flink/pull/24707
--
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
Re: [PR] [FLINK-35190][table] Support create materialized table syntax [flink]
Posted by "hackergin (via GitHub)" <gi...@apache.org>.
hackergin commented on code in PR #24707:
URL: https://github.com/apache/flink/pull/24707#discussion_r1579756740
##########
flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl:
##########
@@ -2189,6 +2267,8 @@ SqlCreate SqlCreateExtended(Span s, boolean replace) :
|
create = SqlCreateTable(s, replace, isTemporary)
|
+ create = SqlCreateMaterializedTable(s)
Review Comment:
Updated, an unsupported exception will be thrown when temporary materialized appears.
--
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
Re: [PR] [FLINK-35190][table] Support create materialized table syntax [flink]
Posted by "lsyldliu (via GitHub)" <gi...@apache.org>.
lsyldliu commented on code in PR #24707:
URL: https://github.com/apache/flink/pull/24707#discussion_r1578771973
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateMaterializedTable.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * 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.SqlUnparseUtils;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+
+import org.apache.calcite.sql.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlCreate;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlIntervalLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlLiteral;
+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.Nullable;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import static java.util.Objects.requireNonNull;
+
+/** CREATE MATERIALIZED TABLE DDL sql call. */
+public class SqlCreateMaterializedTable extends SqlCreate {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE MATERIALIZED TABLE", SqlKind.CREATE_TABLE);
+
+ private final SqlIdentifier tableName;
+
+ private final SqlCharStringLiteral comment;
+
+ private final SqlTableConstraint tableConstraint;
+
+ private final SqlNodeList partitionKeyList;
+
+ private final SqlNodeList propertyList;
+
+ // SqlIntervalLiteral
Review Comment:
Remove this comment, it makes no sense.
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlCreateMaterializedTable.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * 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.SqlUnparseUtils;
+import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
+
+import org.apache.calcite.sql.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlCreate;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlIntervalLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlLiteral;
+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.Nullable;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import static java.util.Objects.requireNonNull;
+
+/** CREATE MATERIALIZED TABLE DDL sql call. */
+public class SqlCreateMaterializedTable extends SqlCreate {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("CREATE MATERIALIZED TABLE", SqlKind.CREATE_TABLE);
+
+ private final SqlIdentifier tableName;
+
+ private final SqlCharStringLiteral comment;
+
+ private final SqlTableConstraint tableConstraint;
+
+ private final SqlNodeList partitionKeyList;
+
+ private final SqlNodeList propertyList;
+
+ // SqlIntervalLiteral
Review Comment:
Remove this comment, it makes no sense.
##########
flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd:
##########
@@ -193,6 +198,8 @@
"ANALYZE"
"COMPUTE"
"STATISTICS"
+ "REFRESH_MODE"
Review Comment:
Please place them in the right place in alphabetical order.
##########
flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableDDLParserTest.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Execution;
+
+import static org.junit.jupiter.api.parallel.ExecutionMode.CONCURRENT;
+
+/** Sql parser test for materialized related syntax. * */
+@Execution(CONCURRENT)
+public class MaterializedTableDDLParserTest extends FlinkSqlParserImplTest {
+
+ @Test
+ void testCreateMaterializedTable() {
+ final String sql =
+ "CREATE MATERIALIZED TABLE tbl1\n"
+ + "(\n"
+ + " PRIMARY KEY (a, b)\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "PARTITIONED BY (a, h)\n"
+ + "WITH (\n"
+ + " 'group.id' = 'latest', \n"
+ + " 'kafka.topic' = 'log.test'\n"
+ + ")\n"
+ + "FRESHNESS = INTERVAL '3' MINUTE\n"
+ + "AS SELECT a, b, h, t m FROM source";
+ final String expected =
+ "CREATE MATERIALIZED TABLE `TBL1`\n"
+ + "(\n"
+ + " PRIMARY KEY (`A`, `B`)\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "PARTITIONED BY (`A`, `H`)\n"
+ + "WITH (\n"
+ + " 'group.id' = 'latest',\n"
+ + " 'kafka.topic' = 'log.test'\n"
+ + ")\n"
+ + "FRESHNESS = INTERVAL '3' MINUTE\n"
+ + "AS\n"
+ + "SELECT `A`, `B`, `H`, `T` AS `M`\n"
+ + "FROM `SOURCE`";
+ sql(sql).ok(expected);
+
+ final String sql2 =
+ "CREATE MATERIALIZED TABLE tbl1\n"
+ + "(\n"
+ + " PRIMARY KEY (a, b)\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "FRESHNESS = INTERVAL '3' MINUTE\n"
+ + "REFRESH_MODE = FULL\n"
+ + "AS SELECT a, b, h, t m FROM source";
+ final String expected2 =
+ "CREATE MATERIALIZED TABLE `TBL1`\n"
+ + "(\n"
+ + " PRIMARY KEY (`A`, `B`)\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "FRESHNESS = INTERVAL '3' MINUTE\n"
+ + "REFRESH_MODE = FULL\n"
+ + "AS\n"
+ + "SELECT `A`, `B`, `H`, `T` AS `M`\n"
+ + "FROM `SOURCE`";
+
+ sql(sql2).ok(expected2);
+
+ final String sql3 =
+ "CREATE MATERIALIZED TABLE tbl1\n"
+ + "COMMENT 'table comment'\n"
+ + "FRESHNESS = INTERVAL '3' YEAR\n"
+ + "REFRESH_MODE = FULL\n"
+ + "AS SELECT a, b, h, t m FROM source";
+ final String expected3 =
+ "CREATE MATERIALIZED TABLE `TBL1`\n"
+ + "COMMENT 'table comment'\n"
+ + "FRESHNESS = INTERVAL '3' YEAR\n"
+ + "REFRESH_MODE = FULL\n"
+ + "AS\n"
+ + "SELECT `A`, `B`, `H`, `T` AS `M`\n"
+ + "FROM `SOURCE`";
+ sql(sql3).ok(expected3);
+ }
+
+ @Test
+ void testCreateMaterializedTableWithBadFreshnessInterval() {
+ final String sql =
+ "CREATE MATERIALIZED TABLE tbl1\n"
+ + "(\n"
+ + " PRIMARY KEY (a, b)\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "PARTITIONED BY (a, h)\n"
+ + "WITH (\n"
+ + " 'group.id' = 'latest', \n"
+ + " 'kafka.topic' = 'log.test'\n"
+ + ")\n"
+ + "FRESHNESS = ^123^\n"
+ + "AS SELECT a, b, h, t m FROM source";
+ sql(sql).fails(
+ "MATERIALIZED table FRESHNESS only support Interval Type, please refer to the materialized table document.");
+
+ final String sql2 =
+ "CREATE MATERIALIZED TABLE tbl1\n"
+ + "(\n"
+ + " PRIMARY KEY (a, b)\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "PARTITIONED BY (a, h)\n"
+ + "WITH (\n"
+ + " 'group.id' = 'latest', \n"
+ + " 'kafka.topic' = 'log.test'\n"
+ + ")\n"
+ + "^AS^ SELECT a, b, h, t m FROM source";
+
+ sql(sql2)
+ .fails(
+ "Encountered \"AS\" at line 11, column 1.\n"
+ + "Was expecting:\n"
+ + " \"FRESHNESS\" ...\n"
+ + " ");
+ }
+
+ @Test
+ void testCreateMaterializedTableWithNoAsQuery() {
Review Comment:
testCreateMaterializedTableWithoutAsQuery?
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/utils/ParserResource.java:
##########
@@ -49,4 +49,8 @@ public interface ParserResource {
@Resources.BaseMessage("Bucket count must be a positive integer.")
Resources.ExInst<ParseException> bucketCountMustBePositiveInteger();
+
+ @Resources.BaseMessage(
+ "MATERIALIZED table FRESHNESS only support Interval Type, please refer to the materialized table document.")
Review Comment:
```suggestion
"CREATE MATERIALIZED TABLE only supports interval type FRESHNESS, please refer to the materialized table document.")
```
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/utils/ParserResource.java:
##########
@@ -49,4 +49,8 @@ public interface ParserResource {
@Resources.BaseMessage("Bucket count must be a positive integer.")
Resources.ExInst<ParseException> bucketCountMustBePositiveInteger();
+
+ @Resources.BaseMessage(
+ "MATERIALIZED table FRESHNESS only support Interval Type, please refer to the materialized table document.")
+ Resources.ExInst<ParseException> freshnessUnsupportedType();
Review Comment:
```suggestion
Resources.ExInst<ParseException> unsupportedFreshnessType();
```
##########
flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableDDLParserTest.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Execution;
+
+import static org.junit.jupiter.api.parallel.ExecutionMode.CONCURRENT;
+
+/** Sql parser test for materialized related syntax. * */
+@Execution(CONCURRENT)
+public class MaterializedTableDDLParserTest extends FlinkSqlParserImplTest {
+
+ @Test
+ void testCreateMaterializedTable() {
+ final String sql =
+ "CREATE MATERIALIZED TABLE tbl1\n"
+ + "(\n"
+ + " PRIMARY KEY (a, b)\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "PARTITIONED BY (a, h)\n"
+ + "WITH (\n"
+ + " 'group.id' = 'latest', \n"
+ + " 'kafka.topic' = 'log.test'\n"
+ + ")\n"
+ + "FRESHNESS = INTERVAL '3' MINUTE\n"
+ + "AS SELECT a, b, h, t m FROM source";
+ final String expected =
+ "CREATE MATERIALIZED TABLE `TBL1`\n"
+ + "(\n"
+ + " PRIMARY KEY (`A`, `B`)\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "PARTITIONED BY (`A`, `H`)\n"
+ + "WITH (\n"
+ + " 'group.id' = 'latest',\n"
+ + " 'kafka.topic' = 'log.test'\n"
+ + ")\n"
+ + "FRESHNESS = INTERVAL '3' MINUTE\n"
+ + "AS\n"
+ + "SELECT `A`, `B`, `H`, `T` AS `M`\n"
+ + "FROM `SOURCE`";
+ sql(sql).ok(expected);
+
+ final String sql2 =
+ "CREATE MATERIALIZED TABLE tbl1\n"
+ + "(\n"
+ + " PRIMARY KEY (a, b)\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "FRESHNESS = INTERVAL '3' MINUTE\n"
+ + "REFRESH_MODE = FULL\n"
+ + "AS SELECT a, b, h, t m FROM source";
+ final String expected2 =
+ "CREATE MATERIALIZED TABLE `TBL1`\n"
+ + "(\n"
+ + " PRIMARY KEY (`A`, `B`)\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "FRESHNESS = INTERVAL '3' MINUTE\n"
+ + "REFRESH_MODE = FULL\n"
+ + "AS\n"
+ + "SELECT `A`, `B`, `H`, `T` AS `M`\n"
+ + "FROM `SOURCE`";
+
+ sql(sql2).ok(expected2);
+
+ final String sql3 =
+ "CREATE MATERIALIZED TABLE tbl1\n"
+ + "COMMENT 'table comment'\n"
+ + "FRESHNESS = INTERVAL '3' YEAR\n"
Review Comment:
We don't support `YEAR` interval unit, so this is not appropriate here
##########
flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableDDLParserTest.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Execution;
+
+import static org.junit.jupiter.api.parallel.ExecutionMode.CONCURRENT;
+
+/** Sql parser test for materialized related syntax. * */
+@Execution(CONCURRENT)
+public class MaterializedTableDDLParserTest extends FlinkSqlParserImplTest {
Review Comment:
Why not extands SqlParserTest directly? In addition, I think renaming this class name to MaterializedTableStatementParserTest will be more suitable. In the future, we may introduce more different kinds syntax.
##########
flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/MaterializedTableDDLParserTest.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Execution;
+
+import static org.junit.jupiter.api.parallel.ExecutionMode.CONCURRENT;
+
+/** Sql parser test for materialized related syntax. * */
+@Execution(CONCURRENT)
+public class MaterializedTableDDLParserTest extends FlinkSqlParserImplTest {
+
+ @Test
+ void testCreateMaterializedTable() {
+ final String sql =
+ "CREATE MATERIALIZED TABLE tbl1\n"
+ + "(\n"
+ + " PRIMARY KEY (a, b)\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "PARTITIONED BY (a, h)\n"
+ + "WITH (\n"
+ + " 'group.id' = 'latest', \n"
+ + " 'kafka.topic' = 'log.test'\n"
+ + ")\n"
+ + "FRESHNESS = INTERVAL '3' MINUTE\n"
+ + "AS SELECT a, b, h, t m FROM source";
+ final String expected =
+ "CREATE MATERIALIZED TABLE `TBL1`\n"
+ + "(\n"
+ + " PRIMARY KEY (`A`, `B`)\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "PARTITIONED BY (`A`, `H`)\n"
+ + "WITH (\n"
+ + " 'group.id' = 'latest',\n"
+ + " 'kafka.topic' = 'log.test'\n"
+ + ")\n"
+ + "FRESHNESS = INTERVAL '3' MINUTE\n"
+ + "AS\n"
+ + "SELECT `A`, `B`, `H`, `T` AS `M`\n"
+ + "FROM `SOURCE`";
+ sql(sql).ok(expected);
+
+ final String sql2 =
+ "CREATE MATERIALIZED TABLE tbl1\n"
+ + "(\n"
+ + " PRIMARY KEY (a, b)\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "FRESHNESS = INTERVAL '3' MINUTE\n"
+ + "REFRESH_MODE = FULL\n"
+ + "AS SELECT a, b, h, t m FROM source";
+ final String expected2 =
+ "CREATE MATERIALIZED TABLE `TBL1`\n"
+ + "(\n"
+ + " PRIMARY KEY (`A`, `B`)\n"
+ + ")\n"
+ + "COMMENT 'table comment'\n"
+ + "FRESHNESS = INTERVAL '3' MINUTE\n"
+ + "REFRESH_MODE = FULL\n"
+ + "AS\n"
+ + "SELECT `A`, `B`, `H`, `T` AS `M`\n"
+ + "FROM `SOURCE`";
+
+ sql(sql2).ok(expected2);
+
+ final String sql3 =
+ "CREATE MATERIALIZED TABLE tbl1\n"
+ + "COMMENT 'table comment'\n"
+ + "FRESHNESS = INTERVAL '3' YEAR\n"
+ + "REFRESH_MODE = FULL\n"
+ + "AS SELECT a, b, h, t m FROM source";
+ final String expected3 =
+ "CREATE MATERIALIZED TABLE `TBL1`\n"
+ + "COMMENT 'table comment'\n"
+ + "FRESHNESS = INTERVAL '3' YEAR\n"
+ + "REFRESH_MODE = FULL\n"
+ + "AS\n"
+ + "SELECT `A`, `B`, `H`, `T` AS `M`\n"
+ + "FROM `SOURCE`";
+ sql(sql3).ok(expected3);
+ }
+
+ @Test
+ void testCreateMaterializedTableWithBadFreshnessInterval() {
Review Comment:
testCreateMaterializedTableWithUnsupportedFreshnessInterval？
--
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
Re: [PR] [FLINK-35190][table] Support create materialized table syntax [flink]
Posted by "lsyldliu (via GitHub)" <gi...@apache.org>.
lsyldliu commented on PR #24707:
URL: https://github.com/apache/flink/pull/24707#issuecomment-2078505570
Sorry for forget to approve.
--
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