Rocksdb: Merged values can silently exceed the size limitation of byte[] in Java.

[StefanRRichter]

RocksDB's JNI bridge currently only offers byte[] as type for keys and values. Not only does this limits the maximum size of keys and values to 2^31, but it can also cause a problem with the merge operator.

When we merge multiple values under one key, they are internally concatenated to one value in RocksDB. This way, it is possible to accumulate a total value size per key that is larger than 2^31 bytes. Whenever we try to access such values that grew beyond this limit through merging, we will encounter an ArrayIndexOutOfBoundsException at best and a segfault at worst (I have seen this when the state grew a lot larger).

This behaviour is particularly problematic, because RocksDB accepts to store all the values, but on the next access the code fails unexpectedly.

What are your thoughts on this problem and is there any workaround that I missed?

Right now, I think the only proper long term solution to this is building RocksDB's JNI bridge around (Direct)ByteBuffer as input and output types, instead of simple byte[]. This would also enable larger keys and values, and potentially avoid copies when interacting through JNI.

[sagar0]

@StephanEwen also mentioned this in-person recently, and this is a valid concern.
@adamretter what do you think? Are there any good workarounds?

[adamretter]

Java uses int internally for the array index, however arrays are zero based addressing, so if I am not mistaken then the maximum theoretical size of a Java byte array is 2^31-1. However, it seems in practice that it can actually be a little less than this depending on various VM and memory limits.

So that would allow you to have about 2GB of storage.

@StefanRRichter Do you really have single values which are ~2GB? It would seem to be that is quite unlikely.

[StefanRRichter]

@adamretter This problem came up for some users of Apache Flink. We use RocksDB as a backend for stateful stream processing. We use merging state to collect events, e.g. events that happen as part of user sessions, over time windows. It seems that there are users with that have many events and/or potentially large time windows (days, weeks). Those users are hitting this boundary. While in some cases there might be workarounds in the way the users write they jobs, it would be better to enable them to use more then 2GB of state per key/window or at least fail fast and with a proper exception in case we cannot go past 2GB.

[adamretter]

@sagar0 @siying Can you guys comment on whether 2GB for a single value in RocksDB is a sensible idea?

[StefanRRichter]

The alternative for our use case would be to introduce artificial subkeys, having one key/value pair per event. Would that make more sense from a performance point of view or just introduce more overhead due to redundant prefix keys? I assumed avoiding this that this is the exact case for using merge operator in the first place?

[StephanEwen]

Is there anything forbidding this in principle? Form the Apache Flink use case perspective, degraded performance for such large values would be okay (much better than crashes) - users can look at their metrics and see why (super large values).

If there was a Java interface that did not use byte[] but rather something that mimics the C++ interface (a buffer_pointer (long) and a buffer_length (long) to return the result through), would that help circumvent the problem?

From Flink's perspective, the buffer_pointer and buffer_length would be very desirable anyways and we could see if we can help out there...

[siying]

@adamretter RocksDB does support 2GB single value. I do have a (disabled) use case that verifies we can handle it. It's hard to imagine why people use RocksDB in this way though. Whether we need to support it in Java depends on whether there is a need. I'm OK with either way.

[StefanRRichter]

@siying If it is hard to imagine why people would do this, what is the best practise you would suggest instead? In case you want to collect (i.e. append) event's values for time windows under a composite key of (window_id | extract_key(event)) and iterate those event values as soon as the window is evaluated? For me this sounded like a good case for applying the merge operator and it seems that there are cases where users have very large states per key/window pair.

[gfosco]

Closing this via automation due to lack of activity. If discussion is still needed here, please re-open or create a new/updated issue.

[adamretter]

@StefanRRichter As I understand it I think you need to set the option max_successive_merges, this will ensure that the merges are merged down to a concrete value when a maximum number of merge operations are present, see https://github.com/facebook/rocksdb/blob/04c11b867df9190da204e38357a14d20296fb244/include/rocksdb/advanced_options.h#L530

@siying anything further to add on this?

[StefanRRichter]

