Support dynamic filter for hash join

presto-expressions/src/main/java/com/facebook/presto/expressions/DynamicFilters.java

@@ -118,8 +118,13 @@ public static Optional<DynamicFilterPlaceholder> getPlaceholder(RowExpression ex
         checkArgument(firstArgument instanceof ConstantExpression);
         checkArgument(firstArgument.getType() instanceof VarcharType);
 
-        String id = ((Slice) ((ConstantExpression) firstArgument).getValue()).toStringUtf8();
-        return Optional.of(new DynamicFilterPlaceholder(id, arguments.get(1)));
+        String[] value = ((Slice) ((ConstantExpression) firstArgument).getValue()).toStringUtf8().split("\\$");
+        if (value.length == 1) {
+            return Optional.of(new DynamicFilterPlaceholder(value[0], arguments.get(1)));
+        }
+        else {
+            return Optional.of(new DynamicFilterPlaceholder(value[0], value[1], arguments.get(1)));
+        }
     }
 
     public static RowExpression removeNestedDynamicFilters(RowExpression expression)
@@ -222,11 +227,18 @@ public List<DynamicFilterPlaceholder> getDynamicConjuncts()
     public static final class DynamicFilterPlaceholder
     {
         private final String id;
+        private final String source;
         private final RowExpression input;
 
         public DynamicFilterPlaceholder(String id, RowExpression input)
+        {
+            this(id, "", input);
+        }
+
+        public DynamicFilterPlaceholder(String id, String source, RowExpression input)
         {
             this.id = requireNonNull(id, "id is null");
+            this.source = requireNonNull(source, "source is null");
             this.input = requireNonNull(input, "input is null");
         }
 
@@ -235,6 +247,11 @@ public String getId()
             return id;
         }
 
+        public String getSource()
+        {
+            return source;
+        }
+
         public RowExpression getInput()
         {
             return input;

presto-main/pom.xml

@@ -358,6 +358,11 @@
             </exclusions>
         </dependency>
 
+        <dependency>
+            <groupId>com.github.luben</groupId>
+            <artifactId>zstd-jni</artifactId>
+        </dependency>
+
         <!-- for testing -->
         <dependency>
             <groupId>org.testng</groupId>

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

@@ -172,6 +172,9 @@ public final class SystemSessionProperties
     public static final String CHECK_ACCESS_CONTROL_ON_UTILIZED_COLUMNS_ONLY = "check_access_control_on_utilized_columns_only";
     public static final String SKIP_REDUNDANT_SORT = "skip_redundant_sort";
     public static final String ALLOW_WINDOW_ORDER_BY_LITERALS = "allow_window_order_by_literals";
+    public static final String ENABLE_HASH_JOIN_DYNAMIC_FILTERING = "enable_hash_join_dynamic_filtering";
+    public static final String BLOOM_FILTER_FOR_DYNAMIC_FILTERING_SIZE = "bloom_filter_for_dynamic_filtering_size";
+    public static final String BLOOM_FILTER_FOR_DYNAMIC_FILTERING_FALSE_POSITIVE_PROBABILITY = "bloom_filter_for_dynamic_filtering_false_positive_probability";
 
     private final List<PropertyMetadata<?>> sessionProperties;
 
@@ -896,6 +899,25 @@ public SystemSessionProperties(
                         ALLOW_WINDOW_ORDER_BY_LITERALS,
                         "Allow ORDER BY literals in window functions",
                         featuresConfig.isAllowWindowOrderByLiterals(),
+                        false),
+                booleanProperty(
+                        ENABLE_HASH_JOIN_DYNAMIC_FILTERING,
+                        "Enable hash join dynamic filtering",
+                        featuresConfig.isEnableHashJoinDynamicFiltering(),
+                        false),
+                new PropertyMetadata<>(
+                        BLOOM_FILTER_FOR_DYNAMIC_FILTERING_SIZE,
+                        "Hash join dynamic filtering bloom filter size",
+                        VARCHAR,
+                        DataSize.class,
+                        featuresConfig.getBloomFilterForDynamicFilterSize(),
+                        false,
+                        value -> DataSize.valueOf((String) value),
+                        DataSize::toString),
+                doubleProperty(
+                        BLOOM_FILTER_FOR_DYNAMIC_FILTERING_FALSE_POSITIVE_PROBABILITY,
+                        "Hash join dynamic filtering bloom filter false positive probability",
+                        featuresConfig.getBloomFilterForDynamicFilteringFalsePositiveProbability(),
                         false));
     }
 
