Encapsulate InternalJoinFilterFunction in JoinFilterFunctionCompiler

InternalJoinFilterFunction is an internal detail of JoinFilterFunction and
should not be exposed to callers. The JoinFilterFuction implementation has
direct access the build pages for performance reasons, and thus can not be
created until the end of the build.  A new JoinFilterFunctionFactory is
added to support this delayed creation.

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

@@ -15,6 +15,7 @@
 
 import com.facebook.presto.spi.Page;
 import com.facebook.presto.spi.type.Type;
+import com.facebook.presto.sql.gen.JoinFilterFunctionCompiler.JoinFilterFunctionFactory;
 import com.facebook.presto.sql.planner.Symbol;
 import com.facebook.presto.sql.planner.plan.PlanNodeId;
 import com.google.common.collect.ImmutableList;
@@ -44,7 +45,7 @@ private enum State {
         private final SettableLookupSourceSupplier lookupSourceSupplier;
         private final List<Integer> hashChannels;
         private final Optional<Integer> hashChannel;
-        private final Optional<InternalJoinFilterFunction> filterFunction;
+        private final Optional<JoinFilterFunctionFactory> filterFunctionFactory;
 
         private final int expectedPositions;
         private State state = State.NOT_CREATED;
@@ -57,7 +58,7 @@ public HashBuilderOperatorFactory(
                 List<Integer> hashChannels,
                 Optional<Integer> hashChannel,
                 boolean outer,
-                Optional<InternalJoinFilterFunction> filterFunction,
+                Optional<JoinFilterFunctionFactory> filterFunctionFactory,
                 int expectedPositions)
         {
             this.operatorId = operatorId;
@@ -69,7 +70,7 @@ public HashBuilderOperatorFactory(
 
             this.hashChannels = ImmutableList.copyOf(requireNonNull(hashChannels, "hashChannels is null"));
             this.hashChannel = requireNonNull(hashChannel, "hashChannel is null");
-            this.filterFunction = requireNonNull(filterFunction, "filterFunction is null");
+            this.filterFunctionFactory = requireNonNull(filterFunctionFactory, "filterFunctionFactory is null");
 
             this.expectedPositions = expectedPositions;
         }
@@ -97,7 +98,7 @@ public Operator createOperator(DriverContext driverContext)
                     lookupSourceSupplier,
                     hashChannels,
                     hashChannel,
-                    filterFunction,
+                    filterFunctionFactory,
                     expectedPositions);
         }
 
@@ -118,7 +119,7 @@ public OperatorFactory duplicate()
     private final SettableLookupSourceSupplier lookupSourceSupplier;
     private final List<Integer> hashChannels;
     private final Optional<Integer> hashChannel;
-    private final Optional<InternalJoinFilterFunction> filterFunction;
+    private final Optional<JoinFilterFunctionFactory> filterFunctionFactory;
 
     private final PagesIndex pagesIndex;
 
@@ -129,7 +130,7 @@ public HashBuilderOperator(
             SettableLookupSourceSupplier lookupSourceSupplier,
             List<Integer> hashChannels,
             Optional<Integer> hashChannel,
-            Optional<InternalJoinFilterFunction> filterFunction,
+            Optional<JoinFilterFunctionFactory> filterFunctionFactory,
             int expectedPositions)
     {
         this.operatorContext = requireNonNull(operatorContext, "operatorContext is null");
@@ -138,7 +139,7 @@ public HashBuilderOperator(
 
         this.hashChannels = ImmutableList.copyOf(requireNonNull(hashChannels, "hashChannels is null"));
         this.hashChannel = requireNonNull(hashChannel, "hashChannel is null");
-        this.filterFunction = requireNonNull(filterFunction, "filterFunction is null");
+        this.filterFunctionFactory = requireNonNull(filterFunctionFactory, "filterFunctionFactory is null");
 
         this.pagesIndex = new PagesIndex(lookupSourceSupplier.getTypes(), expectedPositions);
     }
@@ -163,7 +164,7 @@ public void finish()
         }
 
         // After this point the LookupSource will take over our memory reservation, and ours will be zero
-        LookupSource lookupSource = pagesIndex.createLookupSource(hashChannels, hashChannel, filterFunction);
+        LookupSource lookupSource = pagesIndex.createLookupSource(operatorContext.getSession(), hashChannels, hashChannel, filterFunctionFactory);
         lookupSourceSupplier.setLookupSource(lookupSource, operatorContext);
         finished = true;
     }

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

@@ -13,13 +13,15 @@
  */
 package com.facebook.presto.operator;
 
