Merge branch 'v1.5.0_dev_joingroup_bug' into 'v1.5.0_dev'

todd5167 · todd5167 · commit 1d4cd8935f6c · 2019-07-04T11:46:06.000+08:00
modify appendStream to retractStream



See merge request !27
diff --git a/core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java b/core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java
@@ -593,7 +593,7 @@ private void joinFun(Object pollObj, Map<String, Table> localTableCache,
         }
 
         RowTypeInfo typeInfo = new RowTypeInfo(targetTable.getSchema().getTypes(), targetTable.getSchema().getColumnNames());
-        DataStream adaptStream = tableEnv.toAppendStream(targetTable, org.apache.flink.types.Row.class);
+        DataStream adaptStream = tableEnv.toRetractStream(targetTable, org.apache.flink.types.Row.class);
 
         //join side table before keyby ===> Reducing the size of each dimension table cache of async
         if(sideTableInfo.isPartitionedJoin()){

Original file line number	Diff line number	Diff line change
`@@ -593,7 +593,7 @@ private void joinFun(Object pollObj, Map<String, Table> localTableCache,`
`593`	`593`	`}`
`594`	`594`
`595`	`595`	`RowTypeInfo typeInfo = new RowTypeInfo(targetTable.getSchema().getTypes(), targetTable.getSchema().getColumnNames());`
`596`		`- DataStream adaptStream = tableEnv.toAppendStream(targetTable, org.apache.flink.types.Row.class);`
	`596`	`+ DataStream adaptStream = tableEnv.toRetractStream(targetTable, org.apache.flink.types.Row.class);`
`597`	`597`
`598`	`598`	`//join side table before keyby ===> Reducing the size of each dimension table cache of async`
`599`	`599`	`if(sideTableInfo.isPartitionedJoin()){`