Rocksdb: Java: Using custom Java Comparator on ColumnFamily and using Put leads to UnsupportedOperationException

[jurmous]

I have implemented a custom comparator with the improved Java Comparator API on 6.8.fb branch (I build RocksDB myself to test changes out in #6252). This leads to an UnsupportedOperationException:

java.lang.UnsupportedOperationException
    at java.nio.ByteBuffer.array(ByteBuffer.java:994)
    at org.rocksdb.RocksDB.put(Native Method)
    at org.rocksdb.RocksDB.put(RocksDB.java:658)
        ...

According to ByteBuffer.array() documentation:
@throws UnsupportedOperationException If this buffer is not backed by an accessible array

I guess this has something to do with the changes in #6252.

Steps to reproduce the behavior

• Set up a column family with custom comparator which is loaded with default ComparatorOptions
• Put values.
• Above crash.

[jurmous]

@adamretter Do you have an insight into what is going on here?

[adamretter]

@jurmous So we are expecting that Direct Byte Buffers must be used.
The issue you are seeing is expected... Java's direct byte buffers never implement array() ;-)

[jurmous]

@adamretter How come that the RocksDB native call then internally uses the array method? Isn't that then the issue? Btw I have this issue somehow randomly, sometimes all tests fail with numerous times this issue, sometimes they all succeed.

[adamretter]

Hmmm...

So the only place that RocksJava calls ByteBuffer#array() is here - https://sourcegraph.com/github.com/facebook/[email protected]/-/blob/java/rocksjni/portal.h#L1368

This method should only be used when constructing a non-direct byte buffer, or when copying data to/from a non-direct byte buffer.
Whilst we have several tests that verify Java comparators with both direct and non-direct byte buffers, assuming your comparator is ok, I guess it is possible that we may be taking a wrong branch under some condition.

To be able to help with your problem, I am going to need to know more about your custom comparator and how it is setup. What is useDirectBuffer set to in your ComparatorOptions? Also do you yourself allocate any ByteBuffers direct or otherwise?

[jurmous]

I have abstracted RocksDB into a Kotlin Multiplatform library so I can use the same code on JVM/iOS and Android. It is thus a Kotlin implementation that on the JVM is a direct typealias (so that means the API footprint is the same as on Java but within a different package) Examples of the AbstractComparator in common code and JVM implementation

So don't be scared of the Kotlin as it directly relates to the Java code. :) As you can see I do not change the default parameters so useDirectBuffer is set to true.

This is the comparator:

private const val versionSize = ULong.SIZE_BYTES

/**
 * Takes care that qualifiers are first sorted on their reference/value and then on version
 * Otherwise the version bytes would make the values come before their root qualifiers.
 */
internal class VersionedComparator(
    private val keySize: Int
) : maryk.rocksdb.AbstractComparator(ComparatorOptions()) {
    override fun name() = "maryk.VersionedComparator"
    override fun compare(a: ByteBuffer, b: ByteBuffer): Int {
        return if (a.remaining() > keySize && b.remaining() > keySize) {
            when (val comparison =
                a.compareToWithOffsetAndLength(0, a.remaining() - versionSize, b, 0, b.remaining() - versionSize)) {
                0 -> a.compareToWithOffsetAndLength(
                    a.remaining() - versionSize,
                    versionSize,
                    b,
                    b.remaining() - versionSize,
                    versionSize
                )
                else -> comparison
            }
        } else {
            a.compareWith(b)
        }
    }
}

These are the two helper functions called:

fun ByteBuffer.compareWith(b: ByteBuffer): Int {
    return this.compareToWithOffsetAndLength(0, this.remaining(), b, 0, b.remaining())
}

fun ByteBuffer.compareToWithOffsetAndLength(aOffset: Int, aLength: Int, b: ByteBuffer, bOffset: Int, bLength: Int): Int {
    for (it in 0 until minOf(aLength, bLength)) {
        val aByte = this[it + aOffset].toUByte() and MAX_BYTE
        val bByte = b[it + bOffset].toUByte() and MAX_BYTE
        if (aByte != bByte) {
            return aByte.toUByte().toInt() - bByte.toUByte().toInt()
        }
    }
    return aLength - bLength
}

I checked the code and I do not set any ByteBuffers, direct or otherwise. Although there is now a ByteBuffer api I still pass ByteArrays directly with a put or other RocksDB calls.

(Any suggestions on how to improve the comparator are also welcome :) )

[adamretter]

@jurmous Can you try converting your comparator into Java, and create a test for it. You can use the comparator tests in https://github.com/facebook/rocksdb/tree/master/java/src/test/java/org/rocksdb/util as examples.

I want to rule out any code outside of RocksDB. If the problem then still occurs with your Java comparator and you have the simple test to reproduce, I should be able to fix any issue ASAP.

[jurmous]

I could reproduce the issue with the following code. It does not always fail but for me certainly once every 5 times.

Exception in thread "Thread-105" java.lang.UnsupportedOperationException
    at java.nio.ByteBuffer.array(ByteBuffer.java:994)
Exception in thread "Thread-106" java.lang.UnsupportedOperationException
Exception in thread "main"  at java.nio.ByteBuffer.array(ByteBuffer.java:994)
java.lang.UnsupportedOperationException
    at java.nio.ByteBuffer.array(ByteBuffer.java:994)
    at org.rocksdb.RocksDB.get(Native Method)
    at org.rocksdb.RocksDB.get(RocksDB.java:1886)
    at Main.main(Main.java:37)
Exception in thread "main" java.lang.UnsupportedOperationException
    at java.nio.ByteBuffer.array(ByteBuffer.java:994)
    at org.rocksdb.RocksDB.get(Native Method)
    at org.rocksdb.RocksDB.get(RocksDB.java:1886)
    at Main.main(Main.java:37)
