Repository: prestodb/presto
Issue: Race condition in array_concat

A query like:

``` sql
select count(x) from (
    select concat(a, b) x 
    from t
)
```

where `a` and `b` are of type `array(map(string, double))`, fails non-deterministically with various exceptions (Presto version 0.144.2):

```
java.lang.IllegalStateException: closeEntry called before anything is written
    at com.facebook.presto.spi.block.InterleavedBlockBuilder.closeEntry(InterleavedBlockBuilder.java:212)
    at com.facebook.presto.spi.block.ArrayElementBlockWriter.closeEntry(ArrayElementBlockWriter.java:127)
    at com.facebook.presto.spi.block.AbstractArrayBlock.writePositionTo(AbstractArrayBlock.java:155)
    at com.facebook.presto.type.MapType.appendTo(MapType.java:192)
    at com.facebook.presto.operator.scalar.ArrayConcatFunction.concat(ArrayConcatFunction.java:53)
    at com_facebook_presto_$gen_CursorProcessor_37019.project_0(Unknown Source)
    at com_facebook_presto_$gen_CursorProcessor_37019.process(Unknown Source)
    at com.facebook.presto.operator.ScanFilterAndProjectOperator.getOutput(ScanFilterAndProjectOperator.java:215)
    at com.facebook.presto.operator.Driver.processInternal(Driver.java:380)
    at com.facebook.presto.operator.Driver.processFor(Driver.java:303)
    at com.facebook.presto.execution.SqlTaskExecution$DriverSplitRunner.processFor(SqlTaskExecution.java:577)
    at com.facebook.presto.execution.TaskExecutor$PrioritizedSplitRunner.process(TaskExecutor.java:529)
    at com.facebook.presto.execution.TaskExecutor$Runner.run(TaskExecutor.java:665)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
```

```
Query 20160425_183530_20138_n949w failed: Expected current entry size to be exactly 0 but was 0
java.lang.IllegalStateException: Expected current entry size to be exactly 0 but was 0
    at com.facebook.presto.spi.block.ArrayBlockBuilder.beginBlockEntry(ArrayBlockBuilder.java:193)
    at com.facebook.presto.spi.block.ArrayBlockBuilder.beginBlockEntry(ArrayBlockBuilder.java:24)
    at com.facebook.presto.spi.block.AbstractArrayBlock.writePositionTo(AbstractArrayBlock.java:146)
    at com.facebook.presto.type.MapType.appendTo(MapType.java:192)
    at com.facebook.presto.operator.scalar.ArrayConcatFunction.concat(ArrayConcatFunction.java:53)
    at com_facebook_presto_$gen_CursorProcessor_36742.project_0(Unknown Source)
    at com_facebook_presto_$gen_CursorProcessor_36742.process(Unknown Source)
    at com.facebook.presto.operator.ScanFilterAndProjectOperator.getOutput(ScanFilterAndProjectOperator.java:215)
    at com.facebook.presto.operator.Driver.processInternal(Driver.java:380)
    at com.facebook.presto.operator.Driver.processFor(Driver.java:303)
    at com.facebook.presto.execution.SqlTaskExecution$DriverSplitRunner.processFor(SqlTaskExecution.java:577)
    at com.facebook.presto.execution.TaskExecutor$PrioritizedSplitRunner.process(TaskExecutor.java:529)
    at com.facebook.presto.execution.TaskExecutor$Runner.run(TaskExecutor.java:665)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
```

```
Query 20160425_183539_20170_n949w failed: Expected current entry size to be exactly 0 but was 1
java.lang.IllegalStateException: Expected current entry size to be exactly 0 but was 1
    at com.facebook.presto.spi.block.ArrayBlockBuilder.beginBlockEntry(ArrayBlockBuilder.java:193)
    at com.facebook.presto.spi.block.ArrayBlockBuilder.beginBlockEntry(ArrayBlockBuilder.java:24)
    at com.facebook.presto.spi.block.AbstractArrayBlock.writePositionTo(AbstractArrayBlock.java:146)
    at com.facebook.presto.type.MapType.appendTo(MapType.java:192)
    at com.facebook.presto.operator.scalar.ArrayConcatFunction.concat(ArrayConcatFunction.java:53)
    at com_facebook_presto_$gen_CursorProcessor_37840.project_0(Unknown Source)
    at com_facebook_presto_$gen_CursorProcessor_37840.process(Unknown Source)
    at com.facebook.presto.operator.ScanFilterAndProjectOperator.getOutput(ScanFilterAndProjectOperator.java:215)
    at com.facebook.presto.operator.Driver.processInternal(Driver.java:380)
    at com.facebook.presto.operator.Driver.processFor(Driver.java:303)
    at com.facebook.presto.execution.SqlTaskExecution$DriverSplitRunner.processFor(SqlTaskExecution.java:577)
    at com.facebook.presto.execution.TaskExecutor$PrioritizedSplitRunner.process(TaskExecutor.java:529)
    at com.facebook.presto.execution.TaskExecutor$Runner.run(TaskExecutor.java:665)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
```


