Binary column statistics error when reuse byte[] among rows

[asfimport]

I think it is a common practice when inserting table data as parquet file, one would always reuse the same object among rows, and if a column is byte[] of fixed length, the byte[] would also be reused.

If I use ByteArrayBackedBinary for my byte[], the bug occurs: All of the row groups created by a single task would have the same max & min binary value, just as the last row's binary content.

The reason is BinaryStatistic just keep max & min as parquet.io.api.Binary references, since I use ByteArrayBackedBinary for byte[], the real content of max & min would always point to the reused byte[], therefore the latest row's content.

Does parquet declare somewhere that the user shouldn't reuse byte[] for Binary type? If it doesn't, I think it's a bug and can be reproduced by Spark SQL's RowWriteSupport

The related Spark JIRA ticket: SPARK-6859

Reporter: Yijie Shen / @yjshen
Assignee: Ashish Singh / @SinghAsDev

Related issues:

• Release Parquet 1.8.0 (blocks)
• Improvements in ByteBuffer read path (Is contained by)
• Binary statistics are invalid if buffers are reused (is duplicated by)
• Binary statistics is not updated correctly if an underlying Binary array is modified in place (is duplicated by)

PRs and other links:

• PR #197

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

[asfimport]

Ryan Blue / @rdblue:
It sounds like we need to add a defensive copy to make sure the internal min and max byte arrays aren't changed, and work out a way to ignore min/max stats for all existing data.

[asfimport]

Jacques Nadeau / @jacques-n:
We have a fix for this already in our pull request (49) for PARQUET-77

[asfimport]

Ryan Blue / @rdblue:
@jacques-n, does that include the fix to ignore min/max byte arrays for data written with 1.6.0 and earlier?

[asfimport]

Yijie Shen / @yjshen:
@jacques-n, I checked the PR, It seems you just change

public void updateStats(Binary min_value, Binary max_value) {
    if (min.compareTo(min_value) > 0) { min = min_value; }
    if (max.compareTo(max_value) < 0) { max = max_value; }
  }

to

public void updateStats(Binary min_value, Binary max_value) {
    if (min.compareTo(min_value) > 0) { min = Binary.fromByteArray(min_value.getBytes()); }
    if (max.compareTo(max_value) < 0) { max = Binary.fromByteArray(max_value.getBytes()); }
}

As I mentioned in the description, the bug occurs when I reuse byte[], using ByteArrayBackedBinary as Binary, fromByteArray & getBytes are not really doing the "defensive copy" job mentioned by @rdblue, it's still passing the original byte[].

Correct me if I misunderstand your PR :)

[asfimport]

Jason Altekruse / @jaltekruse:
@yjshen It looks like the issue was only solved in the PR for the case where the new ByteBuffer backed implementation returned by the fromByteBuffer() method in the Binary.java class is used (or others that make a copy themselves). For most implementations of Binary it looks to be making a copy to return a byte[] from the getBytes() method. It looks like a more general approach will be needed if the Binary object may return an array it plans on re-using. That being said, as there is no way to resize an array, it would seem like this getBytes() method might be best to be defined to be required to return a copy, unless the Binary object is immutable or you are only sharing the byte[] where the values are of equal length. This seems to be the entire purpose for having the interface in the first place, which allows a buffer of a large length to stay allocated for all values and have the length set appropriately with each new value.

If this was enforced at the Binary interface level than I think the PR would solve the functional issue.

It does however create the unfortunate overhead of making a ton of copies if a column is full of increasing or decreasing values, it might be worth processing a small list of values to see if it finds a new min/max and make one defensive copy for that group.

[asfimport]

Alex Levenson / @isnotinvain:
I think a Binary is supposed to just be a unifying API over byte[] and various other byte[]-like types (slices of byte arrays, byte buffers, and so on).

So I think that should come with the same caveats as sharing a byte[] – if someone gives you a byte[] you might need to make a defensive copy, as in this case for the statistics. But in cases where you just use the Binary to write to a stream, no need for a copy.

I guess what I'm saying is, it's not Binary that should be responsible for handling shared state, but rather the user of it, right?

[asfimport]

Jason Altekruse / @jaltekruse:
For a design standpoint, I completely agree that it would be clearest to have the Binary objects maintain zero shared state.

I think my concern is that the interface that is trying to unify byte[] with other types might be ill fit to have a method that returns a bare byte[] without an explicit expectation that it be an immutable copy of the data being currently represented. An implementation that re-uses a buffer will not be able to return its byte[] alone if the buffer has a chance of being allocated to the wrong size for the current value, so it would need to make a copy anyway.

