Add bloom filters to parquet statistics

[asfimport]

For row groups with no dictionary, we could still produce a bloom filter. This could be very useful in filtering entire row groups.
Pull request:
#215

Reporter: Alex Levenson / @isnotinvain
Assignee: Junjie Chen / @chenjunjiedada

Subtasks:

• Define the parquet bloom filter statistics in parquet format
• [C++] Cross compatibility support with parquet-mr
• [C++] Bloom filter read/write implementation
• [java]Bloom filter read/write implementation
• [C++] Integrate Bloom filter into row group filter logic
• [C++] Add bloom filter utility class
• Add bloom filter utility class
• [C++] replace shared_ptr to unique_ptr in Bloom filter buffer allocation
• [C++] move Bloom filter test binary to parquet-testing repo
• [java] Integrate Bloom filter logic
• Support nested column Bloom filter
• Perform encoding before bloom filters write out
• Store Bloom filters near to footer.
• update hash naming of bloom filter
• Update parquet thrift to align with spec
• Resolve Bloom filter spec concerns
• merge bloom filter feature branch to master

Related issues:

• [C++] Add bloom filter utility class (Blocked)
• Implement the bloom filter for the ParquetSerde (blocks)
• Optional bloom_filter_offset is filled if no bloom filter is present (relates to)
• Use Parquet bloom filter (relates to)
• Refactor the configuration for bloom filters (relates to)
• Support writing bloom filters in Parquet (is related to)
• Leverage parquet bloom filters (is related to)

PRs and other links:

• GitHub Pull Request #431
• GitHub Pull Request #432
• GitHub Pull Request #112
• GitHub Pull Request #113
• GitHub Pull Request #62
• GitHub Pull Request #99
• GitHub Pull Request #425
• GitHub Pull Request #484
• GitHub Pull Request #757
• Bloom filter calculations: Filter size by FPP and uniqueness and expected FPP from overloading

_{Note: This issue was originally created as PARQUET-41. Please see the migration documentation for further details.}

[asfimport]

Alex Levenson / @isnotinvain:
This will require some planning and discussion around parquet-format.
We need to choose how we will serialize a bloom filter into parquet statistics, how it's size will be configured, etc.
But ultimately it needs to be a bloom filter easily deserialized from C++, and it needs to be durable (so java.io.Serializable or anything similar won't work here.)

[asfimport]

Daniel Weeks / @danielcweeks:
While it would be great to have them be compatible between java/C++/etc, that shouldn't necessarily be a requirement. A bloom filter is an optimization, so if the reader doesn't know how to interpret it, it can simply be ignored.

[asfimport]

Antonios Chalkiopoulos / @Antwnis:
I don't know if it helps - but in scala , using algebird, when i want to serialize a Bloom Filter into a file and then be able to re-read it from another job i do:

def serialize(dataStructure: Any): Array[Byte] = {
val stream = new ByteArrayOutputStream()
val out = new ObjectOutputStream(stream)
out.writeObject(dataStructure)
out.close()
stream.close()
stream.toByteArray
}

def deserialize[T](byteArray: Array[Byte]): T = {
val is = new ObjectInputStream(new ByteArrayInputStream(byteArray))
is.readObject().asInstanceOf[T]
}

and then from i.e. Scalding

pipe.mapTo('bloom -> 'serialized) { bf:BF => io.scalding.approximations.Utils.serialize(bf) }.write( ... )
pipe.mapTo('bloom -> 'serialized) { bf:BF => new String(io.scalding.approximations.Utils.serialize(bf)) }.write( ... )

Where BF is actually com.twitter.algebird.BF

[asfimport]

Alex Levenson / @isnotinvain:
If the data for the bloom filter is going to be stored in the parquet-format thrift statistics schema, I think it should be well a well defined format, that is not specific to java (for example, the format should not be whatever comes out of an ObjectOutputStream).
I think it's fine that the two runtimes support different features, but so far we have maintained binary compatibility between the two.

Even if we didn't care about interoperability, we still need a well defined binary format for the bloom filter that is durable over time, unlike plain java.io serialization.

[asfimport]

Alex Levenson / @isnotinvain:
Right, so while this works for temporary storage / transport of java objects, it's not a durable format with guaranteed stability over time. If the algebird bloom filter changes later, it will serialize differently, etc. So I think we need to still define a thrift / binary format for storing the bloom filter.

[asfimport]

Alex Levenson / @isnotinvain:
To clarify, I'm not saying the C++ has to use the bloom filter at runtime right away, but we should not use a binary format that would make it impossible / difficult for the C++ impl to use it if it wanted to. And I think it would also be bad news if each impl serialized bloom filters in their own way.

[asfimport]

Ferdinand Xu / @winningsix:
Hi @isnotinvain and @Antwnis, thank you for your comments.
I'd like to have a short summary for your summary. We should implement the BF by thrift other than some unstable framework library.
Regards to the size of bloom filter, I think we should configure it from the upper level since user can have a better understanding of the size of column numbers. Any thought about the size configuration?

[asfimport]

Julien Le Dem / @julienledem:
The first step is to specify how we would store the Bloom Filter in parquet-format.
As @isnotinvain mentioned it should be defined at the binary level.
A bloom filter is just a byte array (or possibly a few) and probably the spec of the hash functions used so it should not be that hard to defined.

Alternatively there are custom properties that can be added for non standard things but I would recommend going the way defined above for bloom filters as they are a standard we want to define.

listing a few folks that should probably review the format for this (from Impala, Drill, SparkSQL): [~marcelk] @nongli [~pipfiddle] @jaltekruse @rdblue [~marmbrus]

[asfimport]

Ryan Blue / @rdblue:

As Alex Levenson mentioned it should be defined at the binary level . . . and probably the spec of the hash functions used so it should not be that hard to defined.

+1

[asfimport]

Alex Levenson / @isnotinvain:
Yes, I was just thinking, the hash functions for each data type need to be well defined too but Julien beat me to it :)

So I would say next steps are:

1. Define a binary format for serializing a bloom filter into a byte array, including any configuration data (like size), and also discuss whether that should be global (stored once) or repeated into each bloom filter.

2. Define the hash function used for each data type

3. Create a java implementation of the above, possibly borrowing code snippets from Algebird or Guava

[asfimport]

Jacques Nadeau / @jacques-n:
Agree with @isnotinvain, @julienledem, and @rdblue. We need non-language-specific binary specification. I'd also argue that we need to see highly performant proof of concept code in C and Java before we can decide that a particular specification is acceptable.

[asfimport]

Prateek Rungta:
ORC implemented bloom filters recently. Their design doc - 1 - is worth a gander.

[asfimport]

Ryan Blue / @rdblue:
Thanks [~prateekrungta]! Looks like I need to go read "Less Hashing, Same Performance: Building a Better Bloom Filter" by Kirsch et. al. now.

[asfimport]

Ryan Blue / @rdblue:
@winningsix does the info about ORC or the paper help at all?

[asfimport]

Ferdinand Xu / @winningsix:
Thanks @rdblue and [~prateekrungta] for this information, I will take time look at it.