mysql side with all-cache Strategy

Author: zoudaokoulife

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

@@ -20,8 +20,10 @@
 
 package com.dtstack.flink.sql.parser;
 
+import com.dtstack.flink.sql.util.DtStringUtil;
 import org.apache.flink.calcite.shaded.com.google.common.collect.Maps;
 
+import java.util.List;
 import java.util.Map;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -70,9 +72,9 @@ private Map parseProp(String propsStr){
         String[] strs = propsStr.trim().split("'\\s*,");
         Map<String, Object> propMap = Maps.newHashMap();
         for(int i=0; i<strs.length; i++){
-            String[] ss = strs[i].split("=");
-            String key = ss[0].trim();
-            String value = ss[1].trim().replaceAll("'", "").trim();
+            List<String> ss = DtStringUtil.splitIgnoreQuota(strs[i], '=');
+            String key = ss.get(0).trim();
+            String value = ss.get(1).trim().replaceAll("'", "").trim();
             propMap.put(key, value);
         }
 

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

@@ -20,8 +20,15 @@
 
 package com.dtstack.flink.sql.side;
 
-import org.apache.calcite.interpreter.Row;
+import com.dtstack.flink.sql.threadFactory.DTThreadFactory;
 import org.apache.flink.api.common.functions.RichFlatMapFunction;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.types.Row;
+
+import java.sql.SQLException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Reason:
@@ -31,4 +38,32 @@
  */
 
 public abstract class AllReqRow extends RichFlatMapFunction<Row, Row>{
+
+    protected SideInfo sideInfo;
+
+    private ScheduledExecutorService es;
+
+    public AllReqRow(SideInfo sideInfo){
+        this.sideInfo = sideInfo;
+
+    }
+
+    protected abstract Row fillData(Row input, Object sideInput);
+
+    protected abstract void initCache() throws SQLException;
+
+    protected abstract void reloadCache();
+
+    @Override
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+        initCache();
+        System.out.println("----- all cacheRef init end-----");
+
+        //start reload cache thread
+        SideTableInfo sideTableInfo = sideInfo.getSideTableInfo();
+        es = Executors.newSingleThreadScheduledExecutor(new DTThreadFactory("cache-all-reload"));
+        es.scheduleAtFixedRate(() -> reloadCache(), sideTableInfo.getCacheTimeout(), sideTableInfo.getCacheTimeout(), TimeUnit.MILLISECONDS);
+    }
+
 }

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

@@ -25,23 +25,12 @@
 import com.dtstack.flink.sql.side.cache.CacheObj;
 import com.dtstack.flink.sql.side.cache.LRUSideCache;
 import org.apache.calcite.sql.JoinType;
-import org.apache.calcite.sql.SqlBasicCall;
-import org.apache.calcite.sql.SqlIdentifier;
-import org.apache.calcite.sql.SqlKind;
-import org.apache.calcite.sql.SqlNode;
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
-import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
-import org.apache.flink.calcite.shaded.com.google.common.collect.Maps;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
 import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
 import org.apache.flink.types.Row;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.Collections;
-import java.util.List;
-import java.util.Map;
 
 /**
  * All interfaces inherit naming rules: type + "AsyncReqRow" such as == "MysqlAsyncReqRow
@@ -55,22 +44,22 @@ public abstract class AsyncReqRow extends RichAsyncFunction<Row, Row> {
 
     private static final long serialVersionUID = 2098635244857937717L;
 
-    protected SideReqRow sideReqRow;
+    protected SideInfo sideInfo;
 
-    public AsyncReqRow(SideReqRow sideReqRow){
-        this.sideReqRow = sideReqRow;
+    public AsyncReqRow(SideInfo sideInfo){
+        this.sideInfo = sideInfo;
     }
 
     private void initCache(){
-        SideTableInfo sideTableInfo = sideReqRow.getSideTableInfo();
+        SideTableInfo sideTableInfo = sideInfo.getSideTableInfo();
         if(sideTableInfo.getCacheType() == null || ECacheType.NONE.name().equalsIgnoreCase(sideTableInfo.getCacheType())){
             return;
         }
 
         AbsSideCache sideCache;
         if(ECacheType.LRU.name().equalsIgnoreCase(sideTableInfo.getCacheType())){
             sideCache = new LRUSideCache(sideTableInfo);
-            sideReqRow.setSideCache(sideCache);
+            sideInfo.setSideCache(sideCache);
         }else{
             throw new RuntimeException("not support side cache with type:" + sideTableInfo.getCacheType());
         }
@@ -79,22 +68,22 @@ private void initCache(){
     }
 
     protected CacheObj getFromCache(String key){
-        return sideReqRow.getSideCache().getFromCache(key);
+        return sideInfo.getSideCache().getFromCache(key);
     }
 
     protected void putCache(String key, CacheObj value){
-        sideReqRow.getSideCache().putCache(key, value);
+        sideInfo.getSideCache().putCache(key, value);
     }
 
     protected boolean openCache(){
-        return sideReqRow.getSideCache() != null;
+        return sideInfo.getSideCache() != null;
     }
 
 
     protected abstract Row fillData(Row input, Object sideInput);
 
     protected void dealMissKey(Row input, ResultFuture<Row> resultFuture){
-        if(sideReqRow.getJoinType() == JoinType.LEFT){
+        if(sideInfo.getJoinType() == JoinType.LEFT){
             //Reserved left table data
             Row row = fillData(input, null);
             resultFuture.complete(Collections.singleton(row));

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

@@ -30,6 +30,7 @@
 import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
 import org.apache.flink.calcite.shaded.com.google.common.collect.Maps;
 
+import java.io.Serializable;
 import java.util.List;
 import java.util.Map;
 
@@ -40,7 +41,7 @@
  * @author xuchao
  */
 