Changed files:
presto-main/src/main/java/com/facebook/presto/sql/gen/ExpressionCompiler.java
presto-main/src/main/java/com/facebook/presto/sql/planner/LocalExecutionPlanner.java
presto-main/src/main/java/com/facebook/presto/operator/ScanFilterAndProjectOperator.java
presto-main/src/test/java/com/facebook/presto/operator/scalar/FunctionAssertions.java
presto-hive/src/test/java/com/facebook/presto/hive/TestOrcPageSourceMemoryTracking.java
presto-main/src/test/java/com/facebook/presto/operator/TestScanFilterAndProjectOperator.java

diff --git a/presto-hive/src/test/java/com/facebook/presto/hive/TestOrcPageSourceMemoryTracking.java b/presto-hive/src/test/java/com/facebook/presto/hive/TestOrcPageSourceMemoryTracking.java
index 05aad9aeca..db5ca3f689 100644
--- a/presto-hive/src/test/java/com/facebook/presto/hive/TestOrcPageSourceMemoryTracking.java
+++ b/presto-hive/src/test/java/com/facebook/presto/hive/TestOrcPageSourceMemoryTracking.java
@@ -419,7 +419,7 @@ public class TestOrcPageSourceMemoryTracking
                     new PlanNodeId("test"),
                     new PlanNodeId("0"),
                     (session, split, columnHandles) -> pageSource,
-                    new GenericCursorProcessor(FilterFunctions.TRUE_FUNCTION, projections),
+                    () -> new GenericCursorProcessor(FilterFunctions.TRUE_FUNCTION, projections),
                     () -> new GenericPageProcessor(FilterFunctions.TRUE_FUNCTION, projections),
                     columns.stream().map(columnHandle -> (ColumnHandle) columnHandle).collect(toList()),
                     types
diff --git a/presto-main/src/main/java/com/facebook/presto/operator/ScanFilterAndProjectOperator.java b/presto-main/src/main/java/com/facebook/presto/operator/ScanFilterAndProjectOperator.java
index 0518c3e982..4492c1d74c 100644
--- a/presto-main/src/main/java/com/facebook/presto/operator/ScanFilterAndProjectOperator.java
+++ b/presto-main/src/main/java/com/facebook/presto/operator/ScanFilterAndProjectOperator.java
@@ -299,7 +299,7 @@ public class ScanFilterAndProjectOperator
     {
         private final int operatorId;
         private final PlanNodeId planNodeId;
-        private final CursorProcessor cursorProcessor;
+        private final Supplier<CursorProcessor> cursorProcessor;
         private final Supplier<PageProcessor> pageProcessor;
         private final PlanNodeId sourceId;
         private final PageSourceProvider pageSourceProvider;
@@ -312,7 +312,7 @@ public class ScanFilterAndProjectOperator
                 PlanNodeId planNodeId,
                 PlanNodeId sourceId,
                 PageSourceProvider pageSourceProvider,
-                CursorProcessor cursorProcessor,
+                Supplier<CursorProcessor> cursorProcessor,
                 Supplier<PageProcessor> pageProcessor,
                 Iterable<ColumnHandle> columns,
                 List<Type> types)
@@ -348,7 +348,7 @@ public class ScanFilterAndProjectOperator
                     operatorContext,
                     sourceId,
                     pageSourceProvider,
-                    cursorProcessor,
+                    cursorProcessor.get(),
                     pageProcessor.get(),
                     columns,
                     types);
diff --git a/presto-main/src/main/java/com/facebook/presto/sql/gen/ExpressionCompiler.java b/presto-main/src/main/java/com/facebook/presto/sql/gen/ExpressionCompiler.java
index aad8240bb1..ec012e2b2e 100644
--- a/presto-main/src/main/java/com/facebook/presto/sql/gen/ExpressionCompiler.java
+++ b/presto-main/src/main/java/com/facebook/presto/sql/gen/ExpressionCompiler.java
@@ -81,15 +81,17 @@ public class ExpressionCompiler
         return pageProcessors.size();
     }
 
-    public CursorProcessor compileCursorProcessor(RowExpression filter, List<RowExpression> projections, Object uniqueKey)
+    public Supplier<CursorProcessor> compileCursorProcessor(RowExpression filter, List<RowExpression> projections, Object uniqueKey)
     {
-        try {
-            return cursorProcessors.getUnchecked(new CacheKey(filter, projections, uniqueKey))
-                    .newInstance();
-        }
-        catch (ReflectiveOperationException e) {
-            throw Throwables.propagate(e);
-        }
+        Class<? extends CursorProcessor> cursorProcessor = cursorProcessors.getUnchecked(new CacheKey(filter, projections, uniqueKey));
+        return () -> {
+            try {
+                return cursorProcessor.newInstance();
+            }
+            catch (ReflectiveOperationException e) {
+                throw Throwables.propagate(e);
+            }
+        };
     }
 
     public Supplier<PageProcessor> compilePageProcessor(RowExpression filter, List<RowExpression> projections)
