forked from apache/paimon-webui
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
9 changed files
with
307 additions
and
21 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
92 changes: 92 additions & 0 deletions
92
...ommon/src/main/java/org/apache/paimon/web/engine/flink/common/parser/CustomSqlParser.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,92 @@ | ||
| /* | ||
| * 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.paimon.web.engine.flink.common.parser; | ||
|
|
||
| import org.apache.calcite.config.Lex; | ||
| import org.apache.calcite.sql.SqlIdentifier; | ||
| import org.apache.calcite.sql.SqlLiteral; | ||
| import org.apache.calcite.sql.SqlNode; | ||
| import org.apache.calcite.sql.SqlNodeList; | ||
| import org.apache.calcite.sql.SqlSelect; | ||
| import org.apache.calcite.sql.parser.SqlParseException; | ||
| import org.apache.calcite.sql.parser.SqlParser; | ||
| import org.apache.calcite.sql.parser.SqlParserPos; | ||
| import org.apache.flink.sql.parser.impl.FlinkSqlParserImpl; | ||
| import org.apache.flink.sql.parser.validate.FlinkSqlConformance; | ||
|
|
||
| /** CustomSqlParser to parse Sql list. */ | ||
| public class CustomSqlParser { | ||
|
|
||
| private static final SqlParser.Config config; | ||
| private final SqlParser parser; | ||
| private final int limit; | ||
|
|
||
| private static final int DEFAULT_LIMIT = 500; | ||
|
|
||
| static { | ||
| config = | ||
| SqlParser.config() | ||
| .withParserFactory(FlinkSqlParserImpl.FACTORY) | ||
| .withConformance(FlinkSqlConformance.DEFAULT) | ||
| .withLex(Lex.JAVA) | ||
| .withIdentifierMaxLength(256); | ||
| } | ||
|
|
||
| public CustomSqlParser(String sql) { | ||
| this(sql, DEFAULT_LIMIT); | ||
| } | ||
|
|
||
| public CustomSqlParser(String sql, int limit) { | ||
| this.parser = SqlParser.create(sql, config); | ||
| this.limit = limit; | ||
| } | ||
|
|
||
| public SqlNodeList parseStmtList() throws SqlParseException { | ||
| SqlNodeList nodeList = parser.parseStmtList(); | ||
| for (SqlNode node : nodeList) { | ||
| if (node instanceof SqlSelect) { | ||
| SqlSelect select = (SqlSelect) node; | ||
| if (!hasAggregateOrGroupBy(select) && select.getFetch() == null) { | ||
| SqlLiteral sqlLiteral = | ||
| SqlLiteral.createExactNumeric(String.valueOf(limit), SqlParserPos.ZERO); | ||
| select.setFetch(sqlLiteral); | ||
| } | ||
| } | ||
| } | ||
| return nodeList; | ||
| } | ||
|
|
||
| private boolean hasAggregateOrGroupBy(SqlSelect select) { | ||
| if (select.getGroup() != null && !select.getGroup().isEmpty()) { | ||
| return true; | ||
| } | ||
| return containsComplexOperations(select.getSelectList()); | ||
| } | ||
|
|
||
| private boolean containsComplexOperations(SqlNodeList nodes) { | ||
| if (nodes != null) { | ||
| for (SqlNode node : nodes) { | ||
| if (!(node instanceof SqlIdentifier)) { | ||
| return true; | ||
| } | ||
| } | ||
| } | ||
| return false; | ||
| } | ||
| } |
54 changes: 54 additions & 0 deletions
54
...n/src/test/java/org/apache/paimon/web/engine/flink/common/parser/CustomSqlParserTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,54 @@ | ||
| /* | ||
| * 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.paimon.web.engine.flink.common.parser; | ||
|
|
||
| import org.apache.calcite.sql.SqlNodeList; | ||
| import org.apache.calcite.sql.parser.SqlParseException; | ||
| import org.junit.jupiter.api.Test; | ||
|
|
||
| import static org.apache.paimon.web.engine.flink.common.parser.StatementsConstant.statement1; | ||
| import static org.apache.paimon.web.engine.flink.common.parser.StatementsConstant.statement2; | ||
| import static org.apache.paimon.web.engine.flink.common.parser.StatementsConstant.statement3; | ||
| import static org.assertj.core.api.Assertions.assertThat; | ||
|
|
||
| /** Tests of {@link CustomSqlParser}. */ | ||
| public class CustomSqlParserTest { | ||
|
|
||
| @Test | ||
| public void testParse() throws SqlParseException { | ||
| CustomSqlParser customSqlParser = new CustomSqlParser(statement1); | ||
| SqlNodeList sqlNodeList = customSqlParser.parseStmtList(); | ||
| assertThat(sqlNodeList.size()).isEqualTo(5); | ||
| } | ||
|
|
||
| @Test | ||
| public void testSelectLimit() throws SqlParseException { | ||
| CustomSqlParser customSqlParser = new CustomSqlParser(statement2); | ||
| String actual = customSqlParser.parseStmtList().get(2).toString(); | ||
| assertThat(actual) | ||
| .isEqualToIgnoringWhitespace("SELECT * FROM `t_order` FETCH NEXT 500 ROWS ONLY"); | ||
| } | ||
|
|
||
| @Test | ||
| public void testSelectWithoutLimit() throws SqlParseException { | ||
| CustomSqlParser customSqlParser = new CustomSqlParser(statement3); | ||
| String actual = customSqlParser.parseStmtList().get(2).toString(); | ||
| assertThat(actual).isEqualToIgnoringWhitespace("SELECT COUNT(*) FROM `t_order`"); | ||
| } | ||
| } |
117 changes: 117 additions & 0 deletions
117
...on/src/test/java/org/apache/paimon/web/engine/flink/common/parser/StatementsConstant.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,117 @@ | ||
| /* | ||
| * 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.paimon.web.engine.flink.common.parser; | ||
|
|
||
| /** Statements constant. */ | ||
| public class StatementsConstant { | ||
|
|
||
| public static String statement1 = | ||
| "DROP TABLE IF EXISTS t_order;\n" | ||
| + "CREATE TABLE IF NOT EXISTS t_order(\n" | ||
| + " --order id\n" | ||
| + " `order_id` BIGINT,\n" | ||
| + " --product\n" | ||
| + " `product` BIGINT,\n" | ||
| + " --amount\n" | ||
| + " `amount` BIGINT,\n" | ||
| + " --payment time\n" | ||
| + " `order_time` as CAST(CURRENT_TIMESTAMP AS TIMESTAMP(3)),\n" | ||
| + " --WATERMARK\n" | ||
| + " WATERMARK FOR order_time AS order_time-INTERVAL '2' SECOND\n" | ||
| + ") WITH(\n" | ||
| + " 'connector' = 'datagen',\n" | ||
| + " 'rows-per-second' = '1',\n" | ||
| + " 'fields.order_id.min' = '1',\n" | ||
| + " 'fields.order_id.max' = '2',\n" | ||
| + " 'fields.amount.min' = '1',\n" | ||
| + " 'fields.amount.max' = '10',\n" | ||
| + " 'fields.product.min' = '1',\n" | ||
| + " 'fields.product.max' = '2'\n" | ||
| + ");\n" | ||
| + "-- SELECT * FROM t_order LIMIT 10;\n" | ||
| + "DROP TABLE IF EXISTS sink_table;\n" | ||
| + "CREATE TABLE IF NOT EXISTS sink_table(\n" | ||
| + " --product\n" | ||
| + " `product` BIGINT,\n" | ||
| + " --amount\n" | ||
| + " `amount` BIGINT,\n" | ||
| + " --payment time\n" | ||
| + " `order_time` TIMESTAMP(3),\n" | ||
| + " `one_minute_sum` BIGINT\n" | ||
| + ") WITH('connector' = 'print');\n" | ||
| + "\n" | ||
| + "INSERT INTO\n" | ||
| + " sink_table\n" | ||
| + "SELECT\n" | ||
| + " product,\n" | ||
| + " amount,\n" | ||
| + " order_time,\n" | ||
| + " 0 as one_minute_sum\n" | ||
| + "FROM\n" | ||
| + " t_order;"; | ||
|
|
||
| public static String statement2 = | ||
| "DROP TABLE IF EXISTS t_order;\n" | ||
| + "CREATE TABLE IF NOT EXISTS t_order(\n" | ||
| + " --order id\n" | ||
| + " `order_id` BIGINT,\n" | ||
| + " --product\n" | ||
| + " `product` BIGINT,\n" | ||
| + " --amount\n" | ||
| + " `amount` BIGINT,\n" | ||
| + " --payment time\n" | ||
| + " `order_time` as CAST(CURRENT_TIMESTAMP AS TIMESTAMP(3)),\n" | ||
| + " --WATERMARK\n" | ||
| + " WATERMARK FOR order_time AS order_time-INTERVAL '2' SECOND\n" | ||
| + ") WITH(\n" | ||
| + " 'connector' = 'datagen',\n" | ||
| + " 'rows-per-second' = '1',\n" | ||
| + " 'fields.order_id.min' = '1',\n" | ||
| + " 'fields.order_id.max' = '2',\n" | ||
| + " 'fields.amount.min' = '1',\n" | ||
| + " 'fields.amount.max' = '10',\n" | ||
| + " 'fields.product.min' = '1',\n" | ||
| + " 'fields.product.max' = '2'\n" | ||
| + ");\n" | ||
| + "SELECT * FROM t_order;"; | ||
| public static String statement3 = | ||
| "DROP TABLE IF EXISTS t_order;\n" | ||
| + "CREATE TABLE IF NOT EXISTS t_order(\n" | ||
| + " --order id\n" | ||
| + " `order_id` BIGINT,\n" | ||
| + " --product\n" | ||
| + " `product` BIGINT,\n" | ||
| + " --amount\n" | ||
| + " `amount` BIGINT,\n" | ||
| + " --payment time\n" | ||
| + " `order_time` as CAST(CURRENT_TIMESTAMP AS TIMESTAMP(3)),\n" | ||
| + " --WATERMARK\n" | ||
| + " WATERMARK FOR order_time AS order_time-INTERVAL '2' SECOND\n" | ||
| + ") WITH(\n" | ||
| + " 'connector' = 'datagen',\n" | ||
| + " 'rows-per-second' = '1',\n" | ||
| + " 'fields.order_id.min' = '1',\n" | ||
| + " 'fields.order_id.max' = '2',\n" | ||
| + " 'fields.amount.min' = '1',\n" | ||
| + " 'fields.amount.max' = '10',\n" | ||
| + " 'fields.product.min' = '1',\n" | ||
| + " 'fields.product.max' = '2'\n" | ||
| + ");\n" | ||
| + "SELECT count(*) FROM t_order;"; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.