retractstream to appendstream

maqi · maqi · commit 5c5a51900054 · 2019-07-04T15:50:59.000+08:00
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,8 @@ private void joinFun(Object pollObj, Map<String, Table> localTableCache,
         }
 
         RowTypeInfo typeInfo = new RowTypeInfo(targetTable.getSchema().getTypes(), targetTable.getSchema().getColumnNames());
-        DataStream adaptStream = tableEnv.toRetractStream(targetTable, org.apache.flink.types.Row.class);
+        //DataStream adaptStream = tableEnv.toRetractStream(targetTable, org.apache.flink.types.Row.class);
+        DataStream adaptStream = tableEnv.toAppendStream(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,8 @@ 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.toRetractStream(targetTable, org.apache.flink.types.Row.class);`
	`596`	`+ //DataStream adaptStream = tableEnv.toRetractStream(targetTable, org.apache.flink.types.Row.class);`
	`597`	`+ DataStream adaptStream = tableEnv.toAppendStream(targetTable, org.apache.flink.types.Row.class);`
`597`	`598`
`598`	`599`	`//join side table before keyby ===> Reducing the size of each dimension table cache of async`
`599`	`600`	`if(sideTableInfo.isPartitionedJoin()){`