diff --git a/presto-main/src/main/java/com/facebook/presto/sql/planner/LocalExecutionPlanner.java b/presto-main/src/main/java/com/facebook/presto/sql/planner/LocalExecutionPlanner.java
index f98890a7e4..f4e8e6e6eb 100644
--- a/presto-main/src/main/java/com/facebook/presto/sql/planner/LocalExecutionPlanner.java
+++ b/presto-main/src/main/java/com/facebook/presto/sql/planner/LocalExecutionPlanner.java
@@ -978,7 +978,7 @@ public class LocalExecutionPlanner
 
             try {
                 if (columns != null) {
-                    CursorProcessor cursorProcessor = compiler.compileCursorProcessor(translatedFilter, translatedProjections, sourceNode.getId());
+                    Supplier<CursorProcessor> cursorProcessor = compiler.compileCursorProcessor(translatedFilter, translatedProjections, sourceNode.getId());
                     Supplier<PageProcessor> pageProcessor = compiler.compilePageProcessor(translatedFilter, translatedProjections);
 
                     SourceOperatorFactory operatorFactory = new ScanFilterAndProjectOperator.ScanFilterAndProjectOperatorFactory(
@@ -1050,7 +1050,7 @@ public class LocalExecutionPlanner
                         planNodeId,
                         sourceNode.getId(),
                         pageSourceProvider,
-                        new GenericCursorProcessor(filterFunction, projectionFunctions),
+                        () -> new GenericCursorProcessor(filterFunction, projectionFunctions),
                         () -> new GenericPageProcessor(filterFunction, projectionFunctions),
                         columns,
                         toTypes(projectionFunctions));
diff --git a/presto-main/src/test/java/com/facebook/presto/operator/TestScanFilterAndProjectOperator.java b/presto-main/src/test/java/com/facebook/presto/operator/TestScanFilterAndProjectOperator.java
index 1ed44ce165..d9ccc89e6f 100644
--- a/presto-main/src/test/java/com/facebook/presto/operator/TestScanFilterAndProjectOperator.java
+++ b/presto-main/src/test/java/com/facebook/presto/operator/TestScanFilterAndProjectOperator.java
@@ -71,7 +71,7 @@ public class TestScanFilterAndProjectOperator
                         return new FixedPageSource(ImmutableList.of(input));
                     }
                 },
-                new GenericCursorProcessor(FilterFunctions.TRUE_FUNCTION, ImmutableList.of(singleColumn(VARCHAR, 0))),
+                () -> new GenericCursorProcessor(FilterFunctions.TRUE_FUNCTION, ImmutableList.of(singleColumn(VARCHAR, 0))),
                 () -> new GenericPageProcessor(FilterFunctions.TRUE_FUNCTION, ImmutableList.of(singleColumn(VARCHAR, 0))),
                 ImmutableList.<ColumnHandle>of(),
                 ImmutableList.<Type>of(VARCHAR));
@@ -105,7 +105,7 @@ public class TestScanFilterAndProjectOperator
                         return new RecordPageSource(new PageRecordSet(ImmutableList.<Type>of(VARCHAR), input));
                     }
                 },
-                new GenericCursorProcessor(FilterFunctions.TRUE_FUNCTION, ImmutableList.of(singleColumn(VARCHAR, 0))),
+                () -> new GenericCursorProcessor(FilterFunctions.TRUE_FUNCTION, ImmutableList.of(singleColumn(VARCHAR, 0))),
                 () -> new GenericPageProcessor(FilterFunctions.TRUE_FUNCTION, ImmutableList.of(singleColumn(VARCHAR, 0))),
                 ImmutableList.<ColumnHandle>of(),
                 ImmutableList.<Type>of(VARCHAR));
diff --git a/presto-main/src/test/java/com/facebook/presto/operator/scalar/FunctionAssertions.java b/presto-main/src/test/java/com/facebook/presto/operator/scalar/FunctionAssertions.java
index 0c394c94df..b478b32aa8 100644
--- a/presto-main/src/test/java/com/facebook/presto/operator/scalar/FunctionAssertions.java
+++ b/presto-main/src/test/java/com/facebook/presto/operator/scalar/FunctionAssertions.java
@@ -608,7 +608,7 @@ public final class FunctionAssertions
                 SQL_PARSER, INPUT_TYPES, ImmutableList.of(filter, projection));
 
         try {
-            CursorProcessor cursorProcessor = compiler.compileCursorProcessor(
+            Supplier<CursorProcessor> cursorProcessor = compiler.compileCursorProcessor(
                     toRowExpression(filter, expressionTypes),
                     ImmutableList.of(toRowExpression(projection, expressionTypes)),
                     SOURCE_ID);