-public abstract class SideReqRow {
+public abstract class SideInfo implements Serializable{
 
     protected RowTypeInfo rowTypeInfo;
 
@@ -56,17 +57,21 @@ public abstract class SideReqRow {
 
     protected JoinType joinType;
 
-    //key:Returns the value of the position, returns the index values ​​in the input data
+    //key:Returns the value of the position, value: the ref field index​in the input table
     protected Map<Integer, Integer> inFieldIndex = Maps.newHashMap();
 
+    //key:Returns the value of the position, value:  the ref field index​in the side table
     protected Map<Integer, Integer> sideFieldIndex = Maps.newHashMap();
 
+    //key:Returns the value of the position, value:  the ref field name​in the side table
+    protected Map<Integer, String> sideFieldNameIndex = Maps.newHashMap();
+
     protected SideTableInfo sideTableInfo;
 
     protected AbsSideCache sideCache;
 
-    public SideReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList,
-                       SideTableInfo sideTableInfo){
+    public SideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList,
+                    SideTableInfo sideTableInfo){
         this.rowTypeInfo = rowTypeInfo;
         this.outFieldInfoList = outFieldInfoList;
         this.joinType = joinInfo.getJoinType();
@@ -86,6 +91,7 @@ public void parseSelectFields(JoinInfo joinInfo){
             if(fieldInfo.getTable().equalsIgnoreCase(sideTableName)){
                 fields.add(fieldInfo.getFieldName());
                 sideFieldIndex.put(i, sideIndex);
+                sideFieldNameIndex.put(i, fieldInfo.getFieldName());
                 sideIndex++;
             }else if(fieldInfo.getTable().equalsIgnoreCase(nonSideTableName)){
                 int nonSideIndex = rowTypeInfo.getFieldIndex(fieldInfo.getFieldName());
@@ -241,4 +247,12 @@ public AbsSideCache getSideCache() {
     public void setSideCache(AbsSideCache sideCache) {
         this.sideCache = sideCache;
     }
+
+    public Map<Integer, String> getSideFieldNameIndex() {
+        return sideFieldNameIndex;
+    }
+
+    public void setSideFieldNameIndex(Map<Integer, String> sideFieldNameIndex) {
+        this.sideFieldNameIndex = sideFieldNameIndex;
+    }
 }

…m/dtstack/flink/sql/side/SideReqRow.java …com/dtstack/flink/sql/side/SideInfo.javacore/src/main/java/com/dtstack/flink/sql/side/SideReqRow.java renamed to core/src/main/java/com/dtstack/flink/sql/side/SideInfo.java core/src/main/java/com/dtstack/flink/sql/side/SideReqRow.java renamed to core/src/main/java/com/dtstack/flink/sql/side/SideInfo.java

@@ -20,9 +20,9 @@
 
 package com.dtstack.flink.sql.side;
 
-import com.dtstack.flink.sql.classloader.DtClassLoader;
 import com.dtstack.flink.sql.enums.ECacheType;
-import com.dtstack.flink.sql.util.PluginUtil;
+import com.dtstack.flink.sql.side.operator.SideAsyncOperator;
+import com.dtstack.flink.sql.side.operator.SideWithAllCacheOperator;
 import org.apache.calcite.sql.SqlBasicCall;
 import org.apache.calcite.sql.SqlDataTypeSpec;
 import org.apache.calcite.sql.SqlIdentifier;
@@ -42,7 +42,6 @@
 import org.apache.flink.calcite.shaded.com.google.common.collect.HashBasedTable;
 import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
 import org.apache.flink.calcite.shaded.com.google.common.collect.Maps;
-import org.apache.flink.streaming.api.datastream.AsyncDataStream;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.table.api.Table;
 import org.apache.flink.table.api.java.StreamTableEnvironment;
@@ -51,7 +50,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Queue;
-import java.util.concurrent.TimeUnit;
 
 import static org.apache.calcite.sql.SqlKind.*;
 
@@ -161,9 +159,9 @@ public void exec(String sql, Map<String, SideTableInfo> sideTableMap, StreamTabl
 
                 DataStream dsOut = null;
                 if(ECacheType.ALL.name().equalsIgnoreCase(sideTableInfo.getCacheType())){
-
+                    dsOut = SideWithAllCacheOperator.getSideJoinDataStream(adaptStream, sideTableInfo.getType(), localSqlPluginPath, typeInfo, joinInfo, sideJoinFieldInfo, sideTableInfo);
                 }else{
-                    dsOut = LRUCacheOperator.getSideJoinDataStream(adaptStream, sideTableInfo.getType(), localSqlPluginPath, typeInfo, joinInfo, sideJoinFieldInfo, sideTableInfo);
+                    dsOut = SideAsyncOperator.getSideJoinDataStream(adaptStream, sideTableInfo.getType(), localSqlPluginPath, typeInfo, joinInfo, sideJoinFieldInfo, sideTableInfo);
                 }
 
                 HashBasedTable<String, String, String> mappingTable = HashBasedTable.create();

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

@@ -26,7 +26,7 @@
 import com.dtstack.flink.sql.util.PluginUtil;
 
 /**
- * 流处理
+ * get specify side parser
  * Date: 2018/7/25
  * Company: www.dtstack.com
  * @author xuchao
@@ -36,12 +36,19 @@ public class StreamSideFactory {
 
     private static final String CURR_TYPE = "side";
 
-    public static AbsTableParser getSqlParser(String resultType, String sqlRootDir) throws Exception {
+    private static final String SIDE_DIR_TMPL = "%s%sside";
+
+    public static AbsTableParser getSqlParser(String pluginType, String sqlRootDir, String cacheType) throws Exception {
+
+        cacheType = cacheType == null ? "async" : cacheType;
+        String sideDir = String.format(SIDE_DIR_TMPL, pluginType, cacheType);
         ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
-        String pluginJarPath = PluginUtil.getJarFileDirPath(resultType + CURR_TYPE, sqlRootDir);
+        String pluginJarPath = PluginUtil.getJarFileDirPath(sideDir, sqlRootDir);
+
         DtClassLoader dtClassLoader = (DtClassLoader) classLoader;
         PluginUtil.addPluginJar(pluginJarPath, dtClassLoader);
-        String className = PluginUtil.getSqlParserClassName(resultType, CURR_TYPE);
+        String className = PluginUtil.getSqlParserClassName(pluginType, CURR_TYPE);
+
         Class<?> sideParser = dtClassLoader.loadClass(className);
         if(!AbsSideTableParser.class.isAssignableFrom(sideParser)){
             throw new RuntimeException("class " + sideParser.getName() + " not subClass of AbsSideTableParser");

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

@@ -17,9 +17,13 @@
  */
 
 
-package com.dtstack.flink.sql.side;
+package com.dtstack.flink.sql.side.operator;
 
 import com.dtstack.flink.sql.classloader.DtClassLoader;
+import com.dtstack.flink.sql.side.AsyncReqRow;
+import com.dtstack.flink.sql.side.FieldInfo;
+import com.dtstack.flink.sql.side.JoinInfo;
+import com.dtstack.flink.sql.side.SideTableInfo;
 import com.dtstack.flink.sql.util.PluginUtil;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.streaming.api.datastream.AsyncDataStream;
@@ -36,16 +40,18 @@
  * @author xuchao
  */
 
-public class LRUCacheOperator {
+public class SideAsyncOperator {
+
+    private static final String PATH_FORMAT = "%sasyncside";
 
     private static AsyncReqRow loadAsyncReq(String sideType, String sqlRootDir, RowTypeInfo rowTypeInfo,
-                                    JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) throws Exception {
+                                            JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) throws Exception {
         ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
-        String pathOfType = sideType + "side";
+        String pathOfType = String.format(PATH_FORMAT, sideType);
         String pluginJarPath = PluginUtil.getJarFileDirPath(pathOfType, sqlRootDir);
         DtClassLoader dtClassLoader = (DtClassLoader) classLoader;
         PluginUtil.addPluginJar(pluginJarPath, dtClassLoader);
-        String className = PluginUtil.getSqlSideClassName(sideType, "side");
+        String className = PluginUtil.getSqlSideClassName(sideType, "side", "Async");
         return dtClassLoader.loadClass(className).asSubclass(AsyncReqRow.class)
                 .getConstructor(RowTypeInfo.class, JoinInfo.class, List.class, SideTableInfo.class).newInstance(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
     }