Merge branch 'v1.8.0_dev' into 1.8_v3.9.0_beta_1.0

Author: gituser

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

@@ -20,7 +20,8 @@
 
 package com.dtstack.flink.sql;
 
-import com.dtstack.flink.sql.classloader.DtClassLoader;
+import com.dtstack.flink.sql.config.CalciteConfig;
+import com.dtstack.flink.sql.classloader.ClassLoaderManager;
 import com.dtstack.flink.sql.constrant.ConfigConstrant;
 import com.dtstack.flink.sql.enums.ClusterMode;
 import com.dtstack.flink.sql.enums.ECacheType;
@@ -77,8 +78,6 @@
 import java.net.URL;
 import java.net.URLClassLoader;
 import java.net.URLDecoder;
-import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
@@ -101,10 +100,6 @@ public class Main {
 
     private static final Logger LOG = LoggerFactory.getLogger(Main.class);
 
-    private static Config config = org.apache.calcite.sql.parser.SqlParser
-            .configBuilder()
-            .setLex(Lex.MYSQL)
-            .build();
 
     public static void main(String[] args) throws Exception {
 
@@ -126,10 +121,6 @@ public static void main(String[] args) throws Exception {
             addJarFileList = objMapper.readValue(addJarListStr, List.class);
         }
 
-        ClassLoader threadClassLoader = Thread.currentThread().getContextClassLoader();
-        DtClassLoader parentClassloader = new DtClassLoader(new URL[]{}, threadClassLoader);
-        Thread.currentThread().setContextClassLoader(parentClassloader);
-
         confProp = URLDecoder.decode(confProp, Charsets.UTF_8.toString());
         Properties confProperties = PluginUtil.jsonStrToObject(confProp, Properties.class);
         StreamExecutionEnvironment env = getStreamExeEnv(confProperties, deployMode);
@@ -148,16 +139,14 @@ public static void main(String[] args) throws Exception {
         Map<String, Table> registerTableCache = Maps.newHashMap();
 
         //register udf
-        registerUDF(sqlTree, jarURList, parentClassloader, tableEnv);
+        registerUDF(sqlTree, jarURList, tableEnv);
         //register table schema
         registerTable(sqlTree, env, tableEnv, localSqlPluginPath, remoteSqlPluginPath, sideTableMap, registerTableCache);
 
         sqlTranslation(options,tableEnv,sqlTree,sideTableMap,registerTableCache);
 
         if(env instanceof MyLocalStreamEnvironment) {
-            List<URL> urlList = new ArrayList<>();
-            urlList.addAll(Arrays.asList(parentClassloader.getURLs()));
-            ((MyLocalStreamEnvironment) env).setClasspaths(urlList);
+            ((MyLocalStreamEnvironment) env).setClasspaths(ClassLoaderManager.getClassPath());
         }
 
         env.execute(name);
@@ -180,7 +169,7 @@ private static void sqlTranslation(Options options,StreamTableEnvironment tableE
                     CreateTmpTableParser.SqlParserResult tmp = sqlTree.getTmpTableMap().get(tableName);
                     String realSql = DtStringUtil.replaceIgnoreQuota(result.getExecSql(), "`", "");
 
-                    SqlNode sqlNode = org.apache.calcite.sql.parser.SqlParser.create(realSql,config).parseStmt();
+                    SqlNode sqlNode = org.apache.calcite.sql.parser.SqlParser.create(realSql, CalciteConfig.MYSQL_LEX_CONFIG).parseStmt();
                     String tmpSql = ((SqlInsert) sqlNode).getSource().toString();
                     tmp.setExecSql(tmpSql);
                     sideSqlExec.registerTmpTable(tmp, sideTableMap, tableEnv, registerTableCache);
@@ -224,19 +213,19 @@ private static void addEnvClassPath(StreamExecutionEnvironment env, Set<URL> cla
         }
     }
 
-    private static void registerUDF(SqlTree sqlTree, List<URL> jarURList, URLClassLoader parentClassloader,
-                                    StreamTableEnvironment tableEnv)
-            throws ClassNotFoundException, NoSuchMethodException, IllegalAccessException, InvocationTargetException {
-        List<CreateFuncParser.SqlParserResult> funcList = sqlTree.getFunctionList();
-        if (funcList.isEmpty()) {
-            return;
-        }
-        //load jar
-        URLClassLoader classLoader = FlinkUtil.loadExtraJar(jarURList, parentClassloader);
+    private static void registerUDF(SqlTree sqlTree, List<URL> jarURList, StreamTableEnvironment tableEnv)
+            throws NoSuchMethodException, IllegalAccessException, InvocationTargetException {
         //register urf
+        // udf和tableEnv须由同一个类加载器加载
+        ClassLoader levelClassLoader = tableEnv.getClass().getClassLoader();
+        URLClassLoader classLoader = null;
+        List<CreateFuncParser.SqlParserResult> funcList = sqlTree.getFunctionList();
         for (CreateFuncParser.SqlParserResult funcInfo : funcList) {
-            FlinkUtil.registerUDF(funcInfo.getType(), funcInfo.getClassName(), funcInfo.getName(),
-                    tableEnv, classLoader);
+            //classloader
+            if (classLoader == null) {
+                classLoader = FlinkUtil.loadExtraJar(jarURList, (URLClassLoader)levelClassLoader);
+            }
+            FlinkUtil.registerUDF(funcInfo.getType(), funcInfo.getClassName(), funcInfo.getName(), tableEnv, classLoader);
         }
     }
 

core/src/main/java/com/dtstack/flink/sql/classloader/ClassLoaderManager.java

@@ -0,0 +1,94 @@
+/*
+ * 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.classloader;
+
+import com.dtstack.flink.sql.util.PluginUtil;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * company: www.dtstack.com
+ * author: toutian
+ * create: 2019/10/14
+ */
+public class ClassLoaderManager {
+
+    private static final Logger LOG = LoggerFactory.getLogger(ClassLoaderManager.class);
+
+    private static Map<String, DtClassLoader> pluginClassLoader = new ConcurrentHashMap<>();
+
+    public static <R> R newInstance(String pluginJarPath, ClassLoaderSupplier<R> supplier) throws Exception {
+        ClassLoader classLoader = retrieveClassLoad(pluginJarPath);
+        return ClassLoaderSupplierCallBack.callbackAndReset(supplier, classLoader);
+    }
+
+    public static <R> R newInstance(List<URL> jarUrls, ClassLoaderSupplier<R> supplier) throws Exception {
+        ClassLoader classLoader = retrieveClassLoad(jarUrls);
+        return ClassLoaderSupplierCallBack.callbackAndReset(supplier, classLoader);
+    }
+
+    private static DtClassLoader retrieveClassLoad(String pluginJarPath) {
+        return pluginClassLoader.computeIfAbsent(pluginJarPath, k -> {
+            try {
+                URL[] urls = PluginUtil.getPluginJarUrls(pluginJarPath);
+                ClassLoader parentClassLoader = Thread.currentThread().getContextClassLoader();
+                DtClassLoader classLoader = new DtClassLoader(urls, parentClassLoader);
+                LOG.info("pluginJarPath:{} create ClassLoad successful...", pluginJarPath);
+                return classLoader;
+            } catch (Throwable e) {
+                LOG.error("retrieve ClassLoad happens error:{}", e);
+                throw new RuntimeException("retrieve ClassLoad happens error");
+            }
+        });
+    }
+
+    private static DtClassLoader retrieveClassLoad(List<URL> jarUrls) {
+        jarUrls.sort(Comparator.comparing(URL::toString));
+        String jarUrlkey = StringUtils.join(jarUrls, "_");
+        return pluginClassLoader.computeIfAbsent(jarUrlkey, k -> {
+            try {
+                URL[] urls = jarUrls.toArray(new URL[jarUrls.size()]);
+                ClassLoader parentClassLoader = Thread.currentThread().getContextClassLoader();
+                DtClassLoader classLoader = new DtClassLoader(urls, parentClassLoader);
+                LOG.info("jarUrl:{} create ClassLoad successful...", jarUrlkey);
+                return classLoader;
+            } catch (Throwable e) {
+                LOG.error("retrieve ClassLoad happens error:{}", e);
+                throw new RuntimeException("retrieve ClassLoad happens error");
+            }
+        });
+    }
+
+    public static List<URL> getClassPath() {
+        List<URL> classPaths = new ArrayList<>();
+        for (Map.Entry<String, DtClassLoader> entry : pluginClassLoader.entrySet()) {
+            classPaths.addAll(Arrays.asList(entry.getValue().getURLs()));
+        }
+        return classPaths;
+    }
+}

core/src/main/java/com/dtstack/flink/sql/classloader/ClassLoaderSupplier.java

@@ -0,0 +1,44 @@
+/*
+ * 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.classloader;
+
+/**
+ * Represents a supplier of results.
+ *
+ * <p>There is no requirement that a new or distinct result be returned each
+ * time the supplier is invoked.
+ *
+ * <p>This is a <a href="package-summary.html">functional interface</a>
+ * whose functional method is {@link #get()}.
+ *
+ * @param <T> the type of results supplied by this supplier
+ *
+ * @since 1.8
+ */
+@FunctionalInterface
+public interface ClassLoaderSupplier<T> {
+
+    /**
+     * Gets a result.
+     *
+     * @return a result
+     */
+    T get(ClassLoader cl) throws Exception;
+}

core/src/main/java/com/dtstack/flink/sql/classloader/ClassLoaderSupplierCallBack.java

@@ -0,0 +1,40 @@
+/*
+ * 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.classloader;
+
+/**
+ * company: www.dtstack.com
+ * author: toutian
+ * create: 2019/10/14
+ */
+public class ClassLoaderSupplierCallBack {
+
+    public static <R> R callbackAndReset(ClassLoaderSupplier<R> supplier, ClassLoader toSetClassLoader) throws Exception {
+        ClassLoader oldClassLoader = Thread.currentThread().getContextClassLoader();
+        Thread.currentThread().setContextClassLoader(toSetClassLoader);
+        try {
+            return supplier.get(toSetClassLoader);
+        } finally {
+            Thread.currentThread().setContextClassLoader(oldClassLoader);
+        }
+    }
+
+
+}

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

@@ -0,0 +1,35 @@
+/*
+ * 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.config;
+
+import org.apache.calcite.config.Lex;
+import org.apache.calcite.sql.parser.SqlParser;
+import org.apache.calcite.sql.parser.SqlParser.Config;
+
+public class CalciteConfig {
+
+    public static Config MYSQL_LEX_CONFIG = SqlParser
+            .configBuilder()
+            .setLex(Lex.MYSQL)
+            .build();
+
+
+
+}

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

@@ -25,6 +25,7 @@
 import org.apache.flink.calcite.shaded.com.google.common.base.Strings;
 
 import java.io.Serializable;
+import java.util.Map;
 
 /**
  * Join信息
@@ -40,6 +41,8 @@ public class JoinInfo implements Serializable {
 
     //左表是否是维表
     private boolean leftIsSideTable;
+    //左表是 转换后的中间表
+    private boolean leftIsMidTable;
 
     //右表是否是维表
     private boolean rightIsSideTable;
@@ -63,6 +66,8 @@ public class JoinInfo implements Serializable {
     private SqlNode selectNode;
 
     private JoinType joinType;
+    // 左边是中间转换表，做表映射关系，给替换属性名称使用
+    private Map<String, String> leftTabMapping;
 
     public String getSideTableName(){
         if(leftIsSideTable){
@@ -87,6 +92,22 @@ public String getNewTableName(){
         return leftStr + "_" + rightTableName;
     }
 
+    public boolean isLeftIsMidTable() {
+        return leftIsMidTable;
+    }
+
+    public void setLeftIsMidTable(boolean leftIsMidTable) {
+        this.leftIsMidTable = leftIsMidTable;
+    }
+
+    public Map<String, String> getLeftTabMapping() {
+        return leftTabMapping;
+    }
+
+    public void setLeftTabMapping(Map<String, String> leftTabMapping) {
+        this.leftTabMapping = leftTabMapping;
+    }
+
     public String getNewTableAlias(){
         return leftTableAlias + "_" + rightTableAlias;
     }

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

@@ -41,8 +41,12 @@
 
 public class ParserJoinField {
 
+
     /**
-     * Need to parse the fields of information and where selectlist
+     *  build row by field
+     * @param sqlNode  select node
+     * @param scope    join left and right table all info
+     * @param getAll  true,get all fields from two tables; false, extract useful field from select node
      * @return
      */
     public static List<FieldInfo> getRowTypeInfo(SqlNode sqlNode, JoinScope scope, boolean getAll){