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:https://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;
+ }
+}