modify reqRow to a interface

Author: zoudaokoulife

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

@@ -37,7 +37,7 @@
  * @author xuchao
  */
 
-public abstract class AllReqRow extends RichFlatMapFunction<Row, Row>{
+public abstract class AllReqRow extends RichFlatMapFunction<Row, Row> implements ISideReqRow {
 
     protected SideInfo sideInfo;
 
@@ -48,8 +48,6 @@ public AllReqRow(SideInfo sideInfo){
 
     }
 
-    protected abstract Row fillData(Row input, Object sideInput);
-
     protected abstract void initCache() throws SQLException;
 
     protected abstract void reloadCache();

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

@@ -40,7 +40,7 @@
  * @author xuchao
  */
 
-public abstract class AsyncReqRow extends RichAsyncFunction<Row, Row> {
+public abstract class AsyncReqRow extends RichAsyncFunction<Row, Row> implements ISideReqRow {
 
     private static final long serialVersionUID = 2098635244857937717L;
 
@@ -79,9 +79,6 @@ protected boolean openCache(){
         return sideInfo.getSideCache() != null;
     }
 
-
-    protected abstract Row fillData(Row input, Object sideInput);
-
     protected void dealMissKey(Row input, ResultFuture<Row> resultFuture){
         if(sideInfo.getJoinType() == JoinType.LEFT){
             //Reserved left table data

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

@@ -0,0 +1,33 @@
+/*
+ * 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.types.Row;
+
+/**
+ *
+ * Date: 2018/12/4
+ * Company: www.dtstack.com
+ * @author xuchao
+ */
+public interface ISideReqRow {
+
+    Row fillData(Row input, Object sideInput);
+
+}

hbase/hbase-side/hbase-all-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAllReqRow.java

@@ -66,7 +66,7 @@ public HbaseAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo
     }
 
     @Override
-    protected Row fillData(Row input, Object sideInput) {
+    public Row fillData(Row input, Object sideInput) {
         Map<String, Object> sideInputList = (Map<String, Object>) sideInput;
         Row row = new Row(sideInfo.getOutFieldInfoList().size());
         for(Map.Entry<Integer, Integer> entry : sideInfo.getInFieldIndex().entrySet()){

hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAsyncReqRow.java

@@ -160,7 +160,7 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
     }
 
     @Override
-    protected Row fillData(Row input, Object sideInput){
+    public Row fillData(Row input, Object sideInput){
 
         List<Object> sideInputList = (List<Object>) sideInput;
         Row row = new Row(sideInfo.getOutFieldInfoList().size());

mongo/mongo-side/mongo-all-side/src/main/java/com/dtstack/flink/sql/side/mongo/MongoAllReqRow.java

@@ -80,7 +80,7 @@ public MongoAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo
     }
 
     @Override
-    protected Row fillData(Row input, Object sideInput) {
+    public Row fillData(Row input, Object sideInput) {
         Map<String, Object> cacheInfo = (Map<String, Object>) sideInput;
         Row row = new Row(sideInfo.getOutFieldInfoList().size());
         for (Map.Entry<Integer, Integer> entry : sideInfo.getInFieldIndex().entrySet()) {

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/all/RdbAllReqRow.java

@@ -67,7 +67,7 @@ public RdbAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo>
     }
 
     @Override
-    protected Row fillData(Row input, Object sideInput) {
+    public Row fillData(Row input, Object sideInput) {
         Map<String, Object> cacheInfo = (Map<String, Object>) sideInput;
         Row row = new Row(sideInfo.getOutFieldInfoList().size());
         for (Map.Entry<Integer, Integer> entry : sideInfo.getInFieldIndex().entrySet()) {

redis5/redis5-side/redis-all-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAllReqRow.java

@@ -19,6 +19,7 @@
 package com.dtstack.flink.sql.side.redis;
 
 import com.dtstack.flink.sql.side.*;
+import com.dtstack.flink.sql.side.redis.table.RedisSideReqRow;
 import com.dtstack.flink.sql.side.redis.table.RedisSideTableInfo;
 import org.apache.calcite.sql.JoinType;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
@@ -55,34 +56,16 @@ public class RedisAllReqRow extends AllReqRow{
 
     private AtomicReference<Map<String, Map<String, String>>> cacheRef = new AtomicReference<>();
 
+    private RedisSideReqRow redisSideReqRow;
+
     public RedisAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
         super(new RedisAllSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
+        this.redisSideReqRow = new RedisSideReqRow(super.sideInfo);
     }
 
     @Override
-    protected Row fillData(Row input, Object sideInput) {
-        Map<String, String> sideInputMap = (Map<String, String>) sideInput;
-        Row row = new Row(sideInfo.getOutFieldInfoList().size());
-        for(Map.Entry<Integer, Integer> entry : sideInfo.getInFieldIndex().entrySet()){
-            Object obj = input.getField(entry.getValue());
-            boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(entry.getValue()).getClass());
-
-            if(obj instanceof Timestamp && isTimeIndicatorTypeInfo){
-                obj = ((Timestamp)obj).getTime();
-            }
-            row.setField(entry.getKey(), obj);
-        }
-
-        for(Map.Entry<Integer, Integer> entry : sideInfo.getSideFieldIndex().entrySet()){
-            if(sideInputMap == null){
-                row.setField(entry.getKey(), null);
-            }else{
-                String key = sideInfo.getSideFieldNameIndex().get(entry.getKey());
-                row.setField(entry.getKey(), sideInputMap.get(key));
-            }
-        }
-
-        return row;
+    public Row fillData(Row input, Object sideInput) {
+        return redisSideReqRow.fillData(input, sideInput);
     }
 
     @Override

redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java

@@ -21,6 +21,7 @@
 import com.dtstack.flink.sql.enums.ECacheContentType;
 import com.dtstack.flink.sql.side.*;
 import com.dtstack.flink.sql.side.cache.CacheObj;
+import com.dtstack.flink.sql.side.redis.table.RedisSideReqRow;
 import com.dtstack.flink.sql.side.redis.table.RedisSideTableInfo;
 import io.lettuce.core.KeyValue;
 import io.lettuce.core.RedisClient;
@@ -60,9 +61,11 @@ public class RedisAsyncReqRow extends AsyncReqRow {
 
     private RedisSideTableInfo redisSideTableInfo;
 
+    private RedisSideReqRow redisSideReqRow;
 
     public RedisAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
         super(new RedisAsyncSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
+        redisSideReqRow = new RedisSideReqRow(super.sideInfo);
     }
 
     @Override
@@ -108,30 +111,8 @@ private void buildRedisClient(RedisSideTableInfo tableInfo){
     }
 
     @Override
-    protected Row fillData(Row input, Object sideInput) {
-        Map<String, String> keyValue = (Map<String, String>) sideInput;
-        Row row = new Row(sideInfo.getOutFieldInfoList().size());
-        for(Map.Entry<Integer, Integer> entry : sideInfo.getInFieldIndex().entrySet()){
-            Object obj = input.getField(entry.getValue());
-            boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(entry.getValue()).getClass());
-
-            if(obj instanceof Timestamp && isTimeIndicatorTypeInfo){
-                obj = ((Timestamp)obj).getTime();
-            }
-
-            row.setField(entry.getKey(), obj);
-        }
-
-        for(Map.Entry<Integer, Integer> entry : sideInfo.getSideFieldIndex().entrySet()){
-            if(keyValue == null){
-                row.setField(entry.getKey(), null);
-            }else{
-                String key = sideInfo.getSideFieldNameIndex().get(entry.getKey());
-                row.setField(entry.getKey(), keyValue.get(key));
-            }
-        }
-
-        return row;
+    public Row fillData(Row input, Object sideInput) {
+        return redisSideReqRow.fillData(input, sideInput);
     }
 
     @Override

redis5/redis5-side/redis-side-core/src/main/java/com/dtstack/flink/sql/side/redis/table/RedisSideReqRow.java

@@ -0,0 +1,72 @@
+/*
+ * 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.redis.table;
+
+import com.dtstack.flink.sql.side.ISideReqRow;
+import com.dtstack.flink.sql.side.SideInfo;
+import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
+import org.apache.flink.types.Row;
+
+import java.io.Serializable;
+import java.sql.Timestamp;
+import java.util.Map;
+
+/**
+ * redis fill row data
+ * Date: 2018/12/4
+ * Company: www.dtstack.com
+ * @author xuchao
+ */
+
+public class RedisSideReqRow implements ISideReqRow, Serializable {
+
+    private static final long serialVersionUID = 3751171828444748982L;
+
+    private SideInfo sideInfo;
+
+    public RedisSideReqRow(SideInfo sideInfo){
+        this.sideInfo = sideInfo;
+    }
+
+    @Override
+    public Row fillData(Row input, Object sideInput) {
+        Map<String, String> sideInputMap = (Map<String, String>) sideInput;
+        Row row = new Row(sideInfo.getOutFieldInfoList().size());
+        for(Map.Entry<Integer, Integer> entry : sideInfo.getInFieldIndex().entrySet()){
+            Object obj = input.getField(entry.getValue());
+            boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(entry.getValue()).getClass());
+
+            if(obj instanceof Timestamp && isTimeIndicatorTypeInfo){
+                obj = ((Timestamp)obj).getTime();
+            }
+            row.setField(entry.getKey(), obj);
+        }
+
+        for(Map.Entry<Integer, Integer> entry : sideInfo.getSideFieldIndex().entrySet()){
+            if(sideInputMap == null){
+                row.setField(entry.getKey(), null);
+            }else{
+                String key = sideInfo.getSideFieldNameIndex().get(entry.getKey());
+                row.setField(entry.getKey(), sideInputMap.get(key));
+            }
+        }
+
+        return row;
+    }
+}