Revert "维表join的 驱动流如果是restract-stream 需要过滤掉回溯的值"

This reverts commit 57a7c47

Author: zoudaokoulife

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -252,8 +252,7 @@ private static void registerTable(SqlTree sqlTree, StreamExecutionEnvironment en
 
                 RowTypeInfo typeInfo = new RowTypeInfo(adaptTable.getSchema().getTypes(), adaptTable.getSchema().getColumnNames());
                 DataStream adaptStream = tableEnv.toRetractStream(adaptTable, typeInfo)
-                        .filter((Tuple2<Boolean, Row> f0) -> f0.f0)
-                        .map((Tuple2<Boolean, Row> f0) -> f0.f1)
+                        .map((Tuple2<Boolean, Row> f0) -> { return f0.f1; })
                         .returns(typeInfo);
 
                 String fields = String.join(",", typeInfo.getFieldNames());

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

@@ -770,9 +770,9 @@ 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)
-                .filter((Tuple2<Boolean, Row> f0) -> f0.f0)
-                .map((Tuple2<Boolean, Row> f0) -> f0.f1)
-                .returns(Row.class);
+                                .map((Tuple2<Boolean, Row> f0) -> { return f0.f1; })
+                                .returns(Row.class);
+
 
         //join side table before keyby ===> Reducing the size of each dimension table cache of async
         if(sideTableInfo.isPartitionedJoin()){