diff --git a/algebird-benchmark/src/main/scala/com/twitter/algebird/benchmark/MutableBloomFilterCreateBenchmark.scala b/algebird-benchmark/src/main/scala/com/twitter/algebird/benchmark/MutableBloomFilterCreateBenchmark.scala
new file mode 100644
index 000000000..4583c2575
--- /dev/null
+++ b/algebird-benchmark/src/main/scala/com/twitter/algebird/benchmark/MutableBloomFilterCreateBenchmark.scala
@@ -0,0 +1,54 @@
+package com.twitter.algebird.benchmark
+
+import com.twitter.algebird.mutable.{BloomFilter, BloomFilterAggregator, MutableBF}
+import com.twitter.algebird.benchmark.BloomFilterCreateBenchmark.createRandomString
+import org.openjdk.jmh.annotations._
+
+object MutableBloomFilterCreateBenchmark {
+
+  @State(Scope.Benchmark)
+  class BloomFilterState {
+    @Param(Array("100", "1000", "10000"))
+    var nbrOfElements: Int = 0
+
+    @Param(Array("0.001", "0.01"))
+    var falsePositiveRate: Double = 0
+
+    var randomStrings: Seq[String] = _
+
+    @Setup(Level.Trial)
+    def setup(): Unit =
+      randomStrings = createRandomString(nbrOfElements, 10)
+
+  }
+}
+class MutableBloomFilterBenchmark {
+
+  import MutableBloomFilterCreateBenchmark._
+
+  @Benchmark
+  def createMutableBloomFilter(bloomFilterState: BloomFilterState): MutableBF[String] = {
+    val bfMonoid = BloomFilter[String](bloomFilterState.nbrOfElements, bloomFilterState.falsePositiveRate)
+    val bf = bfMonoid.create(bloomFilterState.randomStrings: _*)
+    bf
+  }
+
+  @Benchmark
+  def createMutableBloomFilterUsingFold(bloomFilterState: BloomFilterState): MutableBF[String] = {
+    val bfMonoid = BloomFilter[String](bloomFilterState.nbrOfElements, bloomFilterState.falsePositiveRate)
+    val bf = bloomFilterState.randomStrings.foldLeft(bfMonoid.zero) {
+      case (filter, string) => filter += string
+    }
+    bf
+  }
+
+  @Benchmark
+  def createMutableBloomFilterAggregator(bloomFilterState: BloomFilterState): MutableBF[String] = {
+    val bfMonoid = BloomFilter[String](bloomFilterState.nbrOfElements, bloomFilterState.falsePositiveRate)
+    val bfAggregator = BloomFilterAggregator(bfMonoid)
+
+    val bf = bloomFilterState.randomStrings.aggregate(bfAggregator.monoid.zero)(_ += _, _ ++= _)
+    bf
+  }
+
+}
diff --git a/algebird-benchmark/src/main/scala/com/twitter/algebird/benchmark/MutableBloomFilterQueryBenchmark.scala b/algebird-benchmark/src/main/scala/com/twitter/algebird/benchmark/MutableBloomFilterQueryBenchmark.scala
new file mode 100644
index 000000000..f9388051d
--- /dev/null
+++ b/algebird-benchmark/src/main/scala/com/twitter/algebird/benchmark/MutableBloomFilterQueryBenchmark.scala
@@ -0,0 +1,36 @@
+package com.twitter.algebird
+package benchmark
+
+import org.openjdk.jmh.annotations._
+
+object MutableBloomFilterQueryBenchmark {
+
+  @State(Scope.Benchmark)
+  class BloomFilterState {
+
+    @Param(Array("100", "1000", "10000"))
+    var nbrOfElements: Int = 0
+
+    @Param(Array("0.001", "0.01"))
+    var falsePositiveRate: Double = 0
+
+    var bf: mutable.MutableBF[String] = _
+
+    @Setup(Level.Trial)
+    def setup(): Unit = {
+      val randomStrings =
+        BloomFilterCreateBenchmark.createRandomString(nbrOfElements, 10)
+      bf = mutable
+        .BloomFilter[String](nbrOfElements, falsePositiveRate)
+        .create(randomStrings: _*)
+    }
+  }
+}
+
+class MutableBloomFilterQueryBenchmark {
+  import MutableBloomFilterQueryBenchmark._
+
+  @Benchmark
+  def queryBloomFilter(bloomFilterState: BloomFilterState): ApproximateBoolean =
+    bloomFilterState.bf.contains("1")
+}
diff --git a/algebird-core/src/main/scala/com/twitter/algebird/mutable/BloomFilter.scala b/algebird-core/src/main/scala/com/twitter/algebird/mutable/BloomFilter.scala
new file mode 100644
index 000000000..af8ccec94
--- /dev/null
+++ b/algebird-core/src/main/scala/com/twitter/algebird/mutable/BloomFilter.scala
@@ -0,0 +1,377 @@
+/*
+Copyright 2019 Twitter, Inc.
+
+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.twitter.algebird.mutable
+
+import java.util
+
+import algebra.BoundedSemilattice
+import com.twitter.algebird.{
+  BFHash => _,
+  BloomFilter => ImmutableBloomFilter,
+  BF => _,
+  BloomFilterAggregator => _,
+  BloomFilterMonoid => _,
+  _
+}
+
+/**
+ * Helpers for creating Bloom Filters. Most implementations regarding estimations are same as
+ * Immutable Bloom Filters and these functions are aliases for them.
+ */
+object BloomFilter {
+
+  def apply[A](numEntries: Int, fpProb: Double)(implicit hash: Hash128[A]): BloomFilterMonoid[A] =
+    optimalWidth(numEntries, fpProb) match {
+      case None =>
+        throw new java.lang.IllegalArgumentException(
+          s"BloomFilter cannot guarantee the specified false positive probability for the number of entries! (numEntries: $numEntries, fpProb: $fpProb)")
+      case Some(width) =>
+        val numHashes = optimalNumHashes(numEntries, width)
+        BloomFilterMonoid[A](numHashes, width)(hash)
+    }
+
+  // Mostly an alias to actual functions defined for Immutable Bloom Filters.
+  def optimalNumHashes(numEntries: Int, width: Int): Int =
+    ImmutableBloomFilter.optimalNumHashes(numEntries, width)
+
+  def optimalWidth(numEntries: Int, fpProb: Double): Option[Int] =
+    ImmutableBloomFilter.optimalWidth(numEntries, fpProb)
+
+  /**
+   * Cardinality estimates are taken from Theorem 1 on page 15 of
+   * "Cardinality estimation and dynamic length adaptation for Bloom filters"
+   * by Papapetrou, Siberski, and Nejdl:
+   * http://www.softnet.tuc.gr/~papapetrou/publications/Bloomfilters-DAPD.pdf
+   *
+   * Roughly, by using bounds on the expected number of true bits after n elements
+   * have been inserted into the Bloom filter, we can go from the actual number of
+   * true bits (which is known) to an estimate of the cardinality.
+   *
+   * approximationWidth defines an interval around the maximum-likelihood cardinality
+   * estimate. Namely, the approximation returned is of the form
+   * (min, estimate, max) =
+   * ((1 - approxWidth) * estimate, estimate, (1 + approxWidth) * estimate)
+   */
+  def sizeEstimate(numBits: Int,
+                   numHashes: Int,
+                   width: Int,
+                   approximationWidth: Double = 0.05): Approximate[Long] =
+    ImmutableBloomFilter.sizeEstimate(numBits, numHashes, width, approximationWidth)
+
+}
+
+/**
+ * Bloom Filter - a probabilistic data structure to test presence of an element.
+ *
+ * Operations
+ * 1) insert: hash the value k times, updating the bitfield at the index equal to each hashed value
+ * 2) query: hash the value k times.  If there are k collisions, then return true; otherwise false.
+ *
+ * http://en.wikipedia.org/wiki/Bloom_filter
+ *
+ * This implementation of the BloomFilterMonoid is mutable and adding elements changes the
+ * filter. This is particularly useful when a filter needs to be created for a large number (>1M)
+ * of elements at once and fast.
+ */
+case class BloomFilterMonoid[A](numHashes: Int, width: Int)(implicit hash: Hash128[A])
+    extends Monoid[MutableBF[A]]
+    with BoundedSemilattice[MutableBF[A]] {
+  val hashes: MutableBFHash[A] = MutableBFHash[A](numHashes, width)(hash)
+
+  val zero: MutableBF[A] = MutableBFZero[A](hashes, width)
+
+  /**
+   * Adds the Bloom Filter on right to the left, mutating and returning Left.
+   * Assume that both have the same number of hashes and width.
+   */
+  override def plus(left: MutableBF[A], right: MutableBF[A]): MutableBF[A] =
+    left ++= right
+
+  override def sumOption(as: TraversableOnce[MutableBF[A]]): Option[MutableBF[A]] =
+    if (as.isEmpty) {
+      None
+    } else {
+      val outputInstance = MutableBFInstance.empty(hashes, width)
+      as.foreach { bf =>
+        outputInstance ++= bf
+      }
+      if (outputInstance.numBits == 0) {
+        Some(MutableBFZero(hashes, width))
+      } else {
+        Some(outputInstance)
+      }
+    }
+
+  /**
+   * Create a bloom filter with one item.
+   */
+  def create(item: A): MutableBF[A] = MutableBFInstance(hashes, width, item)
+
+  /**
+   * Create a bloom filter with multiple items.
+   */
+  def create(data: A*): MutableBF[A] = create(data.iterator)
+
+  /**
+   * Create a bloom filter with multiple items from an iterator
+   */
+  def create(data: Iterator[A]): MutableBF[A] = {
+    val outputInstance = MutableBFInstance.empty(hashes, width)
+    data.foreach { itm =>
+      outputInstance += itm
+    }
+    outputInstance
+  }
+
+}
+
+object MutableBF {
+  implicit def equiv[A]: Equiv[MutableBF[A]] =
+    new Equiv[MutableBF[A]] {
+      def equiv(a: MutableBF[A], b: MutableBF[A]): Boolean =
+        (a eq b) || ((a.numHashes == b.numHashes) &&
+          (a.width == b.width) &&
+          a.toBitSet.equals(b.toBitSet))
+    }
+}
+
+/**
+ * A Mutable Bloom Filter data structure
+ */
+sealed abstract class MutableBF[A] extends java.io.Serializable {
+  def numHashes: Int
+
+  def width: Int
+
+  /**
+   * The number of bits set to true in the bloom filter
+   */
+  def numBits: Int
+
+  /**
+   * Proportion of bits that are set to true.
+   */
+  def density = numBits.toDouble / width
+
+  def ++=(other: MutableBF[A]): MutableBF[A]
+
+  def +=(other: A): MutableBF[A]
+
+  def checkAndAdd(item: A): (MutableBF[A], ApproximateBoolean)
+
+  def contains(item: A): ApproximateBoolean =
+    if (maybeContains(item)) {
+      // The false positive probability (the probability that the Bloom filter erroneously
+      // claims that an element x is in the set when x is not) is roughly
+      // p = (1 - e^(-numHashes * setCardinality / width))^numHashes
+      // See: http://en.wikipedia.org/wiki/Bloom_filter#Probability_of_false_positives
+      //
+      // However, the true set cardinality may not be known. From empirical evidence, though,
+      // it is upper bounded with high probability by 1.1 * estimatedCardinality (as long as the
+      // Bloom filter is not too full), so we plug this into the formula instead.
+      // TODO: investigate this upper bound and density more closely (or derive a better formula).
+      // TODO: The following logic is same for immutable Bloom Filters and may be referred here.
+      val fpProb =
+        if (density > 0.95)
+          1.0 // No confidence in the upper bound on cardinality.
+        else
+          scala.math.pow(1 - scala.math.exp(-numHashes * size.estimate * 1.1 / width), numHashes)
+
+      ApproximateBoolean(true, 1 - fpProb)
+    } else {
+      // False negatives are not possible.
+      ApproximateBoolean.exactFalse
+    }
+
+  /**
+   * This may be faster if you don't care about evaluating
+   * the false positive probability
+   */
+  def maybeContains(item: A): Boolean
+
+  // Estimates the cardinality of the set of elements that have been
+  // inserted into the Bloom Filter.
+  def size: Approximate[Long]
+
+  def toBitSet: util.BitSet
+
+  def copy: MutableBF[A]
+
+  /**
+   * Compute the Hamming distance between the two Bloom filters
+   * `a` and `b`. The distance is defined as the number of bits that
+   * need to change to in order to transform one filter into the other.
+   * This is computed using XOR but it doesn't mutate any BloomFilters
+   */
+  def hammingDistance(that: MutableBF[A]): Int =
+    (this, that) match {
+      // Comparing with empty filter should give number
+      // of bits in other set
+      case (x: MutableBFZero[A], y: MutableBFZero[A]) => 0
+      case (x: MutableBFZero[A], y: MutableBF[A])     => y.numBits
+      case (x: MutableBF[A], y: MutableBFZero[A])     => x.numBits
+
+      // Otherwise compare as bit sets
+      case (_, _) =>
+        // hammingDistance should not mutate BloomFilter
+        val thisCopy = this.toBitSet.clone().asInstanceOf[util.BitSet]
+        thisCopy.xor(that.toBitSet)
+        thisCopy.cardinality()
+    }
+
+}
+
+/**
+ * Empty bloom filter.
+ */
+case class MutableBFZero[A](hashes: MutableBFHash[A], width: Int) extends MutableBF[A] {
+
+  def toBitSet: util.BitSet = new util.BitSet()
+
+  def numHashes: Int = hashes.size
+
+  def numBits = 0
+
+  def ++=(other: MutableBF[A]): MutableBF[A] = other
+
+  def +=(other: A): MutableBF[A] = MutableBFInstance[A](hashes, width, other)
+
+  def checkAndAdd(other: A): (MutableBF[A], ApproximateBoolean) =
+    (this += other, ApproximateBoolean.exactFalse)
+
+  override def contains(item: A) = ApproximateBoolean.exactFalse
+
+  def maybeContains(item: A): Boolean = false
+
+  def size = Approximate.exact[Long](0)
+
+  def copy: MutableBF[A] = MutableBFZero(hashes, width)
+}
+
+/*
+ * Mutable Bloom filter with multiple values
+ */
+case class MutableBFInstance[A](hashes: MutableBFHash[A], bits: util.BitSet, width: Int)
+    extends MutableBF[A] {
+
+  def numHashes: Int = hashes.size
+
+  /**
+   * The number of bits set to true
+   */
+  def numBits: Int = bits.cardinality()
+
+  def toBitSet: util.BitSet = bits
+
+  def ++=(other: MutableBF[A]): MutableBF[A] = {
+    require(this.width == other.width)
+    require(this.numHashes == other.numHashes)
+
+    other match {
+      case MutableBFZero(_, _)                => this
+      case MutableBFInstance(_, otherBits, _) =>
+        // assume same hashes used
+        bits.or(otherBits)
+        this
+    }
+  }
+
+  def +=(item: A): MutableBF[A] = {
+    val itemHashes = hashes(item)
+    itemHashes.foreach(bits.set)
+    this
+  }
+
+  def checkAndAdd(other: A): (MutableBF[A], ApproximateBoolean) = {
+    val doesContain = contains(other)
+    (this += other, doesContain)
+  }
+
+  def maybeContains(item: A): Boolean = {
+    val il = hashes(item)
+    var idx = 0
+    while (idx < il.length) {
+      val i = il(idx)
+      if (!bits.get(i)) return false
+      idx += 1
+    }
+    true
+  }
+
+  // use an approximation width of 0.05
+  def size: Approximate[Long] =
+    BloomFilter.sizeEstimate(numBits, numHashes, width, 0.05)
+
+  def copy: MutableBF[A] = MutableBFInstance(hashes, bits.clone.asInstanceOf[util.BitSet], width)
+}
+
+object MutableBFInstance {
+  def apply[A](hashes: MutableBFHash[A], width: Int, firstElement: A): MutableBF[A] = {
+    val bf = MutableBFInstance.empty(hashes, width)
+    bf += firstElement
+  }
+
+  def apply[A](hashes: MutableBFHash[A], width: Int): MutableBF[A] =
+    empty(hashes, width)
+
+  def empty[A](hashes: MutableBFHash[A], width: Int): MutableBF[A] =
+    MutableBFInstance(hashes, new util.BitSet(), width)
+}
+
+/**
+ * Logic for creating `n` hashes for each item of the BloomFilter.
+ *
+ * The hash functions derived here are different than the one used for
+ * com.twitter.algebird.BloomFilter and hence a com.twitter.algebird.mutable.BloomFilter
+ * is incompatible with the immutable one and cannot be converted to one another.
+ *
+ * This hash function derivation is explained by Adam Kirsch and Michael Mitzenmacher here:
+ * https://www.eecs.harvard.edu/~michaelm/postscripts/esa2006a.pdf
+ *
+ * We have noticed 2 to 4 times higher throughput when using this approach compared the one
+ * for the immutable filter.
+ */
+case class MutableBFHash[A](numHashes: Int, width: Int)(implicit hash128: Hash128[A]) {
+  val size: Int = numHashes
+
+  def apply(valueToHash: A): Array[Int] = {
+    val (hash1, hash2) = hash128.hashWithSeed(numHashes, valueToHash)
+
+    val hashes = new Array[Int](numHashes)
+    var i = 0
+    while (i < numHashes) {
+      // We just need two 32 bit hashes. So just convert toInt, and ignore the rest.
+      hashes(i) = math.abs((hash1.toInt + i * hash2.toInt) % width)
+      i += 1
+    }
+    hashes
+  }
+}
+
+case class BloomFilterAggregator[A](bfMonoid: BloomFilterMonoid[A])
+    extends MonoidAggregator[A, MutableBF[A], MutableBF[A]] {
+  val monoid = bfMonoid
+
+  def prepare(value: A) = monoid.create(value)
+
+  def present(bf: MutableBF[A]) = bf
+}
+
+object BloomFilterAggregator {
+  def apply[A](numHashes: Int, width: Int)(implicit hash: Hash128[A]): BloomFilterAggregator[A] =
+    BloomFilterAggregator[A](BloomFilterMonoid[A](numHashes, width))
+}
diff --git a/algebird-test/src/test/scala/com/twitter/algebird/mutable/BloomFilterTest.scala b/algebird-test/src/test/scala/com/twitter/algebird/mutable/BloomFilterTest.scala
new file mode 100644
index 000000000..6832a0038
--- /dev/null
+++ b/algebird-test/src/test/scala/com/twitter/algebird/mutable/BloomFilterTest.scala
@@ -0,0 +1,407 @@
+package com.twitter.algebird.mutable
+
+import java.io.{ByteArrayOutputStream, ObjectOutputStream}
+
+import com.twitter.algebird.{
+  BloomFilter => _,
+  BloomFilterAggregator => _,
+  BloomFilterMonoid => _,
+  BFHash => _,
+  BF => _,
+  _
+}
+import org.scalacheck.{Arbitrary, Gen}
+import org.scalatest.{Matchers, WordSpec}
+import org.scalacheck.Prop._
+
+class BloomFilterLaws extends CheckProperties {
+
+  import com.twitter.algebird.BaseProperties._
+
+  val NUM_HASHES = 6
+  val WIDTH = 32
+
+  implicit val bfMonoid = new BloomFilterMonoid[String](NUM_HASHES, WIDTH)
+
+  implicit val bfGen: Arbitrary[MutableBF[String]] =
+    Arbitrary {
+      val item = Gen.choose(0, 10000).map { v =>
+        bfMonoid.create(v.toString)
+      }
+      val zero = Gen.const(bfMonoid.zero)
+      Gen.frequency((1, zero), (10, item))
+    }
+
+  property("BloomFilter is a Monoid") {
+    commutativeMonoidLaws[MutableBF[String]]
+  }
+
+  property("++= is the same as plus") {
+    forAll { (a: MutableBF[String], b: MutableBF[String]) =>
+      Equiv[MutableBF[String]].equiv(a ++= b, bfMonoid.plus(a, b))
+    }
+  }
+
+  property("the distance between a filter and itself should be 0") {
+    forAll { (a: MutableBF[String]) =>
+      a.hammingDistance(a) == 0
+    }
+  }
+
+  property(
+    "the distance between a filter and an empty filter should be the number of bits" +
+      "set in the existing filter") {
+    forAll { (a: MutableBF[String]) =>
+      a.hammingDistance(bfMonoid.zero) == a.numBits
+    }
+  }
+
+  property("all equivalent filters should have 0 Hamming distance") {
+    forAll { (a: MutableBF[String], b: MutableBF[String]) =>
+      if (Equiv[MutableBF[String]].equiv(a, b))
+        a.hammingDistance(b) == 0
+      else {
+        val dist = a.hammingDistance(b)
+        (dist > 0) && (dist <= a.width)
+      }
+    }
+  }
+
+  property("distance between filters should be symmetrical") {
+    forAll { (a: MutableBF[String], b: MutableBF[String]) =>
+      a.hammingDistance(b) == b.hammingDistance(a)
+    }
+  }
+
+  property("calculating hamming distance should not modify the Bloom Filter") {
+    forAll { (a: MutableBF[String], b: MutableBF[String]) =>
+      val acopy = a.copy
+      a.hammingDistance(b)
+      Equiv[MutableBF[String]].equiv(a, acopy)
+    }
+  }
+
+  property("+ is the same as adding with create") {
+    forAll { (a: MutableBF[String], b: String) =>
+      Equiv[MutableBF[String]].equiv(a += b, bfMonoid.plus(a, bfMonoid.create(b)))
+    }
+  }
+
+  property("maybeContains is consistent with contains") {
+    forAll { (a: MutableBF[String], b: String) =>
+      a.maybeContains(b) == a.contains(b).isTrue
+    }
+  }
+
+  property("after + maybeContains is true") {
+    forAll { (a: MutableBF[String], b: String) =>
+      (a += b).maybeContains(b)
+    }
+  }
+
+  property("checkAndAdd works like check the add") {
+    forAll { (a: MutableBF[String], b: String) =>
+      val (next, check) = a.copy.checkAndAdd(b) // Treat as immutable BF by creating copies
+      val next1 = a.copy += b
+
+      Equiv[MutableBF[String]].equiv(next, next1) &&
+      (check == a.contains(b))
+    }
+  }
+
+  property("a ++ a = a for BF") {
+    forAll { (a: MutableBF[String]) =>
+      Equiv[MutableBF[String]].equiv(a ++= a, a)
+    }
+  }
+}
+
+class BFHashIndices extends CheckProperties {
+
+  val NUM_HASHES = 10
+  val WIDTH = 4752800
+
+  implicit val bfHash: Arbitrary[MutableBFHash[String]] =
+    Arbitrary {
+      for {
+        hashes <- Gen.choose(1, 10)
+        width <- Gen.choose(100, 5000000)
+      } yield MutableBFHash[String](hashes, width)
+    }
+
+  property("Indices are non negative") {
+    forAll { (hash: MutableBFHash[String], v: Long) =>
+      hash.apply(v.toString).forall { e =>
+        e >= 0
+      }
+    }
+  }
+
+}
+
+class BloomFilterFalsePositives[T: Gen: Hash128](falsePositiveRate: Double) extends ApproximateProperty {
+
+  type Exact = Set[T]
+  type Approx = MutableBF[T]
+
+  type Input = T
+  type Result = Boolean
+
+  val maxNumEntries = 1000
+
+  def exactGenerator =
+    for {
+      numEntries <- Gen.choose(1, maxNumEntries)
+      set <- Gen.containerOfN[Set, T](numEntries, implicitly[Gen[T]])
+    } yield set
+
+  def makeApproximate(set: Set[T]) = {
+    val bfMonoid = BloomFilter[T](set.size, falsePositiveRate)
+
+    val values = set.toSeq
+    bfMonoid.create(values: _*)
+  }
+
+  def inputGenerator(set: Set[T]) =
+    for {
+      randomValues <- Gen.listOfN[T](set.size, implicitly[Gen[T]])
+      x <- Gen.oneOf((set ++ randomValues).toSeq)
+    } yield x
+
+  def exactResult(s: Set[T], t: T) = s.contains(t)
+
+  def approximateResult(bf: MutableBF[T], t: T) = bf.contains(t)
+}
+
+class BloomFilterCardinality[T: Gen: Hash128] extends ApproximateProperty {
+
+  type Exact = Set[T]
+  type Approx = MutableBF[T]
+
+  type Input = Unit
+  type Result = Long
+
+  val maxNumEntries = 10000
+  val falsePositiveRate = 0.01
+
+  def exactGenerator =
+    for {
+      numEntries <- Gen.choose(1, maxNumEntries)
+      set <- Gen.containerOfN[Set, T](numEntries, implicitly[Gen[T]])
+    } yield set
+
+  def makeApproximate(set: Set[T]) = {
+    val bfMonoid = BloomFilter[T](set.size, falsePositiveRate)
+
+    val values = set.toSeq
+    bfMonoid.create(values: _*)
+  }
+
+  def inputGenerator(set: Set[T]) = Gen.const(())
+
+  def exactResult(s: Set[T], u: Unit) = s.size
+  def approximateResult(bf: MutableBF[T], u: Unit) = bf.size
+}
+
+class BloomFilterProperties extends ApproximateProperties("BloomFilter") {
+  import ApproximateProperty.toProp
+
+  for (falsePositiveRate <- List(0.1, 0.01, 0.001)) {
+    property(s"has small false positive rate with false positive rate = $falsePositiveRate") = {
+      implicit val intGen = Gen.choose(1, 1000)
+      toProp(new BloomFilterFalsePositives[Int](falsePositiveRate), 50, 50, 0.01)
+    }
+  }
+
+  property("approximate cardinality") = {
+    implicit val intGen = Gen.choose(1, 1000)
+    toProp(new BloomFilterCardinality[Int], 50, 1, 0.01)
+  }
+}
+
+class BloomFilterTest extends WordSpec with Matchers {
+
+  val RAND = new scala.util.Random
+
+  "MutableBloomFilter" should {
+
+    "be possible to create from an iterator" in {
+      val bfMonoid = new BloomFilterMonoid[String](RAND.nextInt(5) + 1, RAND.nextInt(64) + 32)
+      val entries = (0 until 100).map(_ => RAND.nextInt.toString)
+      val bf = bfMonoid.create(entries.iterator)
+      assert(bf.isInstanceOf[MutableBF[String]])
+    }
+
+    "be possible to create from a sequence" in {
+      val bfMonoid = new BloomFilterMonoid[String](RAND.nextInt(5) + 1, RAND.nextInt(64) + 32)
+      val entries = (0 until 100).map(_ => RAND.nextInt.toString)
+      val bf = bfMonoid.create(entries: _*)
+      assert(bf.isInstanceOf[MutableBF[String]])
+    }
+
+    "identify all true positives" in {
+      (0 to 100).foreach { _ =>
+        {
+          val bfMonoid = new BloomFilterMonoid[String](RAND.nextInt(5) + 1, RAND.nextInt(64) + 32)
+          val numEntries = 5
+          val entries = (0 until numEntries).map(_ => RAND.nextInt.toString)
+          val bf = bfMonoid.create(entries: _*)
+
+          entries.foreach { i =>
+            assert(bf.contains(i.toString).isTrue)
+          }
+        }
+      }
+    }
+
+    "have small false positive rate" in {
+      val iter = 10000
+
+      Seq(0.1, 0.01, 0.005).foreach { fpProb =>
+        {
+          val fps = (0 until iter).par.map { _ =>
+            {
+              val numEntries = RAND.nextInt(10) + 1
+
+              val bfMonoid = BloomFilter[String](numEntries, fpProb)
+
+              val entries = RAND
+                .shuffle((0 until 1000).toList)
+                .take(numEntries + 1)
+                .map(_.toString)
+              val bf = bfMonoid.create(entries.drop(1): _*)
+
+              if (bf.contains(entries(0)).isTrue) 1.0 else 0.0
+            }
+          }
+
+          val observedFpProb = fps.sum / fps.size
+
+          // the 5 is a fudge factor to make the probability of it relatively low
+          // in tests - This is different from the immutable implementation
+          // as the underlying hash functions are different.
+          assert(observedFpProb <= 5 * fpProb)
+        }
+      }
+    }
+
+    "approximate cardinality" in {
+      val bfMonoid = BloomFilterMonoid[String](10, 100000)
+      Seq(10, 100, 1000, 10000).foreach { exactCardinality =>
+        val items = (1 until exactCardinality).map { _.toString }
+        val bf = bfMonoid.create(items: _*)
+        val size = bf.size
+
+        assert(size ~ exactCardinality)
+        assert(size.min <= size.estimate)
+        assert(size.max >= size.estimate)
+      }
+    }
+
+    "work as an Aggregator" in {
+      (0 to 10).foreach { _ =>
+        {
+          val aggregator = BloomFilterAggregator[String](RAND.nextInt(5) + 1, RAND.nextInt(64) + 32)
+          val numEntries = 5
+          val entries = (0 until numEntries).map(_ => RAND.nextInt.toString)
+          val bf = aggregator(entries)
+
+          entries.foreach { i =>
+            assert(bf.contains(i.toString).isTrue)
+          }
+        }
+      }
+    }
+
+    "not serialize @transient BFInstance" in {
+      def serialize(bf: MutableBF[String]): Array[Byte] = {
+        val stream = new ByteArrayOutputStream()
+        val out = new ObjectOutputStream(stream)
+        out.writeObject(bf)
+        out.close()
+        stream.close()
+        stream.toByteArray
+      }
+
+      val items = (1 until 10).map(_.toString)
+      val bf = BloomFilter[String](10, 0.1).create(items: _*)
+      val bytesBeforeSizeCalled = Bytes(serialize(bf))
+      val beforeSize = bf.size
+      assert(bf.contains("1").isTrue)
+      val bytesAfterSizeCalled = Bytes(serialize(bf))
+      assert(bytesBeforeSizeCalled.size == bytesAfterSizeCalled.size)
+      assert(beforeSize == bf.size)
+    }
+
+    "not have negative hash values" in {
+      val NUM_HASHES = 2
+      val WIDTH = 4752800
+      val bfHash = MutableBFHash[String](NUM_HASHES, WIDTH)
+      val s = "7024497610539761509"
+      val index = bfHash.apply(s).head
+
+      assert(index >= 0)
+    }
+  }
+
+  "BloomFilter method `checkAndAdd`" should {
+
+    "be identical to method `+`" in {
+      (0 to 100).foreach { _ =>
+        val bfMonoid = new BloomFilterMonoid[String](RAND.nextInt(5) + 1, RAND.nextInt(64) + 32)
+        val numEntries = 5
+        val entries = (0 until numEntries).map(_ => RAND.nextInt.toString)
+        val bf = bfMonoid.create(entries: _*)
+        val bfWithCheckAndAdd = entries
+          .map { entry =>
+            (entry, bfMonoid.create(entry))
+          }
+          .foldLeft((bfMonoid.zero, bfMonoid.zero)) {
+            case ((left, leftAlt), (entry, right)) =>
+              val (newLeftAlt, contained) = leftAlt.checkAndAdd(entry)
+              left.contains(entry) shouldBe contained
+              (left += entry, newLeftAlt)
+          }
+
+        entries.foreach { i =>
+          assert(bf.contains(i.toString).isTrue)
+        }
+      }
+    }
+  }
+
+  "BloomFilters" should {
+
+    /**
+     *  The distances are different from the immutable bloom filter implementation
+     *  as they use a different method to find the hashes.
+     */
+    "be able to compute Hamming distance to each other" in {
+
+      def createBFWithItems(entries: Seq[String]): MutableBF[String] = {
+        val numOfHashes = 3
+        val width = 64
+        val bfMonoid = new BloomFilterMonoid[String](numOfHashes, width)
+        bfMonoid.create(entries: _*)
+      }
+
+      val firstBloomFilter = createBFWithItems(Seq("A"))
+      val secondBloomFilter = createBFWithItems(Seq("C"))
+
+      val distance1 = firstBloomFilter.hammingDistance(secondBloomFilter)
+      assert(distance1 === 6)
+
+      val thirdBloomFilter = createBFWithItems(Seq("A", "B", "C"))
+      val forthBloomFilter = createBFWithItems(Seq("C", "D", "E"))
+
+      val distance2 = thirdBloomFilter.hammingDistance(forthBloomFilter)
+      assert(distance2 === 6)
+
+      val emptyBloomFilter = createBFWithItems(List())
+      val distanceToEmpty = thirdBloomFilter.hammingDistance(emptyBloomFilter)
+      assert(distanceToEmpty === thirdBloomFilter.numBits)
+
+    }
+  }
+
+}