As far as writing to a stream is concerned, this is actually already covered by methods on Binary itself, which can write the current value to a stream with whatever method they need, specific to how the bytes are stored. I think adding interfaces for any other use cases is the cleanest way to do it. For example how the statistics want to use the data, it might make the most sense for the binary to include a clone method explicitly, that will create a copy of the binary in the most efficient manner possible. For Drill it would be useful if we could have the flexibility to return a new Binary backed by a direct memory buffer that we can track with our own allocator, for another type of binary, maybe one that internally has some amount of compression on individual values, would return a more efficient copy of itself than a copy that could be made of the byte[] returned from this method.

[asfimport]

Alex Levenson / @isnotinvain:
Yes, my point was that Binary is just a wrapper, and the decision to copy / not copy seems like it should be handled outside the wrapper.

So maybe all that is needed is to add a helper method to Binary called copyToByteArray()?
Though toByteArray() usually make a copy, so maybe we should just make that the contract?

Or just some consistency around ownership? If you pass a byte[] to a Binary, should we say the contract is that ownership has passed to the Binary and you shouldn't mutate it any more?

[asfimport]

Ashish Singh / @SinghAsDev:
Hey guys, I am planning to take a stab on this. After following the detailed discussion so far, I think (feel free to correct me) following needs to be done.

1. We need a clone() or copyToByteArray() method in Binary. Having the method should make it obvious to the user that the byte[] should not be mutated and if it plans on doing it clone() or copyToByteArray() should be used. However, this should be explicitly mentioned in the docs as well.

2. Update BinaryStatistics to use clone() or copyToByteArray() on the Binary value passed to it.

3. Ignore min/max byte arrays for data written with 1.6.0 and earlier.

If you guys agree with it, I should be able to submit a PR soon.

[asfimport]

Ashish Singh / @SinghAsDev:
@isnotinvain, @jaltekruse and @rdblue, if time permits can you guys can take a look at the suggestion above. I will wait for at least one of you guys to confirm that my suggestions are inline with what you guys have in mind.

[asfimport]

Jason Altekruse / @jaltekruse:
I think this makes sense to me, I would try to wait for Ryan or Alex's input. I am trying to put some more time into parquet, but I am not a committer on the project, so they would be the best for a final answer on what confirms to the patterns in the current code.

[asfimport]

Alex Levenson / @isnotinvain:
The only thing I want to check is whether we've got this backwards or not.

Binary already has a getBytes() method and I think that is exactly what we need, I don't think we need a clone() method do we? All the getBytes() methods do a copy except for ByteArrayBackedBinary. Maybe it should do a copy. Or maybe a copy should happen when ByteArrayBackedBinary is constructed.

The real question is, who "owns" the byte array passed to ByteArrayBackedBinary? It seems a little odd to "give" this array to ByteArrayBackedBinary and then continue to mutate it right?

But again, one way to think of Binary is as a view to some other byte source. So the real bug is in the statistics collection code path which should be making a defensive copy. But do we need Binary to manage this or not?

Another approach would be to make getBytes() always do a copy (which it currently does for everything except ByteArrayBackedBinary). Most code paths should be using the .writeTo methods anyway right?

[asfimport]

Konstantin Shaposhnikov / @kostya-sh:
I am still learning parquet code base so please take my comments with a grain of salt.

Doing a copy in BinaryStatistics is probably better as it potentially reduces number of copies (and garbage). Though in the worst case when binary field is sorted it will still create a copy for each update.

I think on the write path Binary.getBytes() almost always needs to return a copy. E.g. PlainBinaryDictionaryValuesWriter already does copy() before storing a Binary in the internal dictionary.

However I think that this is not needed on the read path. Which suggests that two methods are required.

One more possible cleanup could be to remove public methods from Statistics classes that are not used or used in tests only (e.g. genericGetMin/Max, getMax, getMin, etc). This will break backward compatibility so I am not sure if this can be done though.

[asfimport]

Ashish Singh / @SinghAsDev:
Thanks for the inputs guys. The reason I was suggesting to add a clone() method is that we do not change the existing behavior. However, on giving another thought, I think getBytes(), generally in java, returns a copy. Maybe we should be consistent with what developers/ users are probably already used to. The onus of cautiously using getBytes() for performance/ space considerations is on the clients.

Does the modified approach below makes sense?

1. Modify getBytes in ByteArrayBackedBinary to return a copy of backing byte array.
2. Update BinaryStatistics to use getBytes() on the Binary value passed to it only when it is updating its min or max.
3. Ignore min/max byte arrays for data written with 1.6.0 and earlier.

If you guys agree with it, I should be able to submit a PR soon.

[asfimport]

Alex Levenson / @isnotinvain:
@SinghAsDev I think that sounds good, but we will now be doing more copies than we used to, and only some of them will be necessary – I think we need to watch out for whether this is a big performance regression or not.

Another approach is to split Binary into two – ImmutableBinary and MutableBinary, and use these two types to signal who has ownership of the backing byte arrays (eg, ImmutableBinary is guaranteed to never be mutated underneath you, but MutableBinary require a copy to safely be held on to).