@@ -1517,4 +1539,19 @@ public static boolean isCheckAccessControlOnUtilizedColumnsOnly(Session session)
     {
         return session.getSystemProperty(CHECK_ACCESS_CONTROL_ON_UTILIZED_COLUMNS_ONLY, Boolean.class);
     }
+
+    public static boolean isEnableHashJoinDynamicFiltering(Session session)
+    {
+        return session.getSystemProperty(ENABLE_HASH_JOIN_DYNAMIC_FILTERING, Boolean.class);
+    }
+
+    public static DataSize getBloomFilterForDynamicFilteringSize(Session session)
+    {
+        return session.getSystemProperty(BLOOM_FILTER_FOR_DYNAMIC_FILTERING_SIZE, DataSize.class);
+    }
+
+    public static Double getBloomFilterForDynamicFilteringFalsePositiveProbability(Session session)
+    {
+        return session.getSystemProperty(BLOOM_FILTER_FOR_DYNAMIC_FILTERING_FALSE_POSITIVE_PROBABILITY, Double.class);
+    }
 }

presto-main/src/main/java/com/facebook/presto/bloomfilter/BitArray.java

@@ -0,0 +1,135 @@
+/*
+ * 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.bloomfilter;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.Arrays;
+
+public final class BitArray
+{
+    private final long[] data;
+    private long bitCount;
+
+    static int numWords(long numBits)
+    {
+        if (numBits <= 0) {
+            throw new IllegalArgumentException("numBits must be positive, but got " + numBits);
+        }
+        long numWords = (long) Math.ceil(numBits / 64.0);
+        if (numWords > Integer.MAX_VALUE) {
+            throw new IllegalArgumentException("Can't allocate enough space for " + numBits + " bits");
+        }
+        return (int) numWords;
+    }
+
+    BitArray(long numBits)
+    {
+        this(new long[numWords(numBits)]);
+    }
+
+    private BitArray(long[] data)
+    {
+        this.data = data;
+        long bitCount = 0;
+        for (long word : data) {
+            bitCount += Long.bitCount(word);
+        }
+        this.bitCount = bitCount;
+    }
+
+    /** Returns true if the bit changed value. */
+    boolean set(long index)
+    {
+        if (!get(index)) {
+            data[(int) (index >>> 6)] |= (1L << index);
+            bitCount++;
+            return true;
+        }
+        return false;
+    }
+
+    boolean get(long index)
+    {
+        return (data[(int) (index >>> 6)] & (1L << index)) != 0;
+    }
+
+    /** Number of bits */
+    long bitSize()
+    {
+        return (long) data.length * Long.SIZE;
+    }
+
+    /** Number of set bits (1s) */
+    long cardinality()
+    {
+        return bitCount;
+    }
+
+    /** Combines the two BitArrays using bitwise OR. */
+    void putAll(BitArray array)
+    {
+        assert data.length == array.data.length : "BitArrays must be of equal length when merging";
+        long bitCount = 0;
+        for (int i = 0; i < data.length; i++) {
+            data[i] |= array.data[i];
+            bitCount += Long.bitCount(data[i]);
+        }
+        this.bitCount = bitCount;
+    }
+
+    void writeTo(DataOutputStream out) throws IOException
+    {
+        out.writeInt(data.length);
+        for (long datum : data) {
+            out.writeLong(datum);
+        }
+    }
+
+    static BitArray readFrom(DataInputStream in) throws IOException
+    {
+        int numWords = in.readInt();
+        long[] data = new long[numWords];
+        for (int i = 0; i < numWords; i++) {
+            data[i] = in.readLong();
+        }
+        return new BitArray(data);
+    }
+
+    public void clear()
+    {
+        Arrays.fill(data, 0);
+        bitCount = 0;
+    }
+
+    @Override
+    public boolean equals(Object other)
+    {
+        if (this == other) {
+            return true;
+        }
+        if (other == null || !(other instanceof BitArray)) {
+            return false;
+        }
+        BitArray that = (BitArray) other;
+        return Arrays.equals(data, that.data);
+    }
+
+    @Override
+    public int hashCode()
+    {
+        return Arrays.hashCode(data);
+    }
+}