Optimize execution of cross join

presto-main/src/main/java/com/facebook/presto/block/BlockEncodingManager.java

@@ -24,6 +24,7 @@
 import com.facebook.presto.spi.block.SliceArrayBlockEncoding;
 import com.facebook.presto.spi.block.VariableWidthBlockEncoding;
 import com.facebook.presto.spi.type.TypeManager;
+import com.facebook.presto.testing.RunLengthBlockEncoding;
 import com.google.common.collect.ImmutableSet;
 import io.airlift.slice.SliceInput;
 import io.airlift.slice.SliceOutput;
@@ -64,6 +65,7 @@ public BlockEncodingManager(TypeManager typeManager, Set<BlockEncodingFactory<?>
         addBlockEncodingFactory(DictionaryBlockEncoding.FACTORY);
         addBlockEncodingFactory(ArrayBlockEncoding.FACTORY);
         addBlockEncodingFactory(InterleavedBlockEncoding.FACTORY);
+        addBlockEncodingFactory(RunLengthBlockEncoding.FACTORY);
 
         for (BlockEncodingFactory<?> factory : requireNonNull(blockEncodingFactories, "blockEncodingFactories is null")) {
             addBlockEncodingFactory(factory);

presto-main/src/main/java/com/facebook/presto/operator/NestedLoopBuildOperator.java

@@ -0,0 +1,145 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.facebook.presto.operator;
+
+import com.facebook.presto.spi.Page;
+import com.facebook.presto.spi.type.Type;
+
+import java.util.List;
+
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
+
+public class NestedLoopBuildOperator
+        implements Operator
+{
+    public static class NestedLoopBuildOperatorFactory
+            implements OperatorFactory
+    {
+        private final int operatorId;
+        private final NestedLoopJoinPagesSupplier nestedLoopJoinPagesSupplier;
+
+        private boolean closed;
+
+        public NestedLoopBuildOperatorFactory(int operatorId, List<Type> types)
+        {
+            this.operatorId = operatorId;
+            nestedLoopJoinPagesSupplier = new NestedLoopJoinPagesSupplier(requireNonNull(types, "types is null"));
+            nestedLoopJoinPagesSupplier.retain();
+        }
+
+        public NestedLoopJoinPagesSupplier getNestedLoopJoinPagesSupplier()
+        {
+            return nestedLoopJoinPagesSupplier;
+        }
+
+        @Override
+        public List<Type> getTypes()
+        {
+            return nestedLoopJoinPagesSupplier.getTypes();
+        }
+
+        @Override
+        public Operator createOperator(DriverContext driverContext)
+        {
+            checkState(!closed, "Factory is already closed");
+            OperatorContext operatorContext = driverContext.addOperatorContext(operatorId, NestedLoopBuildOperator.class.getSimpleName());
+            return new NestedLoopBuildOperator(operatorContext, nestedLoopJoinPagesSupplier);
+        }
+
+        @Override
+        public void close()
+        {
+            if (closed) {
+                return;
+            }
+            closed = true;
+            nestedLoopJoinPagesSupplier.release();
+        }
+    }
+
+    private final OperatorContext operatorContext;
+    private final NestedLoopJoinPagesSupplier nestedLoopJoinPagesSupplier;
+    private final NestedLoopJoinPagesBuilder nestedLoopJoinPagesBuilder;
+    private boolean finished;
+
+    public NestedLoopBuildOperator(OperatorContext operatorContext, NestedLoopJoinPagesSupplier nestedLoopJoinPagesSupplier)
+    {
+        this.operatorContext = requireNonNull(operatorContext, "operatorContext is null");
+        this.nestedLoopJoinPagesSupplier = requireNonNull(nestedLoopJoinPagesSupplier, "nestedLoopJoinPagesSupplier is null");
+        this.nestedLoopJoinPagesBuilder = new NestedLoopJoinPagesBuilder(operatorContext.getDriverContext().getPipelineContext().getTaskContext());
+    }
+
+    @Override
+    public OperatorContext getOperatorContext()
+    {
+        return operatorContext;
+    }
+
+    @Override
+    public List<Type> getTypes()
+    {
+        return nestedLoopJoinPagesSupplier.getTypes();
+    }
+
+    @Override
+    public void finish()
+    {
+        if (finished) {
+            return;
+        }
+
+        // Free memory, as the PageSource is going to take it over
+        operatorContext.setMemoryReservation(0);
+        nestedLoopJoinPagesSupplier.setPages(nestedLoopJoinPagesBuilder.build());
+
+        finished = true;
+    }
+
+    @Override
+    public boolean isFinished()
+    {
+        return finished;
+    }
+
+    @Override
+    public boolean needsInput()
+    {
+        return !finished;
+    }
+
+    @Override
+    public void addInput(Page page)
+    {
+        requireNonNull(page, "page is null");
+        checkState(!isFinished(), "Operator is already finished");
+
+        if (page.getPositionCount() == 0) {
+            return;
+        }
+
+        nestedLoopJoinPagesBuilder.addPage(page);
+        if (!operatorContext.trySetMemoryReservation(nestedLoopJoinPagesBuilder.getEstimatedSize().toBytes())) {
+            nestedLoopJoinPagesBuilder.compact();
+        }
+        operatorContext.setMemoryReservation(nestedLoopJoinPagesBuilder.getEstimatedSize().toBytes());
+        operatorContext.recordGeneratedOutput(page.getSizeInBytes(), page.getPositionCount());
+    }
+
+    @Override
+    public Page getOutput()
+    {
+        return null;
+    }
+}