调整side 结构

Author: zoudaokoulife

core/src/main/java/com/dtstack/flink/sql/enums/ECacheType.java

@@ -27,7 +27,7 @@
  * @author xuchao
  */
 public enum ECacheType {
-    NONE, LRU;
+    NONE, LRU, ALL;
 
     public static boolean isValid(String type){
         for(ECacheType tmpType : ECacheType.values()){

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

@@ -0,0 +1,45 @@
+/*
+ * 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.side;
+
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.streaming.api.datastream.DataStream;
+
+import java.util.List;
+
+/**
+ * get plugin which implement from RichFlatMapFunction
+ * Date: 2018/9/18
+ * Company: www.dtstack.com
+ *
+ * @author xuchao
+ */
+
+public class AllCacheOperator {
+
+    private static void loadFlatMap(){
+
+    }
+
+    public static DataStream getSideJoinDataStream(DataStream inputStream, String sideType, String sqlRootDir, RowTypeInfo rowTypeInfo, JoinInfo joinInfo,
+                                                   List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) throws Exception {
+        return null;
+    }
+}

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

@@ -0,0 +1,34 @@
+/*
+ * 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.side;
+
+import org.apache.calcite.interpreter.Row;
+import org.apache.flink.api.common.functions.RichFlatMapFunction;
+
+/**
+ * Reason:
+ * Date: 2018/9/18
+ * Company: www.dtstack.com
+ * @author xuchao
+ */
+
+public abstract class AllReqRow extends RichFlatMapFunction<Row, Row>{
+}

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

@@ -53,50 +53,24 @@
 
 public abstract class AsyncReqRow extends RichAsyncFunction<Row, Row> {
 
-    private static final Logger LOG = LoggerFactory.getLogger(AsyncReqRow.class);
-
     private static final long serialVersionUID = 2098635244857937717L;
 
-    protected RowTypeInfo rowTypeInfo;
-
-    protected List<FieldInfo> outFieldInfoList;
-
-    protected List<String> equalFieldList = Lists.newArrayList();
-
-    protected List<Integer> equalValIndex = Lists.newArrayList();
-
-    protected String sqlCondition = "";
-
-    protected String sideSelectFields = "";
-
-    protected JoinType joinType;
+    protected SideReqRow sideReqRow;
 
-    //key:Returns the value of the position, returns the index values ​​in the input data
-    protected Map<Integer, Integer> inFieldIndex = Maps.newHashMap();
-
-    protected Map<Integer, Integer> sideFieldIndex = Maps.newHashMap();
-
-    protected SideTableInfo sideTableInfo;
-
-    protected AbsSideCache sideCache;
-
-    public AsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList,
-                       SideTableInfo sideTableInfo){
-        this.rowTypeInfo = rowTypeInfo;
-        this.outFieldInfoList = outFieldInfoList;
-        this.joinType = joinInfo.getJoinType();
-        this.sideTableInfo = sideTableInfo;
-        parseSelectFields(joinInfo);
-        buildEqualInfo(joinInfo, sideTableInfo);
+    public AsyncReqRow(SideReqRow sideReqRow){
+        this.sideReqRow = sideReqRow;
     }
 
     private void initCache(){
+        SideTableInfo sideTableInfo = sideReqRow.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);
         }else{
             throw new RuntimeException("not support side cache with type:" + sideTableInfo.getCacheType());
         }
@@ -105,101 +79,22 @@ private void initCache(){
     }
 
     protected CacheObj getFromCache(String key){
-        return sideCache.getFromCache(key);
+        return sideReqRow.getSideCache().getFromCache(key);
     }
 
     protected void putCache(String key, CacheObj value){
-        sideCache.putCache(key, value);
+        sideReqRow.getSideCache().putCache(key, value);
     }
 
     protected boolean openCache(){
-        return sideCache != null;
+        return sideReqRow.getSideCache() != null;
     }
 
-    public void parseSelectFields(JoinInfo joinInfo){
-        String sideTableName = joinInfo.getSideTableName();
-        String nonSideTableName = joinInfo.getNonSideTable();
-        List<String> fields = Lists.newArrayList();
-
-        int sideIndex = 0;
-        for( int i=0; i<outFieldInfoList.size(); i++){
-            FieldInfo fieldInfo = outFieldInfoList.get(i);
-            if(fieldInfo.getTable().equalsIgnoreCase(sideTableName)){
-                fields.add(fieldInfo.getFieldName());
-                sideFieldIndex.put(i, sideIndex);
-                sideIndex++;
-            }else if(fieldInfo.getTable().equalsIgnoreCase(nonSideTableName)){
-                int nonSideIndex = rowTypeInfo.getFieldIndex(fieldInfo.getFieldName());
-                inFieldIndex.put(i, nonSideIndex);
-            }else{
-                throw new RuntimeException("unknown table " + fieldInfo.getTable());
-            }
-        }
-
-        if(fields.size() == 0){
-            throw new RuntimeException("select non field from table " +  sideTableName);
-        }
-
-        sideSelectFields = String.join(",", fields);
-    }
-
-    public abstract void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo);
-
-    public void dealOneEqualCon(SqlNode sqlNode, String sideTableName){
-        if(sqlNode.getKind() != SqlKind.EQUALS){
-            throw new RuntimeException("not equal operator.");
-        }
-
-        SqlIdentifier left = (SqlIdentifier)((SqlBasicCall)sqlNode).getOperands()[0];
-        SqlIdentifier right = (SqlIdentifier)((SqlBasicCall)sqlNode).getOperands()[1];
-
-        String leftTableName = left.getComponent(0).getSimple();
-        String leftField = left.getComponent(1).getSimple();
-
-        String rightTableName = right.getComponent(0).getSimple();
-        String rightField = right.getComponent(1).getSimple();
-
-        if(leftTableName.equalsIgnoreCase(sideTableName)){
-            equalFieldList.add(leftField);
-            int equalFieldIndex = -1;
-            for(int i=0; i<rowTypeInfo.getFieldNames().length; i++){
-                String fieldName = rowTypeInfo.getFieldNames()[i];
-                if(fieldName.equalsIgnoreCase(rightField)){
-                    equalFieldIndex = i;
-                }
-            }
-            if(equalFieldIndex == -1){
-                throw new RuntimeException("can't find equal field " + rightField);
-            }
-
-            equalValIndex.add(equalFieldIndex);
-
-        }else if(rightTableName.equalsIgnoreCase(sideTableName)){
-
-            equalFieldList.add(rightField);
-            int equalFieldIndex = -1;
-            for(int i=0; i<rowTypeInfo.getFieldNames().length; i++){
-                String fieldName = rowTypeInfo.getFieldNames()[i];
-                if(fieldName.equalsIgnoreCase(leftField)){
-                    equalFieldIndex = i;
-                }
-            }
-            if(equalFieldIndex == -1){
-                throw new RuntimeException("can't find equal field " + rightField);
-            }
-
-            equalValIndex.add(equalFieldIndex);
-
-        }else{
-            throw new RuntimeException("resolve equalFieldList error:" + sqlNode.toString());
-        }
-
-    }
 
     protected abstract Row fillData(Row input, Object sideInput);
 
     protected void dealMissKey(Row input, ResultFuture<Row> resultFuture){
-        if(joinType == JoinType.LEFT){
+        if(sideReqRow.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/LRUCacheOperator.java

@@ -0,0 +1,60 @@
+/*
+ * 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.side;
+
+import com.dtstack.flink.sql.classloader.DtClassLoader;
+import com.dtstack.flink.sql.util.PluginUtil;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.streaming.api.datastream.AsyncDataStream;
+import org.apache.flink.streaming.api.datastream.DataStream;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * fill data with lru cache
+ * get data from  External data source with async operator
+ * Date: 2018/9/18
+ * Company: www.dtstack.com
+ * @author xuchao
+ */
+
+public class LRUCacheOperator {
+
+    private static AsyncReqRow loadAsyncReq(String sideType, String sqlRootDir, RowTypeInfo rowTypeInfo,
+                                    JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) throws Exception {
+        ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+        String pathOfType = sideType + "side";
+        String pluginJarPath = PluginUtil.getJarFileDirPath(pathOfType, sqlRootDir);
+        DtClassLoader dtClassLoader = (DtClassLoader) classLoader;
+        PluginUtil.addPluginJar(pluginJarPath, dtClassLoader);
+        String className = PluginUtil.getSqlSideClassName(sideType, "side");
+        return dtClassLoader.loadClass(className).asSubclass(AsyncReqRow.class)
+                .getConstructor(RowTypeInfo.class, JoinInfo.class, List.class, SideTableInfo.class).newInstance(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
+    }
+
+    public static DataStream getSideJoinDataStream(DataStream inputStream, String sideType, String sqlRootDir, RowTypeInfo rowTypeInfo,  JoinInfo joinInfo,
+                                            List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) throws Exception {
+        AsyncReqRow asyncDbReq = loadAsyncReq(sideType, sqlRootDir, rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
+        //TODO How much should be set for the degree of parallelism? Timeout? capacity settings?
+        return AsyncDataStream.orderedWait(inputStream, asyncDbReq, 10000, TimeUnit.MILLISECONDS, 10)
+                .setParallelism(sideTableInfo.getParallelism());
+    }
+}