Merge branch '1.8_test_3.1.0x_mergeElasticsearch6-side' into '1.8_test_3.10.x'

添加elasticsearch6-side中缺的sendOutputRow()方法



See merge request !240

Author: yinxi

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

@@ -23,8 +23,11 @@
 import org.apache.flink.api.common.functions.RichFlatMapFunction;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.runtime.types.CRow;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Collector;
 
 import com.dtstack.flink.sql.factory.DTThreadFactory;
+import org.apache.calcite.sql.JoinType;
 
 import java.sql.SQLException;
 import java.util.concurrent.Executors;
@@ -65,4 +68,13 @@ public void open(Configuration parameters) throws Exception {
         es.scheduleAtFixedRate(() -> reloadCache(), sideTableInfo.getCacheTimeout(), sideTableInfo.getCacheTimeout(), TimeUnit.MILLISECONDS);
     }
 
+    protected void sendOutputRow(CRow value, Object sideInput, Collector<CRow> out){
+        if(sideInput == null && sideInfo.getJoinType() != JoinType.LEFT){
+            return;
+        }
+
+        Row row = fillData(value.row(), sideInput);
+        out.collect(new CRow(row, value.change()));
+    }
+
 }