@adamretter Thanks a lot for your input on this topic. From a user perspective I find it a bit unfortunate that you have to know and tune this parameter to stop errors that can lead to data loss. Even if you know about it, a "safe" choice for the parameter can depend on the size of the merged values, which is hard to know upfront if you are using RocksDB as a backend inside another system without prior knowledge about the potential values that users could merge when using it. I think the fix from @azagrebin is already a good improvement to catch the error eventually (but unfortunately after it is already too late from a data loss point of view) and report a proper cause. In the long term, I wonder if we could also expose support for larger values through JNI, potentially also without doing a copy by reading from a pinned slice directly?

[adamretter]

We could potentially expose larger values through RocksJava, but the best
mechanism to do so is unclear to me. Both byte[] and ByteBuffer are limited
to 32bits of storage size. Also I really wonder if it is desirable to be
copying around 2GB+ arrays in memory, especially as each represents a
single key or value.

On Wed, 16 May 2018, 13:39 Stefan Richter, [e-mail] wrote:

@adamretter https://github.com/adamretter Thanks a lot for your input
on this topic. From a user perspective I find it a bit unfortunate that you
have to know and tune this parameter to stop errors that can lead to data
loss. Even if you know about it, a "safe" choice for the parameter can
depend on the size of the merged values, which is hard to know upfront if
you are using RocksDB as a backend inside another system without prior
knowledge about the potential values that users could merge when using it.
I think the fix from @azagrebin https://github.com/azagrebin is already
a good improvement to catch the error and report a proper cause. In the
long term, I wonder if we could also expose support for larger values
through JNI?

—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub
https://github.com/facebook/rocksdb/issues/2383#issuecomment-389433197,
or mute the thread
https://github.com/notifications/unsubscribe-auth/ABNJuV0fvXgBdyrPX3N6ROleCq9gBVq0ks5ty97GgaJpZM4NpSv6
.

[StefanRRichter]

Yes, the size limitations of ByteBuffer and byte[] are problematic, and copying data can be as well. I think a solution could be based on the same principle as a direct byte buffer, but would require a custom implementation that supports long typed offsets and indexes. I am not familar with the internal memory management of RocksDB, but would it be possible to pin the slice with the value data and return the memory address via JNI so that we can use this in Java to initialize the memory address inside our custom "ReadOnlyBigDirectyByteBuffer", which then operates on the internal data without creating a copy. I guess on downside, the buffer would require to be AutoCloseable to unpin the internal data after usage.

[adamretter]

Something like that would indeed be possible. We would want to read in
chunks though to avoid too many JNI boundary calls (which have quite the
performance detriment).

On Wed, 16 May 2018, 14:05 Stefan Richter, [e-mail] wrote:

Yes, the size limitations of ByteBuffer and byte[] are problematic, and
copying data can be as well. I think a solution could be based on the same
principle as a direct byte buffer, but would require a custom
implementation that supports long typed offsets and indexes. I am not
familar with the internal memory management of RocksDB, but would it be
possible to pin the slice with the value data and return the memory address
via JNI so that we can use this in Java to initialize the memory address
inside our custom "ReadOnlyBigDirectyByteBuffer", which then operates on
the internal data without creating a copy. I guess on downside, the buffer
would require to be AutoCloseable to unpin the internal data after usage.

—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub
https://github.com/facebook/rocksdb/issues/2383#issuecomment-389440363,
or mute the thread
https://github.com/notifications/unsubscribe-auth/ABNJueAKb-oJPV672Nn9DYYqKuCsVBggks5ty-TPgaJpZM4NpSv6
.

[StefanRRichter]

I think that would be really nice to have! I think that a more direct data transfer out of (also potentially into) of RockDB with zero data copy could also be beneficial for performance. Do you think there is a realistic chance to see something like that in the near future? Would an outside contribution help and have a chance to make it in?

[StephanEwen]

@adamretter If we would follow a similar approach to DirectByteBuffer, then there is no JNI boundary per access to the buffer - that is efficiently handled by Unsafe / VarHandle. The JNI boundary would be crossed once per read/write row call, similar to the current API.

