DRILL-6327: Update unary operators to handle IterOutcome.EMIT

            Note: Handles for Non-Blocking Unary operators (like Filter/Project/etc) with EMIT Iter.Outcome

closes #1240

2 files changed, 5 insertions, 1 deletions

diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterRecordBatch.java
index f0b832a78..ac6d99fcc 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterRecordBatch.java
@@ -81,7 +81,7 @@ public class FilterRecordBatch extends AbstractSingleRecordBatch<Filter> {
       throw new UnsupportedOperationException(e);
     }
 
-    return IterOutcome.OK;
+    return getFinalOutcome(false);
   }
 
   @Override
@@ -168,6 +168,9 @@ public class FilterRecordBatch extends AbstractSingleRecordBatch<Filter> {
     final ErrorCollector collector = new ErrorCollectorImpl();
     final List<TransferPair> transfers = Lists.newArrayList();
     final ClassGenerator<Filterer> cg = CodeGenerator.getRoot(Filterer.TEMPLATE_DEFINITION2, context.getOptions());
+    // Uncomment below lines to enable saving generated code file for debugging
+    // cg.getCodeGenerator().plainJavaCapable(true);
+    // cg.getCodeGenerator().saveCodeForDebugging(true);
 
     final LogicalExpression expr = ExpressionTreeMaterializer.materialize(popConfig.getExpr(), incoming, collector,
             context.getFunctionRegistry(), false, unionTypeEnabled);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterTemplate2.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterTemplate2.java
index 52533bd50..6d1f03462 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterTemplate2.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterTemplate2.java
@@ -61,6 +61,7 @@ public abstract class FilterTemplate2 implements Filterer {
   @Override
   public void filterBatch(int recordCount) throws SchemaChangeException{
     if (recordCount == 0) {
+      outgoingSelectionVector.setRecordCount(0);
       return;
     }
     if (! outgoingSelectionVector.allocateNewSafe(recordCount)) {