+import com.facebook.presto.Session;
 import com.facebook.presto.spi.Page;
 import com.facebook.presto.spi.PageBuilder;
 import com.facebook.presto.spi.block.Block;
 import com.facebook.presto.spi.block.BlockBuilder;
 import com.facebook.presto.spi.block.SortOrder;
 import com.facebook.presto.spi.type.Type;
 import com.facebook.presto.sql.gen.JoinCompiler;
+import com.facebook.presto.sql.gen.JoinFilterFunctionCompiler.JoinFilterFunctionFactory;
 import com.facebook.presto.sql.gen.OrderingCompiler;
 import com.google.common.collect.ImmutableList;
 import io.airlift.log.Logger;
@@ -302,9 +304,9 @@ private PagesIndexOrdering createPagesIndexComparator(List<Integer> sortChannels
         return orderingCompiler.compilePagesIndexOrdering(sortTypes, sortChannels, sortOrders);
     }
 
-    public LookupSource createLookupSource(List<Integer> joinChannels)
+    public LookupSource createLookupSource(Session session, List<Integer> joinChannels)
     {
-        return createLookupSource(joinChannels, Optional.empty(), Optional.empty());
+        return createLookupSource(session, joinChannels, Optional.empty(), Optional.empty());
     }
 
     public PagesHashStrategy createPagesHashStrategy(List<Integer> joinChannels, Optional<Integer> hashChannel)
@@ -321,13 +323,12 @@ public PagesHashStrategy createPagesHashStrategy(List<Integer> joinChannels, Opt
         return new SimplePagesHashStrategy(types, ImmutableList.<List<Block>>copyOf(channels), joinChannels, hashChannel);
     }
 
-    private Optional<JoinFilterFunction> createJoinFilterFunction(Optional<InternalJoinFilterFunction> filterFunctionOptional, List<List<Block>> channels)
+    public LookupSource createLookupSource(Session session, List<Integer> joinChannels, Optional<Integer> hashChannel, Optional<JoinFilterFunctionFactory> filterFunctionFactory)
     {
-        return filterFunctionOptional.map(filterFunction -> joinCompiler.compileJoinFilterFunctionFactory(filterFunction).create(filterFunction, channels));
-    }
+        Optional<JoinFilterFunction> joinFilterFunction = filterFunctionFactory.map(factory -> factory.create(
+                session.toConnectorSession(),
+                ImmutableList.copyOf(channels)));
 
-    public LookupSource createLookupSource(List<Integer> joinChannels, Optional<Integer> hashChannel, Optional<InternalJoinFilterFunction> filterFunction)
-    {
         if (!joinChannels.isEmpty()) {
             // todo compiled implementation of lookup join does not support when we are joining with empty join channels.
             // This code path will trigger only for OUTER joins. To fix that we need to add support for
@@ -340,7 +341,7 @@ public LookupSource createLookupSource(List<Integer> joinChannels, Optional<Inte
                         valueAddresses,
                         ImmutableList.copyOf(channels),
                         hashChannel,
-                        createJoinFilterFunction(filterFunction, ImmutableList.copyOf(channels)));
+                        joinFilterFunction);
             }
             catch (Exception e) {
                 log.error(e, "Lookup source compile failed for types=%s error=%s", types, e);
@@ -355,7 +356,7 @@ public LookupSource createLookupSource(List<Integer> joinChannels, Optional<Inte
                 hashChannel
         );
 
-        return new InMemoryJoinHash(valueAddresses, hashStrategy, createJoinFilterFunction(filterFunction, ImmutableList.copyOf(channels)));
+        return new InMemoryJoinHash(valueAddresses, hashStrategy, joinFilterFunction);
     }
 
     @Override

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

@@ -15,6 +15,7 @@
 
 import com.facebook.presto.spi.Page;
 import com.facebook.presto.spi.type.Type;
+import com.facebook.presto.sql.gen.JoinFilterFunctionCompiler.JoinFilterFunctionFactory;
 import com.facebook.presto.sql.planner.Symbol;
 import com.facebook.presto.sql.planner.plan.PlanNodeId;
 import com.google.common.collect.ImmutableList;
@@ -43,7 +44,7 @@ public static class ParallelHashBuildOperatorFactory
         private final PartitionedLookupSourceSupplier lookupSourceSupplier;
         private final List<Integer> hashChannels;
         private final Optional<Integer> preComputedHashChannel;
-        private final Optional<InternalJoinFilterFunction> filterFunction;
+        private final Optional<JoinFilterFunctionFactory> filterFunctionFactory;
 
         private final int expectedPositions;
 