Exception in thread "Thread-107" java.lang.UnsupportedOperationException
    at java.nio.ByteBuffer.array(ByteBuffer.java:994)

import org.rocksdb.AbstractComparator;
import org.rocksdb.ComparatorOptions;

import java.nio.ByteBuffer;

public class VersionedComparator extends AbstractComparator {
  private final int versionSize = 8;

  protected VersionedComparator(ComparatorOptions copt) {
    super(copt);
  }

  @Override
  public String name() {
    return "test";
  }

  @Override
  public int compare(ByteBuffer a, ByteBuffer b) {
      return compareToWithOffsetAndLength(a, a.remaining() - versionSize, versionSize, b, b.remaining() - versionSize, versionSize);
  }

  private int compareToWithOffsetAndLength(ByteBuffer a, int aOffset, int aLength, ByteBuffer b, int bOffset, int bLength) {
    int minLength = Math.min(aLength, bLength);

    for (int i = 0; i < minLength; i++) {
      int aByte = a.get(i + aOffset) & 0xFF;
      int bByte = b.get(i + bOffset) & 0xFF;
      if (aByte != bByte) {
        return aByte - bByte;
      }
    }
    return aLength - bLength;
  }
}

import org.rocksdb.ColumnFamilyDescriptor;
import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.ComparatorOptions;
import org.rocksdb.DBOptions;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public class Main {
  public static void main(String[] args) {
    ColumnFamilyOptions columnFamilyOptions = new ColumnFamilyOptions();
    columnFamilyOptions.setComparator(
        new VersionedComparator(
            new ComparatorOptions()
        )
    );

    byte[] columnFamilyName = RocksDB.DEFAULT_COLUMN_FAMILY;
    ColumnFamilyDescriptor columnFamilyDescriptor = new ColumnFamilyDescriptor(columnFamilyName, columnFamilyOptions);

    List<ColumnFamilyDescriptor> columnFamilyDescriptors = Collections.singletonList(columnFamilyDescriptor);
    ArrayList<ColumnFamilyHandle> columnFamilyHandles = new ArrayList<>(1);

    DBOptions dbOptions = new DBOptions();
    dbOptions.setCreateIfMissing(true);
    dbOptions.setCreateMissingColumnFamilies(true);

    try(RocksDB rocksdb = RocksDB.open(dbOptions, "test", columnFamilyDescriptors, columnFamilyHandles)) {
      ColumnFamilyHandle cf1 = columnFamilyHandles.get(0);

      rocksdb.put(cf1, ("justanotherrandomkey").getBytes(), "value".getBytes());

      System.out.println(new String(rocksdb.get(cf1, ("justanotherrandomkey").getBytes())));

    } catch (RocksDBException e) {
      e.printStackTrace();
    }
  }
}

[adamretter]

@jurmous Thanks that is great, I will try and take a look in the next few days.

Is veryveryverylongkey" a place-holder? If so can you give me an idea of the length of the key?

[jurmous]

@adamretter It was me messing around with the key. It was first only key but then it would not work with the VersionedComparator its versioned length, so I added those words in front. I changed the key in the example to confuse less which also fails. The keys in the code I use are around 20 to 30 characters.

Thanks for looking into it! 😄

[adamretter]

@jurmous Okay I ran it and it occurred on the 9th iteration locally. I will try and dig into it tomorrow eve...

[jurmous]

@adamretter Did you manage to take a look? 😊

[adamretter]

@jurmous I have taken a look but didn't make much progress, I am afraid I had some personal issues which took me away from work for a few days. I will get back to it ASAP

[adamretter]

@jurmous Okay I found the issue, and it's a doozy!

If you change your Java code from:

ColumnFamilyOptions columnFamilyOptions = new ColumnFamilyOptions();
    columnFamilyOptions.setComparator(
        new VersionedComparator(
            new ComparatorOptions()
        )
    );

to:

ComparatorOptions comparatorOptions = new ComparatorOptions();
ColumnFamilyOptions columnFamilyOptions = new ColumnFamilyOptions();
    columnFamilyOptions.setComparator(
        new VersionedComparator(
            comparatorOptions
        )
    );

I think your problem will disappear. Please let me know if it doesn't...

It's a bit late in the day here (Western Europe) and the issue itself is quite far reaching, and not specific to Comparators. I will write it up and paste a link here soon...

[jurmous]

Thanks for going through the trouble and finding out what is the cause! I adapted it in test and the actual code and both are running reliably now! 😄
It seemed I did not want to adapt the signature of the Comparator when upgrading so I added it in the constructor, and was not thinking to properly put it outside while trying to find the cause.

And thanks for all your work on improving the speed of the Java Comparators! ❤️

[adamretter]

@jurmous You are welcome.

Also to reassure you, there was nothing wrong with your Java code before. It is more to do with how RocksJava manages the lifetime of C++ objects. I'll paste a link here when I have written up the issue.

[mattchabala]

@adamretter Has there been any updates or more information on what was causing this? I'm currently running into a similar issue when using a custom Java Comparator with a TtlDB that uses Options instead of ColumnFamilyOptions.

[hasan169]

@adamretter

I am also facing this exception with this settings.

```
ComparatorOptions comparatorOptions = new ComparatorOptions();
comparatorOptions.useDirectBuffer();
options.setComparator(new CustomComparator(comparatorOptions));

Can you provide a workaround on this?

[jurmous]

@hasan169 Can you try to move new CustomComparator(comparatorOptions) into a variable instead of directly injecting it?

[hasan169]

@jurmous

No luck.