Use bloom filter for evaluating dynamic filters on strings

Improves efficiency of evaluating dynamic filters on strings with
the potential for some false positives compared to exsitng approach

BenchmarkDynamicPageFilter.filterPages
(filterSize)  (inputDataSet)  (inputNullChance)  (nonNullsSelectivity)  (nullsAllowed)   Mode  Cnt     Before Score      After Score  Units
           2  VARCHAR_RANDOM               0.01                    0.2           false  thrpt   10   80.908 ± 1.927  172.244 ± 1.067  ops/s
           5  VARCHAR_RANDOM               0.01                    0.2           false  thrpt   10   81.052 ± 2.569  175.619 ± 1.225  ops/s
          10  VARCHAR_RANDOM               0.01                    0.2           false  thrpt   10   76.787 ± 1.561  176.371 ± 0.559  ops/s
         100  VARCHAR_RANDOM               0.01                    0.2           false  thrpt   10   75.631 ± 1.372  174.288 ± 1.024  ops/s
        1000  VARCHAR_RANDOM               0.01                    0.2           false  thrpt   10   69.615 ± 0.721  173.340 ± 0.867  ops/s
       10000  VARCHAR_RANDOM               0.01                    0.2           false  thrpt   10   75.401 ± 1.233  173.285 ± 1.752  ops/s
      100000  VARCHAR_RANDOM               0.01                    0.2           false  thrpt   10   64.335 ± 2.936  170.087 ± 1.370  ops/s
     1000000  VARCHAR_RANDOM               0.01                    0.2           false  thrpt   10   16.808 ± 3.205  170.403 ± 1.471  ops/s
     5000000  VARCHAR_RANDOM               0.01                    0.2           false  thrpt   10   15.766 ± 0.820  150.588 ± 4.034  ops/s

core/trino-main/src/main/java/io/trino/sql/gen/columnar/BloomFilter.java