@@ -58,7 +59,7 @@ public ParallelHashBuildOperatorFactory(
                 List<Integer> hashChannels,
                 Optional<Integer> preComputedHashChannel,
                 boolean outer,
-                Optional<InternalJoinFilterFunction> filterFunction,
+                Optional<JoinFilterFunctionFactory> filterFunctionFactory,
                 int expectedPositions,
                 int partitionCount)
         {
@@ -76,7 +77,7 @@ public ParallelHashBuildOperatorFactory(
             checkArgument(!hashChannels.isEmpty(), "hashChannels is empty");
             this.hashChannels = ImmutableList.copyOf(requireNonNull(hashChannels, "hashChannels is null"));
             this.preComputedHashChannel = requireNonNull(preComputedHashChannel, "preComputedHashChannel is null");
-            this.filterFunction = requireNonNull(filterFunction, "filterFunction is null");
+            this.filterFunctionFactory = requireNonNull(filterFunctionFactory, "filterFunctionFactory is null");
 
             this.expectedPositions = expectedPositions;
         }
@@ -103,7 +104,7 @@ public Operator createOperator(DriverContext driverContext)
                     partitionIndex,
                     hashChannels,
                     preComputedHashChannel,
-                    filterFunction,
+                    filterFunctionFactory,
                     expectedPositions);
 
             partitionIndex++;
@@ -129,7 +130,7 @@ public OperatorFactory duplicate()
 
     private final List<Integer> hashChannels;
     private final Optional<Integer> preComputedHashChannel;
-    private final Optional<InternalJoinFilterFunction> filterFunction;
+    private final Optional<JoinFilterFunctionFactory> filterFunctionFactory;
 
     private final PagesIndex index;
 
@@ -141,12 +142,12 @@ public ParallelHashBuildOperator(
             int partitionIndex,
             List<Integer> hashChannels,
             Optional<Integer> preComputedHashChannel,
-            Optional<InternalJoinFilterFunction> filterFunction,
+            Optional<JoinFilterFunctionFactory> filterFunctionFactory,
             int expectedPositions)
     {
         this.operatorContext = operatorContext;
         this.partitionIndex = partitionIndex;
-        this.filterFunction = filterFunction;
+        this.filterFunctionFactory = filterFunctionFactory;
 
         this.index = new PagesIndex(lookupSourceSupplier.getTypes(), expectedPositions);
         this.lookupSourceSupplier = lookupSourceSupplier;
@@ -175,7 +176,7 @@ public void finish()
         }
         finishing = true;
 
-        LookupSource lookupSource = index.createLookupSource(hashChannels, preComputedHashChannel, filterFunction);
+        LookupSource lookupSource = index.createLookupSource(operatorContext.getSession(), hashChannels, preComputedHashChannel, filterFunctionFactory);
         lookupSourceSupplier.setLookupSource(partitionIndex, lookupSource);
 
         operatorContext.setMemoryReservation(lookupSource.getInMemorySizeInBytes());

presto-main/src/main/java/com/facebook/presto/operator/index/IndexSnapshotBuilder.java

@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.operator.index;
 
+import com.facebook.presto.Session;
 import com.facebook.presto.operator.DriverContext;
 import com.facebook.presto.operator.LookupSource;
 import com.facebook.presto.operator.PagesIndex;
