Sort position links for faster non-equi joins

presto-benchmark/src/test/java/com/facebook/presto/benchmark/BenchmarkDecimalAggregation.java

@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.benchmark;
 
+import com.facebook.presto.spi.Page;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
 import org.openjdk.jmh.annotations.Fork;
@@ -28,6 +29,8 @@
 import org.openjdk.jmh.runner.options.OptionsBuilder;
 import org.openjdk.jmh.runner.options.VerboseMode;
 
+import java.util.List;
+
 import static java.lang.String.format;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static org.openjdk.jmh.annotations.Mode.AverageTime;
@@ -55,24 +58,25 @@ public static class AggregationContext
 
         private final MemoryLocalQueryRunner queryRunner = new MemoryLocalQueryRunner();
 
+        public final MemoryLocalQueryRunner getQueryRunner()
+        {
+            return queryRunner;
+        }
+
         @Setup
         public void setUp()
         {
             queryRunner.execute(format(
                     "CREATE TABLE memory.default.orders AS SELECT orderstatus, cast(totalprice as %s) totalprice FROM tpch.sf1.orders",
                     type));
         }
-
-        public void run()
-        {
-            queryRunner.execute(format("SELECT %s FROM orders GROUP BY orderstatus", project));
-        }
     }
 
     @Benchmark
-    public void benchmarkBuildHash(AggregationContext context)
+    public List<Page> benchmarkBuildHash(AggregationContext context)
     {
-        context.run();
+        return context.getQueryRunner()
+                .execute(String.format("SELECT %s FROM orders GROUP BY orderstatus", context.project));
     }
 
     public static void main(String[] args)

presto-benchmark/src/test/java/com/facebook/presto/benchmark/BenchmarkInequalityJoin.java

@@ -0,0 +1,120 @@
+/*
+ * 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.benchmark;
+
+import com.facebook.presto.SystemSessionProperties;
+import com.facebook.presto.spi.Page;
+import com.google.common.collect.ImmutableMap;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+import org.openjdk.jmh.runner.options.VerboseMode;
+
+import java.util.List;
+
+import static java.lang.String.format;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.openjdk.jmh.annotations.Mode.AverageTime;
+import static org.openjdk.jmh.annotations.Scope.Thread;
+
+/**
+ * This benchmark a case when there is almost like a cross join query
+ * but with very selective inequality join condition. In other words
+ * for each probe position there are lots of matching build positions
+ * which are filtered out by filtering function.
+ */
+@SuppressWarnings("MethodMayBeStatic")
+@State(Thread)
+@OutputTimeUnit(MILLISECONDS)
+@BenchmarkMode(AverageTime)
+@Fork(3)
+@Warmup(iterations = 10)
+@Measurement(iterations = 10)
+public class BenchmarkInequalityJoin
+{
+    @State(Thread)
+    public static class Context
+    {
+        private MemoryLocalQueryRunner queryRunner;
+
+        @Param({"true", "false"})
+        private String fastInequalityJoin;
+
+        // number of buckets. The smaller number of buckets, the longer position links chain
+        @Param({"100", "1000", "10000", "60000"})
+        private int buckets;
+
+        // How many positions out of 1000 will be actually joined
+        // 10 means 1 - 10/1000 = 99/100 positions will be filtered out
+        @Param({"10"})
+        private int filterOutCoefficient;
+
+        public MemoryLocalQueryRunner getQueryRunner()
+        {
+            return queryRunner;
+        }
+
+        @Setup
+        public void setUp()
+        {
+            queryRunner = new MemoryLocalQueryRunner(ImmutableMap.of(SystemSessionProperties.FAST_INEQUALITY_JOIN, fastInequalityJoin));
+
+            // t1.val1 is in range [0, 1000)
+            // t1.bucket is in [0, 1000)
+            queryRunner.execute(format(
+                    "CREATE TABLE memory.default.t1 AS SELECT " +
+                        "orderkey %% %d bucket, " +
+                        "(orderkey * 13) %% 1000 val1 " +
+                    "FROM tpch.tiny.lineitem",
+                    buckets));
+            // t2.val2 is in range [0, 10)
+            // t2.bucket is in [0, 1000)
+            queryRunner.execute(format(
+                    "CREATE TABLE memory.default.t2 AS SELECT " +
+                            "orderkey %% %d bucket, " +
+                            "(orderkey * 379) %% %d val2 " +
+                            "FROM tpch.tiny.lineitem",
+                    buckets,
+                    filterOutCoefficient));
+        }
+    }
+
+    @Benchmark
+    public List<Page> benchmarkJoin(Context context)
+    {
+        return context.getQueryRunner()
+                .execute("SELECT count(*) FROM t1 JOIN t2 on (t1.bucket = t2.bucket) WHERE t1.val1 < t2.val2");
+    }
+
+    public static void main(String[] args)
+            throws RunnerException
+    {
+        Options options = new OptionsBuilder()
+                .verbosity(VerboseMode.NORMAL)
+                .include(".*" + BenchmarkInequalityJoin.class.getSimpleName() + ".*")
+                .build();
+
+        new Runner(options).run();
+    }
+}