[unoexperto]

@StefanRRichter @StephanEwen Hi guys! I just stumbled up on this problem with 6.7.3 and I was wondering if there is plan to address this edge case ? Maybe we can have an API that returns linked list of ByteBuffers or PinnableSlices instead of copying byte[] (in iterator value() and get()) ? In majority of use-cases I don't even need raw byte array because it's parsed to some Java object.

As work-around I'd like to be able to know in advance (before I call RocksIterator.value()) size of the value. Is it possible in existing API ? Or perhaps it would be nice to ignore merge operation if size of value exceeded some limit.

[StephanEwen]

That would be a question for @adamretter

I personally would be super happy to see an API that returns a native ByteBuffer, avoids a copy and object allocation on the way. If there are any plans for that, we'd be happy to contribute.

[adamretter]

@unoexperto @StephanEwen Hey guys, let's revisit this...

We now in 6.7.3 and newer have ByteBuffer versions of most functions, to which you can pass a Direct ByteBuffer. So that should eliminate some inefficiency with copies etc.

Regards large merge queues I think we suggested that max_successive_merges should be used.

Was there another use-case for having single (i.e. merged) values that are 2GB in size? If so, wouldn't this be a better use-case for BlobDB?

[unoexperto]

@adamretter Thank you for prompt response!

We now in 6.7.3 and newer have ByteBuffer versions of most functions, to which you can pass a Direct ByteBuffer. So that should eliminate some inefficiency with copies etc.

Could you please point to Java classes that allow getting ByteBuffer from iterator or use ByteBuffer in put(), get(), merge() calls ? I'm on 6.7.3 and all these methods still operate with byte[].

Regards large merge queues I think we suggested that max_successive_merges should be used.

I understood max_successive_merges only as optimization that defines how many merge operations should be in WAL until those merged and actual value is updated. So it's not limiting number of total merges per KV pair. Am I correct ?

Was there another use-case for having single (i.e. merged) values that are 2GB in size? If so, wouldn't this be a better use-case for BlobDB?

My use-case was similar to what you guys did at Facebook with LogDevice. I store logs in 10 mins chunks where key is <CustomerID><EpochOfSegment><LogLevel>. I relied on native String appender to merge individual log entries in value. Average size of log entry is ~1100 bytes. It was nice until I stumbled upon customer who generated more than 2GB in 10 mins timeframe :) So right now I'm planning to write each log entry as individual KV pair and either use partitioned index or manually partition customers into different column families. Perhaps RocksDB is wrong choice completely for append only task but I liked its reliability/caching/compression features and didn't want to reinvent them. Thoughts ?

[adamretter]

Could you please point to Java classes that allow getting ByteBuffer from iterator or use ByteBuffer in put(), get(), merge() calls ? I'm on 6.7.3 and all these methods still operate with byte[].

Apologies, it seems that it did not make it into 6.7.3. It is in master and I see that it will be in the upcoming 6.8.x series: https://github.com/facebook/rocksdb/blob/6.8.fb/java/src/main/java/org/rocksdb/RocksIterator.java#L92

I understood max_successive_merges only as optimization that defines how many merge operations should be in WAL until those merged and actual value is updated. So it's not limiting number of total merges per KV pair. Am I correct ?

I am not clear on the implementation itself, but the documentation seems to imply "total merges per KV pair", i.e. it says "This will ensure that there are never more than max_successive_merges merge operations in the memtable."

Personally - I can't really comment on if RocksDB is the right choice, or how best to structure your data. If you need >2GB values and RocksDB supports that in the C++ layer, we could expose a suitable mechanism in the Java layer. Whether storing single >2GB values into RocksDB is sensible or not is a different discussion.

[StephanEwen]

Thanks, @adamretter, great to see the ByteBuffer API coming up.

From our use cases (Apache Flink) the case where row size exceeds 2GB are certainly not frequent.

They are mostly "accidents", some extreme unexpected data skew makes a key extremely hot (many values that get concatenated (merged)). It would be nice if that only caused slowdown (you can identify the issue in the metrics) and not lead to an unrecoverable error.