path: root/common/sketch/src/main/java/org/apache/spark/util/sketch/BloomFilter.java

/*
 * 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.spark.util.sketch;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

/**
 * A Bloom filter is a space-efficient probabilistic data structure that offers an approximate
 * containment test with one-sided error: if it claims that an item is contained in it, this
 * might be in error, but if it claims that an item is <i>not</i> contained in it, then this is
 * definitely true. Currently supported data types include:
 * <ul>
 *   <li>{@link Byte}</li>
 *   <li>{@link Short}</li>
 *   <li>{@link Integer}</li>
 *   <li>{@link Long}</li>
 *   <li>{@link String}</li>
 * </ul>
 * The false positive probability ({@code FPP}) of a Bloom filter is defined as the probability that
 * {@linkplain #mightContain(Object)} will erroneously return {@code true} for an object that hasu
 * not actually been put in the {@code BloomFilter}.
 *
 * The implementation is largely based on the {@code BloomFilter} class from Guava.
 */
public abstract class BloomFilter {

  public enum Version {
    /**
     * {@code BloomFilter} binary format version 1. All values written in big-endian order:
     * <ul>
     *   <li>Version number, always 1 (32 bit)</li>
     *   <li>Number of hash functions (32 bit)</li>
     *   <li>Total number of words of the underlying bit array (32 bit)</li>
     *   <li>The words/longs (numWords * 64 bit)</li>
     * </ul>
     */
    V1(1);

    private final int versionNumber;

    Version(int versionNumber) {
      this.versionNumber = versionNumber;
    }

    int getVersionNumber() {
      return versionNumber;
    }
  }

  /**
   * Returns the probability that {@linkplain #mightContain(Object)} erroneously return {@code true}
   * for an object that has not actually been put in the {@code BloomFilter}.
   *
   * Ideally, this number should be close to the {@code fpp} parameter passed in
   * {@linkplain #create(long, double)}, or smaller. If it is significantly higher, it is usually
   * the case that too many items (more than expected) have been put in the {@code BloomFilter},
   * degenerating it.
   */
  public abstract double expectedFpp();

  /**
   * Returns the number of bits in the underlying bit array.
   */
  public abstract long bitSize();

  /**
   * Puts an item into this {@code BloomFilter}. Ensures that subsequent invocations of
   * {@linkplain #mightContain(Object)} with the same item will always return {@code true}.
   *
   * @return true if the bloom filter's bits changed as a result of this operation. If the bits
   *     changed, this is <i>definitely</i> the first time {@code object} has been added to the
   *     filter. If the bits haven't changed, this <i>might</i> be the first time {@code object}
   *     has been added to the filter. Note that {@code put(t)} always returns the
   *     <i>opposite</i> result to what {@code mightContain(t)} would have returned at the time
   *     it is called.
   */
  public abstract boolean put(Object item);

  /**
   * A specialized variant of {@link #put(Object)} that only supports {@code String} items.
   */
  public abstract boolean putString(String item);

  /**
   * A specialized variant of {@link #put(Object)} that only supports {@code long} items.
   */
  public abstract boolean putLong(long item);

  /**
   * A specialized variant of {@link #put(Object)} that only supports byte array items.
   */
  public abstract boolean putBinary(byte[] item);

  /**
   * Determines whether a given bloom filter is compatible with this bloom filter. For two
   * bloom filters to be compatible, they must have the same bit size.
   *
   * @param other The bloom filter to check for compatibility.
   */
  public abstract boolean isCompatible(BloomFilter other);

  /**
   * Combines this bloom filter with another bloom filter by performing a bitwise OR of the
   * underlying data. The mutations happen to <b>this</b> instance. Callers must ensure the
   * bloom filters are appropriately sized to avoid saturating them.
   *
   * @param other The bloom filter to combine this bloom filter with. It is not mutated.
   * @throws IncompatibleMergeException if {@code isCompatible(other) == false}
   */
  public abstract BloomFilter mergeInPlace(BloomFilter other) throws IncompatibleMergeException;