@@ -34,6 +35,7 @@
 
 public class IndexSnapshotBuilder
 {
+    private final Session session;
     private final int expectedPositions;
     private final List<Type> outputTypes;
     private final List<Type> missingKeysTypes;
@@ -63,6 +65,7 @@ public IndexSnapshotBuilder(List<Type> outputTypes,
         requireNonNull(maxMemoryInBytes, "maxMemoryInBytes is null");
         checkArgument(expectedPositions > 0, "expectedPositions must be greater than zero");
 
+        this.session = driverContext.getSession();
         this.outputTypes = ImmutableList.copyOf(outputTypes);
         this.expectedPositions = expectedPositions;
         this.keyOutputChannels = ImmutableList.copyOf(keyOutputChannels);
@@ -81,7 +84,7 @@ public IndexSnapshotBuilder(List<Type> outputTypes,
 
         this.outputPagesIndex = new PagesIndex(outputTypes, expectedPositions);
         this.missingKeysIndex = new PagesIndex(missingKeysTypes.build(), expectedPositions);
-        this.missingKeys = missingKeysIndex.createLookupSource(this.missingKeysChannels);
+        this.missingKeys = missingKeysIndex.createLookupSource(session, this.missingKeysChannels);
     }
 
     public List<Type> getOutputTypes()
@@ -118,7 +121,7 @@ public IndexSnapshot createIndexSnapshot(UnloadedIndexKeyRecordSet indexKeysReco
         }
         pages.clear();
 
-        LookupSource lookupSource = outputPagesIndex.createLookupSource(keyOutputChannels, keyOutputHashChannel, Optional.empty());
+        LookupSource lookupSource = outputPagesIndex.createLookupSource(session, keyOutputChannels, keyOutputHashChannel, Optional.empty());
 
         // Build a page containing the keys that produced no output rows, so in future requests can skip these keys
         PageBuilder missingKeysPageBuilder = new PageBuilder(missingKeysIndex.getTypes());
@@ -146,7 +149,7 @@ public IndexSnapshot createIndexSnapshot(UnloadedIndexKeyRecordSet indexKeysReco
         // only update missing keys if we have new missing keys
         if (!missingKeysPageBuilder.isEmpty()) {
             missingKeysIndex.addPage(missingKeysPage);
-            missingKeys = missingKeysIndex.createLookupSource(missingKeysChannels);
+            missingKeys = missingKeysIndex.createLookupSource(session, missingKeysChannels);
         }
 
         return new IndexSnapshot(lookupSource, missingKeys);

presto-main/src/main/java/com/facebook/presto/sql/gen/JoinCompiler.java

@@ -27,11 +27,9 @@
 import com.facebook.presto.bytecode.expression.BytecodeExpression;
 import com.facebook.presto.bytecode.instruction.LabelNode;
 import com.facebook.presto.operator.InMemoryJoinHash;
-import com.facebook.presto.operator.InternalJoinFilterFunction;
 import com.facebook.presto.operator.JoinFilterFunction;
 import com.facebook.presto.operator.LookupSource;
 import com.facebook.presto.operator.PagesHashStrategy;
-import com.facebook.presto.operator.StandardJoinFilterFunction;
 import com.facebook.presto.spi.Page;
 import com.facebook.presto.spi.PageBuilder;
 import com.facebook.presto.spi.block.Block;
@@ -48,7 +46,6 @@
 import it.unimi.dsi.fastutil.longs.LongArrayList;
 
 import java.lang.reflect.Constructor;
-import java.lang.reflect.InvocationTargetException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
@@ -96,21 +93,6 @@ public Class<? extends PagesHashStrategy> load(CacheKey key)
                 }
             });
 
-    private final LoadingCache<Class<? extends InternalJoinFilterFunction>, Class<? extends JoinFilterFunction>> joinFilterFunctionClasses = CacheBuilder.newBuilder().maximumSize(1000).build(
-            new CacheLoader<Class<? extends InternalJoinFilterFunction>, Class<? extends JoinFilterFunction>>()
-            {
-                @Override
-                public Class<? extends JoinFilterFunction> load(Class<? extends InternalJoinFilterFunction> key)
-                        throws Exception
-                {
-                    return IsolatedClass.isolateClass(
-                            new DynamicClassLoader(getClass().getClassLoader()),
-                            JoinFilterFunction.class,
-                            StandardJoinFilterFunction.class
-                    );
-                }
-            });
-
     public LookupSourceFactory compileLookupSourceFactory(List<? extends Type> types, List<Integer> joinChannels)
     {
         try {
@@ -121,21 +103,6 @@ public LookupSourceFactory compileLookupSourceFactory(List<? extends Type> types
         }
     }
 
-    public JoinFilterFunctionFactory compileJoinFilterFunctionFactory(InternalJoinFilterFunction internalJoinFilterFunction)
-    {
-        return ((filterFunction, channels) -> {
-            try {
-                return joinFilterFunctionClasses
-                        .get(internalJoinFilterFunction.getClass())
-                        .getConstructor(InternalJoinFilterFunction.class, List.class)
-                        .newInstance(filterFunction, channels);
-            }
-            catch (ExecutionException | UncheckedExecutionException | ExecutionError | NoSuchMethodException | InstantiationException | IllegalAccessException | InvocationTargetException e) {
-                throw Throwables.propagate(e.getCause());
-            }
-        });
-    }
-
     public PagesHashStrategyFactory compilePagesHashStrategyFactory(List<Type> types, List<Integer> joinChannels)
     {
         requireNonNull(types, "types is null");
@@ -751,11 +718,6 @@ public PagesHashStrategy createPagesHashStrategy(List<? extends List<Block>> cha
         }
     }
 
-    public interface JoinFilterFunctionFactory
-    {
-        JoinFilterFunction create(InternalJoinFilterFunction filterFunction, List<List<Block>> channels);
-    }
-
     private static final class CacheKey
     {
         private final List<Type> types;