@@ -0,0 +1,344 @@
+/*
+ * 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 io.trino.sql.gen.columnar;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Throwables;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import io.airlift.bytecode.BytecodeBlock;
+import io.airlift.bytecode.ClassDefinition;
+import io.airlift.bytecode.FieldDefinition;
+import io.airlift.bytecode.MethodDefinition;
+import io.airlift.bytecode.Parameter;
+import io.airlift.bytecode.Scope;
+import io.airlift.bytecode.Variable;
+import io.airlift.bytecode.control.ForLoop;
+import io.airlift.bytecode.expression.BytecodeExpression;
+import io.airlift.slice.Slice;
+import io.airlift.slice.XxHash64;
+import io.trino.annotation.UsedByGeneratedCode;
+import io.trino.cache.NonEvictableCache;
+import io.trino.operator.project.InputChannels;
+import io.trino.spi.Page;
+import io.trino.spi.TrinoException;
+import io.trino.spi.block.Block;
+import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.predicate.Domain;
+import io.trino.spi.type.CharType;
+import io.trino.spi.type.Type;
+import io.trino.spi.type.VarbinaryType;
+import io.trino.spi.type.VarcharType;
+import org.objectweb.asm.MethodTooLargeException;
+
+import java.lang.invoke.MethodHandle;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import static com.google.common.base.Verify.verify;
+import static io.airlift.bytecode.Access.FINAL;
+import static io.airlift.bytecode.Access.PRIVATE;
+import static io.airlift.bytecode.Access.PUBLIC;
+import static io.airlift.bytecode.Access.a;
+import static io.airlift.bytecode.Parameter.arg;
+import static io.airlift.bytecode.ParameterizedType.type;
+import static io.airlift.bytecode.expression.BytecodeExpressions.add;
+import static io.airlift.bytecode.expression.BytecodeExpressions.constantInt;
+import static io.airlift.bytecode.expression.BytecodeExpressions.inlineIf;
+import static io.airlift.bytecode.expression.BytecodeExpressions.lessThan;
+import static io.trino.cache.CacheUtils.uncheckedCacheGet;
+import static io.trino.cache.SafeCaches.buildNonEvictableCache;
+import static io.trino.spi.StandardErrorCode.COMPILER_ERROR;
+import static io.trino.spi.StandardErrorCode.QUERY_EXCEEDED_COMPILER_LIMIT;
+import static io.trino.sql.gen.columnar.ColumnarFilterCompiler.updateOutputPositions;
+import static io.trino.util.CompilerUtils.defineClass;
+import static io.trino.util.CompilerUtils.makeClassName;
+import static io.trino.util.Reflection.constructorMethodHandle;
+import static java.util.Objects.requireNonNull;
+
+public class BloomFilter
+{
+    // Generate a ColumnarBloomFilter class per Type to avoid mega-morphic call site when reading a position from input block
+    private static final NonEvictableCache<Type, MethodHandle> COLUMNAR_BLOOM_FILTER_CACHE = buildNonEvictableCache(
+            CacheBuilder.newBuilder()
+                    .maximumSize(100)
+                    .expireAfterWrite(2, TimeUnit.HOURS));
+
+    private BloomFilter() {}
+
+    public static boolean canUseBloomFilter(Domain domain)
+    {
+        Type type = domain.getType();
+        if (type instanceof VarcharType || type instanceof CharType || type instanceof VarbinaryType) {
+            verify(type.getJavaType() == Slice.class, "Type is not backed by Slice");
+            return !domain.isNone()
+                    && !domain.isAll()
+                    && domain.isNullableDiscreteSet()
+                    && domain.getValues().getRanges().getRangeCount() > 1; // Bloom filter is not faster to evaluate for single value
+        }
+        return false;
+    }
+
+    public static Supplier<FilterEvaluator> createBloomFilterEvaluator(Domain domain, int inputChannel)
+    {
+        return () -> new ColumnarFilterEvaluator(
+                new DictionaryAwareColumnarFilter(
+                        createColumnarBloomFilter(domain.getType(), inputChannel, domain.getNullableDiscreteSet()).get()));
+    }
+
+    private static Supplier<ColumnarFilter> createColumnarBloomFilter(Type type, int inputChannel, Domain.DiscreteSet discreteSet)
+    {
+        MethodHandle filterConstructor = uncheckedCacheGet(
+                COLUMNAR_BLOOM_FILTER_CACHE,
+                type,
+                () -> generateColumnarBloomFilterClass(type));
+
+        return () -> {
+            try {
+                SliceBloomFilter filter = new SliceBloomFilter((List<Slice>) (List<?>) discreteSet.getNonNullValues(), discreteSet.containsNull(), type);
+                InputChannels inputChannels = new InputChannels(ImmutableList.of(inputChannel), ImmutableList.of(inputChannel));
+                return (ColumnarFilter) filterConstructor.invoke(filter, inputChannels);
+            }
+            catch (Throwable e) {
+                throw new RuntimeException(e);
+            }
+        };
+    }
+
+    private static MethodHandle generateColumnarBloomFilterClass(Type type)
+    {
+        ClassDefinition classDefinition = new ClassDefinition(
+                a(PUBLIC, FINAL),
+                makeClassName(ColumnarFilter.class.getSimpleName() + "_bloom_filter_" + type, Optional.empty()),
+                type(Object.class),
+                type(ColumnarFilter.class));
+
+        FieldDefinition filterField = classDefinition.declareField(a(PRIVATE, FINAL), "filter", SliceBloomFilter.class);
+        FieldDefinition inputChannelsField = classDefinition.declareField(a(PRIVATE, FINAL), "inputChannels", InputChannels.class);
+        Parameter filterParameter = arg("filter", SliceBloomFilter.class);
+        Parameter inputChannelsParameter = arg("inputChannels", InputChannels.class);
+        MethodDefinition constructorDefinition = classDefinition.declareConstructor(a(PUBLIC), filterParameter, inputChannelsParameter);
+        BytecodeBlock body = constructorDefinition.getBody();
+        Variable thisVariable = constructorDefinition.getThis();
+        body.comment("super();")
+                .append(thisVariable)
+                .invokeConstructor(Object.class)
+                .append(thisVariable.setField(filterField, filterParameter))
+                .append(thisVariable.setField(inputChannelsField, inputChannelsParameter))
+                .ret();
+
+        // getInputChannels
+        MethodDefinition method = classDefinition.declareMethod(a(PUBLIC), "getInputChannels", type(InputChannels.class));
+        Scope scope = method.getScope();
+        method.getBody().append(scope.getThis().getField(inputChannelsField).ret());
+
+        generateFilterRangeMethod(classDefinition);
+        generateFilterListMethod(classDefinition);
+
+        Class<? extends ColumnarFilter> filterClass;
+        try {
+            filterClass = defineClass(classDefinition, ColumnarFilter.class, ImmutableMap.of(), ColumnarFilterCompiler.class.getClassLoader());
+        }
+        catch (Exception e) {
+            if (Throwables.getRootCause(e) instanceof MethodTooLargeException) {
+                throw new TrinoException(QUERY_EXCEEDED_COMPILER_LIMIT,
+                        "Query exceeded maximum filters. Please reduce the number of filters referenced and re-run the query.", e);
+            }
+            throw new TrinoException(COMPILER_ERROR, e.getCause());
+        }
+        return constructorMethodHandle(filterClass, SliceBloomFilter.class, InputChannels.class);
+    }
+
+    private static void generateFilterRangeMethod(ClassDefinition classDefinition)
+    {
+        Parameter session = arg("session", ConnectorSession.class);
+        Parameter outputPositions = arg("outputPositions", int[].class);
+        Parameter offset = arg("offset", int.class);
+        Parameter size = arg("size", int.class);
+        Parameter page = arg("page", Page.class);
+
+        MethodDefinition method = classDefinition.declareMethod(
+                a(PUBLIC),
+                "filterPositionsRange",
+                type(int.class),
+                ImmutableList.of(session, outputPositions, offset, size, page));
+        Scope scope = method.getScope();
+        BytecodeBlock body = method.getBody();
+
+        Variable block = declareBlockVariable(page, scope, body);
+        Variable outputPositionsCount = scope.declareVariable("outputPositionsCount", body, constantInt(0));
+        Variable position = scope.declareVariable(int.class, "position");
+        Variable result = scope.declareVariable(boolean.class, "result");
+
+        /* for(int position = offset; position < offset + size; ++position) {
+         *   boolean result = block.isNull(position) ? this.filter.containsNull() : this.filter.test(block, position);
+         *   outputPositions[outputPositionsCount] = position;
+         *   outputPositionsCount += result ? 1 : 0;
+         * }
+         */
+        body.append(new ForLoop("nullable range based loop")
+                .initialize(position.set(offset))
+                .condition(lessThan(position, add(offset, size)))
+                .update(position.increment())
+                .body(new BytecodeBlock()
+                                .append(generateBloomFilterTest(scope, block, position, result))
+                                .append(updateOutputPositions(result, position, outputPositions, outputPositionsCount))));
+
+        body.append(outputPositionsCount.ret());
+    }
+
+    private static void generateFilterListMethod(ClassDefinition classDefinition)
+    {
+        Parameter session = arg("session", ConnectorSession.class);
+        Parameter outputPositions = arg("outputPositions", int[].class);
+        Parameter activePositions = arg("activePositions", int[].class);
+        Parameter offset = arg("offset", int.class);
+        Parameter size = arg("size", int.class);
+        Parameter page = arg("page", Page.class);
+
+        MethodDefinition method = classDefinition.declareMethod(
+                a(PUBLIC),
+                "filterPositionsList",
+                type(int.class),
+                ImmutableList.of(session, outputPositions, activePositions, offset, size, page));
+        Scope scope = method.getScope();
+        BytecodeBlock body = method.getBody();
+
+        Variable block = declareBlockVariable(page, scope, body);
+        Variable outputPositionsCount = scope.declareVariable("outputPositionsCount", body, constantInt(0));
+        Variable index = scope.declareVariable(int.class, "index");
+        Variable position = scope.declareVariable(int.class, "position");
+        Variable result = scope.declareVariable(boolean.class, "result");
+
+        /* for(int index = offset; index < offset + size; ++index) {
+         *   int position = activePositions[index];
+         *   boolean result = block.isNull(position) ? this.filter.containsNull() : this.filter.test(block, position);
+         *   outputPositions[outputPositionsCount] = position;
+         *   outputPositionsCount += result ? 1 : 0;
+         * }
+         */
+        body.append(new ForLoop("nullable range based loop")
+                .initialize(index.set(offset))
+                .condition(lessThan(index, add(offset, size)))
+                .update(index.increment())
+                .body(new BytecodeBlock()
+                        .append(position.set(activePositions.getElement(index)))
+                        .append(generateBloomFilterTest(scope, block, position, result))
+                        .append(updateOutputPositions(result, position, outputPositions, outputPositionsCount))));
+
+        body.append(outputPositionsCount.ret());
+    }
+
+    private static Variable declareBlockVariable(Parameter page, Scope scope, BytecodeBlock body)
+    {
+        return scope.declareVariable(
+                "block",
+                body,
+                page.invoke("getBlock", Block.class, constantInt(0)));
+    }
+
+    private static BytecodeBlock generateBloomFilterTest(Scope scope, Variable block, Variable position, Variable result)
+    {
+        BytecodeExpression filter = scope.getThis().getField("filter", SliceBloomFilter.class);
+        // boolean result = block.isNull(position) ? this.filter.containsNull() : this.filter.test(block, position)
+        return new BytecodeBlock()
+                .append(result.set(inlineIf(
+                        block.invoke("isNull", boolean.class, position),
+                        filter.invoke("containsNull", boolean.class),
+                        filter.invoke("test", boolean.class, block, position))));
+    }
+
+    public static final class SliceBloomFilter
+    {
+        private final long[] bloom;
+        private final int bloomSizeMask;
+        private final Type type;
+        private final boolean containsNull;
+
+        /**
+         * A Bloom filter for a set of Slice values.
+         * This is approx 2X faster than the Bloom filter implementations in ORC and parquet because
+         * it uses single hash function and uses that to set 3 bits within a 64 bit word.
+         * The memory footprint is up to (4 * values.size()) bytes, which is much smaller than maintaining a hash set of strings.
+         *
+         * @param values       values used for filtering
+         * @param containsNull whether null values are contained by the filter
+         * @param type         type of the values
+         */
+        public SliceBloomFilter(List<Slice> values, boolean containsNull, Type type)
+        {
+            this.containsNull = containsNull;
+            this.type = requireNonNull(type, "type is null");
+            int bloomSize = getBloomFilterSize(values.size());
+            bloom = new long[bloomSize];
+            bloomSizeMask = bloomSize - 1;
+            for (Slice value : values) {
+                long hashCode = XxHash64.hash(value);
+                // Set 3 bits in a 64 bit word
+                bloom[bloomIndex(hashCode)] |= bloomMask(hashCode);
+            }
+        }
+
+        @UsedByGeneratedCode
+        public boolean containsNull()
+        {
+            return containsNull;
+        }
+
+        @UsedByGeneratedCode
+        public boolean test(Block block, int position)
+        {
+            return contains(type.getSlice(block, position));
+        }
+
+        @VisibleForTesting
+        public boolean contains(Slice data)
+        {
+            long hashCode = XxHash64.hash(data);
+            long mask = bloomMask(hashCode);
+            return mask == (bloom[bloomIndex(hashCode)] & mask);
+        }
+
+        private int bloomIndex(long hashCode)
+        {
+            // Lower 21 bits are not used by bloomMask
+            // These are enough for the maximum size array that will be used here
+            return (int) (hashCode & bloomSizeMask);
+        }
+
+        private static long bloomMask(long hashCode)
+        {
+            // returned mask sets 3 bits based on portions of given hash
+            // Extract 38th to 43rd bits
+            return (1L << ((hashCode >> 21) & 63))
+                    // Extract 32nd to 37th bits
+                    | (1L << ((hashCode >> 27) & 63))
+                    // Extract 26th to 31st bits
+                    | (1L << ((hashCode >> 33) & 63));
+        }
+
+        private static int getBloomFilterSize(int valuesCount)
+        {
+            // Linear hash table size is the highest power of two less than or equal to number of values * 4. This means that the
+            // table is under half full, e.g. 127 elements gets 256 slots.
+            int hashTableSize = Integer.highestOneBit(valuesCount * 4);
+            // We will allocate 8 bits in the bloom filter for every slot in a comparable hash table.
+            // The bloomSize is a count of longs, hence / 8.
+            return Math.max(1, hashTableSize / 8);
+        }
+    }
+}