[Backport] use a sha512 hash of bloom filter for cache key instead of filter bytes

extensions-core/druid-bloom-filter/src/main/java/org/apache/druid/guice/BloomFilterSerializersModule.java

@@ -20,14 +20,14 @@
 
 import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.core.JsonParser;
-import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.DeserializationContext;
 import com.fasterxml.jackson.databind.SerializerProvider;
 import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
 import com.fasterxml.jackson.databind.jsontype.NamedType;
 import com.fasterxml.jackson.databind.module.SimpleModule;
 import com.fasterxml.jackson.databind.ser.std.StdSerializer;
 import org.apache.druid.query.filter.BloomDimFilter;
+import org.apache.druid.query.filter.BloomKFilterHolder;
 import org.apache.hive.common.util.BloomKFilter;
 
 import java.io.ByteArrayInputStream;
@@ -45,12 +45,13 @@ public BloomFilterSerializersModule()
     );
     addSerializer(BloomKFilter.class, new BloomKFilterSerializer());
     addDeserializer(BloomKFilter.class, new BloomKFilterDeserializer());
+    addDeserializer(BloomKFilterHolder.class, new BloomKFilterHolderDeserializer());
   }
 
-  public static class BloomKFilterSerializer extends StdSerializer<BloomKFilter>
+  private static class BloomKFilterSerializer extends StdSerializer<BloomKFilter>
   {
 
-    public BloomKFilterSerializer()
+    BloomKFilterSerializer()
     {
       super(BloomKFilter.class);
     }
@@ -60,29 +61,52 @@ public void serialize(
         BloomKFilter bloomKFilter, JsonGenerator jsonGenerator, SerializerProvider serializerProvider
     ) throws IOException
     {
-      ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
-      BloomKFilter.serialize(byteArrayOutputStream, bloomKFilter);
-      byte[] bytes = byteArrayOutputStream.toByteArray();
-      jsonGenerator.writeBinary(bytes);
+      jsonGenerator.writeBinary(bloomKFilterToBytes(bloomKFilter));
     }
   }
 
-  public static class BloomKFilterDeserializer extends StdDeserializer<BloomKFilter>
+  private static class BloomKFilterDeserializer extends StdDeserializer<BloomKFilter>
   {
 
-    protected BloomKFilterDeserializer()
+    BloomKFilterDeserializer()
     {
       super(BloomKFilter.class);
     }
 
     @Override
     public BloomKFilter deserialize(
         JsonParser jsonParser, DeserializationContext deserializationContext
-    ) throws IOException, JsonProcessingException
+    ) throws IOException
     {
-      byte[] bytes = jsonParser.getBinaryValue();
-      return BloomKFilter.deserialize(new ByteArrayInputStream(bytes));
+      return bloomKFilterFromBytes(jsonParser.getBinaryValue());
+    }
+  }
 
+  private static class BloomKFilterHolderDeserializer extends StdDeserializer<BloomKFilterHolder>
+  {
+    BloomKFilterHolderDeserializer()
+    {
+      super(BloomKFilterHolder.class);
     }
+
+    @Override
+    public BloomKFilterHolder deserialize(
+        JsonParser jsonParser, DeserializationContext deserializationContext
+    ) throws IOException
+    {
+      return BloomKFilterHolder.fromBytes(jsonParser.getBinaryValue());
+    }
+  }
+
+  public static byte[] bloomKFilterToBytes(BloomKFilter bloomKFilter) throws IOException
+  {
+    ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+    BloomKFilter.serialize(byteArrayOutputStream, bloomKFilter);
+    return byteArrayOutputStream.toByteArray();
+  }
+
+  public static BloomKFilter bloomKFilterFromBytes(byte[] bytes) throws IOException
+  {
+    return BloomKFilter.deserialize(new ByteArrayInputStream(bytes));
   }
 }

extensions-core/druid-bloom-filter/src/main/java/org/apache/druid/query/filter/BloomDimFilter.java

@@ -25,15 +25,13 @@
 import com.google.common.base.Predicate;
 import com.google.common.collect.RangeSet;
 import com.google.common.collect.Sets;