presto-benchmark/src/test/java/com/facebook/presto/benchmark/MemoryLocalQueryRunner.java

@@ -21,45 +21,65 @@
 import com.facebook.presto.operator.Driver;
 import com.facebook.presto.operator.TaskContext;
 import com.facebook.presto.plugin.memory.MemoryConnectorFactory;
+import com.facebook.presto.spi.Page;
 import com.facebook.presto.spi.QueryId;
 import com.facebook.presto.spi.memory.MemoryPoolId;
 import com.facebook.presto.spiller.SpillSpaceTracker;
 import com.facebook.presto.testing.LocalQueryRunner;
-import com.facebook.presto.testing.NullOutputOperator;
+import com.facebook.presto.testing.PageConsumerOperator;
 import com.facebook.presto.tpch.TpchConnectorFactory;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import io.airlift.units.DataSize;
 import org.intellij.lang.annotations.Language;
 
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ExecutorService;
 
 import static com.facebook.presto.testing.TestingSession.testSessionBuilder;
 import static io.airlift.units.DataSize.Unit.GIGABYTE;
-import static io.airlift.units.DataSize.Unit.MEGABYTE;
 
 public class MemoryLocalQueryRunner
 {
-    protected LocalQueryRunner localQueryRunner = createMemoryLocalQueryRunner();
+    protected final LocalQueryRunner localQueryRunner;
+    protected final Session session;
 
-    public void execute(@Language("SQL") String query)
+    public MemoryLocalQueryRunner()
+    {
+        this(ImmutableMap.of());
+    }
+
+    public MemoryLocalQueryRunner(Map<String, String> properties)
+    {
+        Session.SessionBuilder sessionBuilder = testSessionBuilder()
+                .setCatalog("memory")
+                .setSchema("default");
+        properties.forEach(sessionBuilder::setSystemProperty);
+
+        session = sessionBuilder.build();
+        localQueryRunner = createMemoryLocalQueryRunner(session);
+    }
+
+    public List<Page> execute(@Language("SQL") String query)
     {
-        Session session = testSessionBuilder()
-                .setSystemProperty("optimizer.optimize-hash-generation", "true")
-                .build();
         ExecutorService executor = localQueryRunner.getExecutor();
-        MemoryPool memoryPool = new MemoryPool(new MemoryPoolId("test"), new DataSize(1, GIGABYTE));
-        MemoryPool systemMemoryPool = new MemoryPool(new MemoryPoolId("testSystem"), new DataSize(1, GIGABYTE));
-        SpillSpaceTracker spillSpaceTracker = new SpillSpaceTracker(new DataSize(1, GIGABYTE));
+        MemoryPool memoryPool = new MemoryPool(new MemoryPoolId("test"), new DataSize(2, GIGABYTE));
+        MemoryPool systemMemoryPool = new MemoryPool(new MemoryPoolId("testSystem"), new DataSize(2, GIGABYTE));
 
-        TaskContext taskContext = new QueryContext(new QueryId("test"), new DataSize(256, MEGABYTE), memoryPool, systemMemoryPool, executor, new DataSize(1, GIGABYTE), spillSpaceTracker)
+        SpillSpaceTracker spillSpaceTracker = new SpillSpaceTracker(new DataSize(1, GIGABYTE));
+        TaskContext taskContext = new QueryContext(new QueryId("test"), new DataSize(1, GIGABYTE), memoryPool, systemMemoryPool, executor, new DataSize(4, GIGABYTE), spillSpaceTracker)
                 .addTaskContext(new TaskStateMachine(new TaskId("query", 0, 0), executor),
                         session,
                         false,
                         false);
 
         // Use NullOutputFactory to avoid coping out results to avoid affecting benchmark results
-        List<Driver> drivers = localQueryRunner.createDrivers(query, new NullOutputOperator.NullOutputFactory(), taskContext);
+        ImmutableList.Builder<Page> output = ImmutableList.builder();
+        List<Driver> drivers = localQueryRunner.createDrivers(
+                query,
+                new PageConsumerOperator.PageConsumerOutputFactory(types -> output::add),
+                taskContext);
 
         boolean done = false;
         while (!done) {
@@ -72,20 +92,20 @@ public void execute(@Language("SQL") String query)
             }
             done = !processed;
         }
+
+        return output.build();
     }
 
