[SPARK-12936][SQL] Initial bloom filter implementation

[cloud-fan]

This PR adds an initial implementation of bloom filter in the newly added sketch module. The implementation is based on the BloomFilter class in guava.

Some difference from the design doc:

• expose bitSize instead of sizeInBytes to user.
• always need the expectedInsertions parameter when create bloom filter.

[cloud-fan]

cc @rxin @liancheng

[SparkQA]

Test build #49939 has finished for PR 10883 at commit bbf3822.

• This patch fails RAT tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • public abstract class BloomFilter
  • public class DefaultBloomFilter extends BloomFilter
  • static final class BitArray

[SparkQA]

Test build #49940 has finished for PR 10883 at commit 2db0171.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • public abstract class BloomFilter
  • public class DefaultBloomFilter extends BloomFilter
  • static final class BitArray

[SparkQA]

Test build #49941 has finished for PR 10883 at commit 1617226.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • public abstract class BloomFilter
  • public class DefaultBloomFilter extends BloomFilter
  • static final class BitArray

[rxin]

we should be consistent in naming with the count-min sketch one, i.e. rename this BloomFilterImpl.

[cloud-fan]

The string part is same with count-min, but long part is different, is it worth to abstract it? cc @rxin

[rxin]

Why is the long part different?

[cloud-fan]

because they use different strategy to produce n hash values for a long.

[rxin]

let's add some inline comment explaining why they are different

[SparkQA]

Test build #50002 has finished for PR 10883 at commit 920f292.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[liancheng]

I think it's OK to use allItems.take(numInsertion).foreach(filter.put) for simplicity if this code path doesn't slow down test execution too much.

[SparkQA]

Test build #50005 has finished for PR 10883 at commit 3633952.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #50012 has finished for PR 10883 at commit 4fce26e.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[liancheng]

Does the 0.001 here stand for epsilon? Should we wrap 0.03 - filter1.expectedFpp() with Math.abs()?

[liancheng]

Do we still need to special case "Byte" here when we already have numItems (it's 200 for Byte type)?

[liancheng]

Would be nice to make 0.01 a private static constant, e.g. EPSILON.

[liancheng]

Actually we should swap 0.03 and filter1.expectedFpp() here.

[liancheng]

One high level comment about testing code: I'm a little bit confused by those magic numbers there. Would be nice to name them.

[liancheng]

Maybe we should have a public static constant DEFAULT_EXPECTED_FPP for this 0.03. I was pretty confused when finding the magic number 0.03 in the testing code.

[liancheng]

Tip: I renamed CountMinSketchMergeException to IncompatibleMergeException and made it checked in #10893. You can use it in mergeInPlace.

[liancheng]

Would be nice to add another test case for incompatible merge (like this one).

[SparkQA]

Test build #50031 has finished for PR 10883 at commit b850bfd.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #50034 has finished for PR 10883 at commit a9a6e83.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • \"Cannot merge bloom filter of class \" + other.getClass().getName()

[liancheng]

LGTM

[rxin]

As discussed offline, it might be better to have put(String) and put(Long) in addition to put(Object).

I'm going to merge this pull request. Please address the remaining comments in a new pull request.