-import org.apache.druid.java.util.common.ISE;
+import com.google.common.hash.HashCode;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.query.cache.CacheKeyBuilder;
 import org.apache.druid.query.extraction.ExtractionFn;
 import org.apache.druid.segment.filter.DimensionPredicateFilter;
 import org.apache.hive.common.util.BloomKFilter;
 
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
 import java.util.HashSet;
 
 /**
@@ -43,39 +41,33 @@ public class BloomDimFilter implements DimFilter
 
   private final String dimension;
   private final BloomKFilter bloomKFilter;
+  private final HashCode hash;
   private final ExtractionFn extractionFn;
 
   @JsonCreator
   public BloomDimFilter(
       @JsonProperty("dimension") String dimension,
-      @JsonProperty("bloomKFilter") BloomKFilter bloomKFilter,
+      @JsonProperty("bloomKFilter") BloomKFilterHolder bloomKFilterHolder,
       @JsonProperty("extractionFn") ExtractionFn extractionFn
   )
   {
     Preconditions.checkArgument(dimension != null, "dimension must not be null");
-    Preconditions.checkNotNull(bloomKFilter);
+    Preconditions.checkNotNull(bloomKFilterHolder);
     this.dimension = dimension;
-    this.bloomKFilter = bloomKFilter;
+    this.bloomKFilter = bloomKFilterHolder.getFilter();
+    this.hash = bloomKFilterHolder.getFilterHash();
     this.extractionFn = extractionFn;
   }
 
   @Override
   public byte[] getCacheKey()
   {
-    ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
-    try {
-      BloomKFilter.serialize(byteArrayOutputStream, bloomKFilter);
-    }
-    catch (IOException e) {
-      throw new ISE(e, "Exception when generating cache key for [%s]", this);
-    }
-    byte[] bloomFilterBytes = byteArrayOutputStream.toByteArray();
     return new CacheKeyBuilder(DimFilterUtils.BLOOM_DIM_FILTER_CACHE_ID)
         .appendString(dimension)
         .appendByte(DimFilterUtils.STRING_SEPARATOR)
         .appendByteArray(extractionFn == null ? new byte[0] : extractionFn.getCacheKey())
         .appendByte(DimFilterUtils.STRING_SEPARATOR)
-        .appendByteArray(bloomFilterBytes)
+        .appendByteArray(hash.asBytes())
         .build();
   }
 
@@ -187,9 +179,9 @@ public ExtractionFn getExtractionFn()
   public String toString()
   {
     if (extractionFn != null) {
-      return StringUtils.format("%s(%s) = %s", extractionFn, dimension, bloomKFilter);
+      return StringUtils.format("%s(%s) = %s", extractionFn, dimension, hash.toString());
     } else {
-      return StringUtils.format("%s = %s", dimension, bloomKFilter);
+      return StringUtils.format("%s = %s", dimension, hash.toString());
     }
   }
 
@@ -208,7 +200,7 @@ public boolean equals(Object o)
     if (!dimension.equals(that.dimension)) {
       return false;
     }
-    if (bloomKFilter != null ? !bloomKFilter.equals(that.bloomKFilter) : that.bloomKFilter != null) {
+    if (hash != null ? !hash.equals(that.hash) : that.hash != null) {
       return false;
     }
     return extractionFn != null ? extractionFn.equals(that.extractionFn) : that.extractionFn == null;
@@ -230,7 +222,7 @@ public HashSet<String> getRequiredColumns()
   public int hashCode()
   {
     int result = dimension.hashCode();
-    result = 31 * result + (bloomKFilter != null ? bloomKFilter.hashCode() : 0);
+    result = 31 * result + (hash != null ? hash.hashCode() : 0);
     result = 31 * result + (extractionFn != null ? extractionFn.hashCode() : 0);
     return result;
   }

extensions-core/druid-bloom-filter/src/main/java/org/apache/druid/query/filter/BloomKFilterHolder.java

@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.druid.query.filter;
+
+import com.google.common.hash.HashCode;
+import com.google.common.hash.Hashing;
+import org.apache.druid.guice.BloomFilterSerializersModule;
+import org.apache.hive.common.util.BloomKFilter;
+
+import java.io.IOException;
+import java.util.Objects;
+
+public class BloomKFilterHolder
+{
+  private final BloomKFilter filter;
+  private final HashCode hash;
+
+  public BloomKFilterHolder(BloomKFilter filter, HashCode hash)
+  {
+    this.filter = filter;
+    this.hash = hash;
+  }
+
+  BloomKFilter getFilter()
+  {
+    return filter;
+  }
+
+  HashCode getFilterHash()
+  {
+    return hash;
+  }
+
+  public static BloomKFilterHolder fromBloomKFilter(BloomKFilter filter) throws IOException
+  {
+    byte[] bytes = BloomFilterSerializersModule.bloomKFilterToBytes(filter);
+
+    return new BloomKFilterHolder(filter, Hashing.sha512().hashBytes(bytes));
+  }
+
+  public static BloomKFilterHolder fromBytes(byte[] bytes) throws IOException
+  {
+    return new BloomKFilterHolder(
+        BloomFilterSerializersModule.bloomKFilterFromBytes(bytes),
+        Hashing.sha512().hashBytes(bytes)
+    );
+  }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    BloomKFilterHolder that = (BloomKFilterHolder) o;
+    return Objects.equals(this.hash, that.hash);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(filter, hash);
+  }
+}