Revert "replace exchange with fetchAll"

This reverts commit 8606d76.

Author: aqni

optimizer/src/main/java/cn/edu/tsinghua/iginx/physical/optimizer/naive/NaivePhysicalPlanner.java

@@ -21,7 +21,6 @@
 
 import cn.edu.tsinghua.iginx.conf.ConfigDescriptor;
 import cn.edu.tsinghua.iginx.engine.physical.memory.execute.executor.unary.stateful.AddSequenceExecutor;
-import cn.edu.tsinghua.iginx.engine.physical.memory.execute.executor.unary.stateful.FetchAllUnaryExecutor;
 import cn.edu.tsinghua.iginx.engine.physical.memory.execute.executor.unary.stateful.LimitUnaryExecutor;
 import cn.edu.tsinghua.iginx.engine.physical.memory.execute.executor.unary.stateful.RemoveNullColumnExecutor;
 import cn.edu.tsinghua.iginx.engine.physical.memory.execute.executor.unary.stateless.ValueToSelectedPathExecutor;
@@ -183,13 +182,13 @@ public PhysicalTask<BatchStream> construct(
   public PhysicalTask<BatchStream> fetchAsync(Source source, RequestContext context) {
     PhysicalTask<?> sourceTask = fetch(source, context);
     // TODO: 是否所有的情况都应当使用 Exchange？ 是否存在一些情况使用 FetchAll 更好？
-     return new UnarySinkMemoryPhysicalTask(
-         convert(sourceTask, context, BatchStream.class),
-         Collections.emptyList(),
-         context,
-         (ctx, schema) -> new FetchAllUnaryExecutor(ctx, schema.raw()));
-//    return new FetchAsyncMemoryPhysicalTask(
-//        convert(sourceTask, context, BatchStream.class), context);
+    // return new UnarySinkMemoryPhysicalTask(
+    //     convert(sourceTask, context, BatchStream.class),
+    //     Collections.emptyList(),
+    //     context,
+    //     (ctx, schema) -> new FetchAllUnaryExecutor(ctx, schema.raw()));
+    return new FetchAsyncMemoryPhysicalTask(
+        convert(sourceTask, context, BatchStream.class), context);
   }
 
   public PhysicalTask<BatchStream> construct(CombineNonQuery operator, RequestContext context) {