  /**
   * Returns {@code true} if the element <i>might</i> have been put in this Bloom filter,
   * {@code false} if this is <i>definitely</i> not the case.
   */
  public abstract boolean mightContain(Object item);

  /**
   * A specialized variant of {@link #mightContain(Object)} that only tests {@code String} items.
   */
  public abstract boolean mightContainString(String item);

  /**
   * A specialized variant of {@link #mightContain(Object)} that only tests {@code long} items.
   */
  public abstract boolean mightContainLong(long item);

  /**
   * A specialized variant of {@link #mightContain(Object)} that only tests byte array items.
   */
  public abstract boolean mightContainBinary(byte[] item);

  /**
   * Writes out this {@link BloomFilter} to an output stream in binary format. It is the caller's
   * responsibility to close the stream.
   */
  public abstract void writeTo(OutputStream out) throws IOException;

  /**
   * Reads in a {@link BloomFilter} from an input stream. It is the caller's responsibility to close
   * the stream.
   */
  public static BloomFilter readFrom(InputStream in) throws IOException {
    return BloomFilterImpl.readFrom(in);
  }

  /**
   * Computes the optimal k (number of hashes per item inserted in Bloom filter), given the
   * expected insertions and total number of bits in the Bloom filter.
   *
   * See http://en.wikipedia.org/wiki/File:Bloom_filter_fp_probability.svg for the formula.
   *
   * @param n expected insertions (must be positive)
   * @param m total number of bits in Bloom filter (must be positive)
   */
  private static int optimalNumOfHashFunctions(long n, long m) {
    // (m / n) * log(2), but avoid truncation due to division!
    return Math.max(1, (int) Math.round((double) m / n * Math.log(2)));
  }

  /**
   * Computes m (total bits of Bloom filter) which is expected to achieve, for the specified
   * expected insertions, the required false positive probability.
   *
   * See http://en.wikipedia.org/wiki/Bloom_filter#Probability_of_false_positives for the formula.
   *
   * @param n expected insertions (must be positive)
   * @param p false positive rate (must be 0 < p < 1)
   */
  private static long optimalNumOfBits(long n, double p) {
    return (long) (-n * Math.log(p) / (Math.log(2) * Math.log(2)));
  }

  static final double DEFAULT_FPP = 0.03;

  /**
   * Creates a {@link BloomFilter} with the expected number of insertions and a default expected
   * false positive probability of 3%.
   *
   * Note that overflowing a {@code BloomFilter} with significantly more elements than specified,
   * will result in its saturation, and a sharp deterioration of its false positive probability.
   */
  public static BloomFilter create(long expectedNumItems) {
    return create(expectedNumItems, DEFAULT_FPP);
  }

  /**
   * Creates a {@link BloomFilter} with the expected number of insertions and expected false
   * positive probability.
   *
   * Note that overflowing a {@code BloomFilter} with significantly more elements than specified,
   * will result in its saturation, and a sharp deterioration of its false positive probability.
   */
  public static BloomFilter create(long expectedNumItems, double fpp) {
    if (fpp <= 0D || fpp >= 1D) {
      throw new IllegalArgumentException(
        "False positive probability must be within range (0.0, 1.0)"
      );
    }

    return create(expectedNumItems, optimalNumOfBits(expectedNumItems, fpp));
  }

  /**
   * Creates a {@link BloomFilter} with given {@code expectedNumItems} and {@code numBits}, it will
   * pick an optimal {@code numHashFunctions} which can minimize {@code fpp} for the bloom filter.
   */
  public static BloomFilter create(long expectedNumItems, long numBits) {
    if (expectedNumItems <= 0) {
      throw new IllegalArgumentException("Expected insertions must be positive");
    }

    if (numBits <= 0) {
      throw new IllegalArgumentException("Number of bits must be positive");
    }

    return new BloomFilterImpl(optimalNumOfHashFunctions(expectedNumItems, numBits), numBits);
  }
}