use flink planner

Author: todd5167

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -19,7 +19,6 @@
 
 package com.dtstack.flink.sql;
 
-import com.dtstack.flink.sql.config.CalciteConfig;
 import com.dtstack.flink.sql.classloader.ClassLoaderManager;
 import com.dtstack.flink.sql.enums.ClusterMode;
 import com.dtstack.flink.sql.enums.ECacheType;
@@ -29,6 +28,7 @@
 import com.dtstack.flink.sql.option.OptionParser;
 import com.dtstack.flink.sql.parser.CreateFuncParser;
 import com.dtstack.flink.sql.parser.CreateTmpTableParser;
+import com.dtstack.flink.sql.parser.FlinkPlanner;
 import com.dtstack.flink.sql.parser.InsertSqlParser;
 import com.dtstack.flink.sql.parser.SqlParser;
 import com.dtstack.flink.sql.parser.SqlTree;
@@ -60,6 +60,7 @@
 import org.apache.flink.table.api.Table;
 import org.apache.flink.table.api.TableEnvironment;
 import org.apache.flink.table.api.java.StreamTableEnvironment;
+import org.apache.flink.table.calcite.FlinkPlannerImpl;
 import org.apache.flink.table.sinks.TableSink;
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
@@ -72,7 +73,6 @@
 import java.net.URLDecoder;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.Properties;
 import java.util.Set;
 