-    private static LocalQueryRunner createMemoryLocalQueryRunner()
+    private static LocalQueryRunner createMemoryLocalQueryRunner(Session session)
     {
-        Session.SessionBuilder sessionBuilder = testSessionBuilder()
-                .setCatalog("memory")
-                .setSchema("default");
-
-        Session session = sessionBuilder.build();
         LocalQueryRunner localQueryRunner = LocalQueryRunner.queryRunnerWithInitialTransaction(session);
 
         // add tpch
-        localQueryRunner.createCatalog("tpch", new TpchConnectorFactory(1), ImmutableMap.<String, String>of());
-        localQueryRunner.createCatalog("memory", new MemoryConnectorFactory(), ImmutableMap.<String, String>of());
+        localQueryRunner.createCatalog("tpch", new TpchConnectorFactory(1), ImmutableMap.of());
+        localQueryRunner.createCatalog(
+                "memory",
+                new MemoryConnectorFactory(),
+                ImmutableMap.of("memory.max-data-per-node", "4GB"));
 
         return localQueryRunner;
     }

presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraDistributed.java

@@ -39,6 +39,12 @@ protected boolean supportsViews()
         return false;
     }
 
+    @Override
+    public void testJoinWithLessThanOnDatesInJoinClause()
+    {
+        // Cassandra does not support DATE
+    }
+
     @Override
     public void testGroupingSetMixedExpressionAndColumn()
     {

presto-main/src/main/java/com/facebook/presto/SystemSessionProperties.java

@@ -62,6 +62,7 @@ public final class SystemSessionProperties
     public static final String INITIAL_SPLITS_PER_NODE = "initial_splits_per_node";
     public static final String SPLIT_CONCURRENCY_ADJUSTMENT_INTERVAL = "split_concurrency_adjustment_interval";
     public static final String OPTIMIZE_METADATA_QUERIES = "optimize_metadata_queries";
+    public static final String FAST_INEQUALITY_JOIN = "fast_inequality_join";
     public static final String QUERY_PRIORITY = "query_priority";
     public static final String SPILL_ENABLED = "spill_enabled";
     public static final String OPERATOR_MEMORY_LIMIT_BEFORE_SPILL = "operator_memory_limit_before_spill";
@@ -241,6 +242,11 @@ public SystemSessionProperties(
                         "Experimental: Reorder joins to optimize plan",
                         featuresConfig.isJoinReorderingEnabled(),
                         false),
+                booleanSessionProperty(
+                        FAST_INEQUALITY_JOIN,
+                        "Experimental: Use faster handling of inequality join if it is possible",
+                        featuresConfig.isFastInequalityJoins(),
+                        false),
                 booleanSessionProperty(
                         COLOCATED_JOIN,
                         "Experimental: Use a colocated join when possible",
@@ -398,6 +404,11 @@ public static boolean planWithTableNodePartitioning(Session session)
         return session.getSystemProperty(PLAN_WITH_TABLE_NODE_PARTITIONING, Boolean.class);
     }
 
+    public static boolean isFastInequalityJoin(Session session)
+    {
+        return session.getSystemProperty(FAST_INEQUALITY_JOIN, Boolean.class);
+    }
+
     public static boolean isJoinReorderingEnabled(Session session)
     {
         return session.getSystemProperty(REORDER_JOINS, Boolean.class);