@@ -120,6 +120,8 @@ public static void main(String[] args) throws Exception {
         StreamExecutionEnvironment env = getStreamExeEnv(confProperties, deployMode);
         StreamTableEnvironment tableEnv =  StreamTableEnvironment.getTableEnvironment(env);
         StreamQueryConfig streamQueryConfig = StreamEnvConfigManager.getStreamQueryConfig(tableEnv, confProperties);
+        // init global flinkPlanner
+        FlinkPlanner.createFlinkPlanner(tableEnv.getFrameworkConfig(), tableEnv.getPlanner(), tableEnv.getTypeFactory());
 
         List<URL> jarURList = Lists.newArrayList();
         SqlTree sqlTree = SqlParser.parseSql(sql);
@@ -171,8 +173,9 @@ private static void sqlTranslation(String localSqlPluginPath,
                 if (sqlTree.getTmpTableMap().containsKey(tableName)) {
                     CreateTmpTableParser.SqlParserResult tmp = sqlTree.getTmpTableMap().get(tableName);
                     String realSql = DtStringUtil.replaceIgnoreQuota(result.getExecSql(), "`", "");
+                    FlinkPlannerImpl flinkPlanner = FlinkPlanner.getFlinkPlanner();
 
-                    SqlNode sqlNode = org.apache.calcite.sql.parser.SqlParser.create(realSql, CalciteConfig.MYSQL_LEX_CONFIG).parseStmt();
+                    SqlNode sqlNode = flinkPlanner.parse(realSql);
                     String tmpSql = ((SqlInsert) sqlNode).getSource().toString();
                     tmp.setExecSql(tmpSql);
                     sideSqlExec.exec(tmp.getExecSql(), sideTableMap, tableEnv, registerTableCache, queryConfig, tmp);

core/src/main/java/com/dtstack/flink/sql/config/CalciteConfig.java

@@ -21,11 +21,10 @@
 package com.dtstack.flink.sql.parser;
 
 import com.dtstack.flink.sql.util.DtStringUtil;
-import org.apache.calcite.config.Lex;
 import org.apache.calcite.sql.*;
-import org.apache.calcite.sql.parser.SqlParseException;
-import org.apache.calcite.sql.parser.SqlParser;
 import com.google.common.collect.Lists;
+import org.apache.flink.table.calcite.FlinkPlannerImpl;
+
 import java.util.List;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -71,17 +70,12 @@ public void parseSql(String sql, SqlTree sqlTree) {
                 tableName = matcher.group(1);
                 selectSql = "select " + matcher.group(2);
             }
-
-            SqlParser.Config config = SqlParser
-                    .configBuilder()
-                    .setLex(Lex.MYSQL)
-                    .build();
-            SqlParser sqlParser = SqlParser.create(selectSql,config);
+            FlinkPlannerImpl flinkPlanner = FlinkPlanner.getFlinkPlanner();
 
             SqlNode sqlNode = null;
             try {
-                sqlNode = sqlParser.parseStmt();
-            } catch (SqlParseException e) {
+                sqlNode = flinkPlanner.parse(selectSql);
+            } catch (Exception e) {
                 throw new RuntimeException("", e);
             }
 

core/src/main/java/com/dtstack/flink/sql/parser/CreateTmpTableParser.java

@@ -0,0 +1,52 @@
+/*
+ * 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 com.dtstack.flink.sql.parser;
+
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.tools.FrameworkConfig;
+import org.apache.flink.table.calcite.FlinkPlannerImpl;
+import org.apache.flink.table.calcite.FlinkTypeFactory;
+
+/**
+ * Date: 2020/3/31
+ * Company: www.dtstack.com
+ * @author maqi
+ */
+public class FlinkPlanner {
+
+    public static volatile FlinkPlannerImpl flinkPlanner;
+
+    private FlinkPlanner() {
+    }
+
+    public static FlinkPlannerImpl createFlinkPlanner(FrameworkConfig frameworkConfig, RelOptPlanner relOptPlanner, FlinkTypeFactory typeFactory) {
+        if (flinkPlanner == null) {
+            synchronized (FlinkPlanner.class) {
+                if (flinkPlanner == null) {
+                    flinkPlanner = new FlinkPlannerImpl(frameworkConfig, relOptPlanner, typeFactory);
+                }
+            }
+        }
+        return flinkPlanner;
+    }
+
+    public static FlinkPlannerImpl getFlinkPlanner() {
+        return flinkPlanner;
+    }
+}

core/src/main/java/com/dtstack/flink/sql/parser/FlinkPlanner.java

@@ -19,7 +19,7 @@
 
 package com.dtstack.flink.sql.side;
 
-import com.dtstack.flink.sql.config.CalciteConfig;
+import com.dtstack.flink.sql.parser.FlinkPlanner;
 import com.dtstack.flink.sql.util.ParseUtils;
 import com.dtstack.flink.sql.util.TableUtils;
 import com.google.common.base.Preconditions;
@@ -38,11 +38,11 @@
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.fun.SqlCase;
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
-import org.apache.calcite.sql.parser.SqlParser;
 import org.apache.calcite.sql.parser.SqlParserPos;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.table.calcite.FlinkPlannerImpl;
 
 import java.util.List;
 import java.util.Map;
@@ -416,8 +416,9 @@ private void extractTemporaryQuery(SqlNode node, String tableAlias,
                     node.toString(),
                     extractConditionStr);
 
-            SqlParser sqlParser = SqlParser.create(tmpSelectSql, CalciteConfig.MYSQL_LEX_CONFIG);
-            SqlNode sqlNode = sqlParser.parseStmt();
+            FlinkPlannerImpl flinkPlanner = FlinkPlanner.getFlinkPlanner();
+            SqlNode sqlNode = flinkPlanner.parse(tmpSelectSql);
+
             SqlBasicCall sqlBasicCall = buildAsSqlNode(tableAlias, sqlNode);
             queueInfo.offer(sqlBasicCall);
 

core/src/main/java/com/dtstack/flink/sql/side/JoinNodeDealer.java

@@ -18,7 +18,7 @@
 
 package com.dtstack.flink.sql.side;
 
-import com.dtstack.flink.sql.config.CalciteConfig;
+import com.dtstack.flink.sql.parser.FlinkPlanner;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import org.apache.calcite.sql.SqlBasicCall;
@@ -30,11 +30,10 @@
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.SqlSelect;
 import org.apache.calcite.sql.parser.SqlParseException;
-import org.apache.calcite.sql.parser.SqlParser;
+import org.apache.flink.table.calcite.FlinkPlannerImpl;
 
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 import static org.apache.calcite.sql.SqlKind.*;
 
@@ -47,8 +46,8 @@
  */
 public class SidePredicatesParser {
     public void fillPredicatesForSideTable(String exeSql, Map<String, SideTableInfo> sideTableMap) throws SqlParseException {
-        SqlParser sqlParser = SqlParser.create(exeSql, CalciteConfig.MYSQL_LEX_CONFIG);
-        SqlNode sqlNode = sqlParser.parseStmt();
+        FlinkPlannerImpl flinkPlanner = FlinkPlanner.getFlinkPlanner();
+        SqlNode sqlNode = flinkPlanner.parse(exeSql);
         parseSql(sqlNode, sideTableMap, Maps.newHashMap());
     }
 

core/src/main/java/com/dtstack/flink/sql/side/SidePredicatesParser.java

@@ -20,43 +20,29 @@
 
 package com.dtstack.flink.sql.side;
 
-import com.dtstack.flink.sql.config.CalciteConfig;
+import com.dtstack.flink.sql.parser.FlinkPlanner;
 import com.dtstack.flink.sql.util.TableUtils;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Queues;
 import com.google.common.collect.Sets;
-import org.apache.calcite.sql.JoinType;
-import org.apache.calcite.sql.SqlAsOperator;
 import org.apache.calcite.sql.SqlBasicCall;
-import org.apache.calcite.sql.SqlBinaryOperator;
-import org.apache.calcite.sql.SqlDataTypeSpec;
 import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlInsert;
 import org.apache.calcite.sql.SqlJoin;
 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.SqlOrderBy;
 import org.apache.calcite.sql.SqlSelect;
 import org.apache.calcite.sql.SqlWith;
 import org.apache.calcite.sql.SqlWithItem;
-import org.apache.calcite.sql.fun.SqlCase;
-import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.apache.calcite.sql.parser.SqlParseException;
-import org.apache.calcite.sql.parser.SqlParser;
-import org.apache.calcite.sql.parser.SqlParserPos;
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.table.api.Table;
+import org.apache.flink.table.calcite.FlinkPlannerImpl;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.List;
 import java.util.Map;
 import java.util.Queue;
 import java.util.Set;
@@ -82,8 +68,8 @@ public Queue<Object> getExeQueue(String exeSql, Set<String> sideTableSet) throws
         LOG.info(exeSql);
 
         Queue<Object> queueInfo = Queues.newLinkedBlockingQueue();
-        SqlParser sqlParser = SqlParser.create(exeSql, CalciteConfig.MYSQL_LEX_CONFIG);
-        SqlNode sqlNode = sqlParser.parseStmt();
+        FlinkPlannerImpl flinkPlanner = FlinkPlanner.getFlinkPlanner();
+        SqlNode sqlNode = flinkPlanner.parse(exeSql);
 
         parseSql(sqlNode, sideTableSet, queueInfo, null, null, null);
         queueInfo.offer(sqlNode);