Skip to content

CASSANALYTICS-187: Fix FastByteOperations Silently Falling Back to Pure-Java Comparator #228

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yifan-c merged 2 commits into from
Aug 3, 2026
Merged

CASSANALYTICS-187: Fix FastByteOperations Silently Falling Back to Pure-Java Comparator #228

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a26025e
CASSANALYTICS-187: FastByteOperations fails to load Unsafe
yifan-c Aug 3, 2026
ed7b33c
update changes.txt
yifan-c Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGES.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
0.5.0
-----
* Fix FastByteOperations Silently Falling Back to Pure-Java Comparator (CASSANALYTICS-187)
* AbstractSchemaBuilder.build() drops the cdc flag, silently disabling CDC on all tables (CASSANALYTICS-186)
* Fix total timeout calculation for getAllNodeSettings (CASSANALYTICS-179)
* Support vector data type (CASSANALYTICS-26)
Expand Down
137 changes: 76 additions & 61 deletions ...ics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/util/FastByteOperations.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@

import com.google.common.primitives.Longs;
import com.google.common.primitives.UnsignedBytes;
import com.google.common.primitives.UnsignedLongs;

import sun.misc.Unsafe; // CHECKSTYLE IGNORE: Used with care to speed byte operations up

Expand Down Expand Up @@ -105,43 +104,55 @@ static ByteOperations getBest()

}

@SuppressWarnings("unused") // used via reflection
public static final class UnsafeOperations implements ByteOperations
{
static final Unsafe UNSAFE;
static final Unsafe theUnsafe;
/**
* The offset to the first element in a byte array
* The offset to the first element in a byte array.
*/
static final long BYTE_ARRAY_BASE_OFFSET;
static final long DIRECT_BUFFER_ADDRESS_OFFSET;

static
{
UNSAFE = (Unsafe) AccessController.doPrivileged((PrivilegedAction<Object>) () -> {
try
{
Field field = Unsafe.class.getDeclaredField("UNSAFE");
field.setAccessible(true);
return field.get(null);
}
catch (NoSuchFieldException | IllegalAccessException exception)
{
// It doesn't matter what we throw; it's swallowed in getBest()
throw new Error();
}
});
theUnsafe = (Unsafe) AccessController.doPrivileged(
new PrivilegedAction<Object>()
{
@Override
public Object run()
{
try
{
Field f = Unsafe.class.getDeclaredField("theUnsafe");
f.setAccessible(true);
return f.get(null);
}
catch (NoSuchFieldException e)
{
// It doesn't matter what we throw;
// it's swallowed in getBest().
throw new Error();
}
catch (IllegalAccessException e)
{
throw new Error();
}
}
});

try
{
BYTE_ARRAY_BASE_OFFSET = UNSAFE.arrayBaseOffset(byte[].class);
DIRECT_BUFFER_ADDRESS_OFFSET = UNSAFE.objectFieldOffset(Buffer.class.getDeclaredField("address"));
BYTE_ARRAY_BASE_OFFSET = theUnsafe.arrayBaseOffset(byte[].class);
DIRECT_BUFFER_ADDRESS_OFFSET = theUnsafe.objectFieldOffset(Buffer.class.getDeclaredField("address"));
}
catch (Exception exception)
catch (Exception e)
{
throw new AssertionError(exception);
throw new AssertionError(e);
}

// Sanity check - this should never fail
if (UNSAFE.arrayIndexScale(byte[].class) != 1)
// sanity check - this should never fail
if (theUnsafe.arrayIndexScale(byte[].class) != 1)
{
throw new AssertionError();
}
Expand All @@ -154,63 +165,62 @@ public int compare(ByteBuffer buffer1, ByteBuffer buffer2)
return compareTo(buffer1, buffer2);
}

static int compareTo(ByteBuffer buffer1, ByteBuffer buffer2)
public static int compareTo(ByteBuffer buffer1, ByteBuffer buffer2)
{
Object object1;
Object obj1;
long offset1;
int length1;
if (buffer1.hasArray())
{
object1 = buffer1.array();
obj1 = buffer1.array();
offset1 = BYTE_ARRAY_BASE_OFFSET + buffer1.arrayOffset();
}
else
{
object1 = null;
offset1 = UNSAFE.getLong(buffer1, DIRECT_BUFFER_ADDRESS_OFFSET);
obj1 = null;
offset1 = theUnsafe.getLong(buffer1, DIRECT_BUFFER_ADDRESS_OFFSET);
}
offset1 += buffer1.position();
length1 = buffer1.remaining();
return compareTo(object1, offset1, length1, buffer2);
return compareTo(obj1, offset1, length1, buffer2);
}

static int compareTo(Object buffer1, long offset1, int length1, ByteBuffer buffer2)
public static int compareTo(Object buffer1, long offset1, int length1, ByteBuffer buffer)
{
Object object2;
Object obj2;
long offset2;

int position = buffer2.position();
int limit = buffer2.limit();
if (buffer2.hasArray())
int position = buffer.position();
int limit = buffer.limit();
if (buffer.hasArray())
{
object2 = buffer2.array();
offset2 = BYTE_ARRAY_BASE_OFFSET + buffer2.arrayOffset();
obj2 = buffer.array();
offset2 = BYTE_ARRAY_BASE_OFFSET + buffer.arrayOffset();
}
else
{
object2 = null;
offset2 = UNSAFE.getLong(buffer2, DIRECT_BUFFER_ADDRESS_OFFSET);
obj2 = null;
offset2 = theUnsafe.getLong(buffer, DIRECT_BUFFER_ADDRESS_OFFSET);
}
int length2 = limit - position;
offset2 += position;

return compareTo(buffer1, offset1, length1, object2, offset2, length2);
return compareTo(buffer1, offset1, length1, obj2, offset2, length2);
}

/**
* Lexicographically compare two arrays
* Lexicographically compare two arrays.
*
* @param buffer1 left operand: a byte[] or null
* @param buffer2 right operand: a byte[] or null
* @param offset1 Where to start comparing in the left buffer
* (pure memory address if buffer1 is null, or relative otherwise)
* @param offset2 Where to start comparing in the right buffer
* (pure memory address if buffer1 is null, or relative otherwise)
* @param memoryOffset1 Where to start comparing in the left buffer (pure memory address if buffer1 is null, or relative otherwise)
* @param memoryOffset2 Where to start comparing in the right buffer (pure memory address if buffer1 is null, or relative otherwise)
* @param length1 How much to compare from the left buffer
* @param length2 How much to compare from the right buffer
* @return 0 if equal, < 0 if left is less than right, etc.
* @return 0 if equal, {@code < 0} if left is less than right, etc.
*/
static int compareTo(Object buffer1, long offset1, int length1, Object buffer2, long offset2, int length2)
public static int compareTo(Object buffer1, long memoryOffset1, int length1,
Object buffer2, long memoryOffset2, int length2)
{
int minLength = Math.min(length1, length2);

Expand All @@ -220,45 +230,50 @@ static int compareTo(Object buffer1, long offset1, int length1, Object buffer2,
* On the other hand, it is substantially faster on 64-bit.
*/
int wordComparisons = minLength & ~7;
for (int index = 0; index < wordComparisons; index += Longs.BYTES)
for (int i = 0; i < wordComparisons; i += Longs.BYTES)
{
long long1 = UNSAFE.getLong(buffer1, offset1 + index);
long long2 = UNSAFE.getLong(buffer2, offset2 + index);
long lw = theUnsafe.getLong(buffer1, memoryOffset1 + i);
long rw = theUnsafe.getLong(buffer2, memoryOffset2 + i);

if (long1 != long2)
if (lw != rw)
{
return BIG_ENDIAN ? UnsignedLongs.compare(long1, long2)
: UnsignedLongs.compare(Long.reverseBytes(long1), Long.reverseBytes(long2));
if (BIG_ENDIAN)
{
return Long.compareUnsigned(lw, rw);
}

return Long.compareUnsigned(Long.reverseBytes(lw), Long.reverseBytes(rw));
}
}

for (int index = wordComparisons; index < minLength; index++)
for (int i = wordComparisons; i < minLength; i++)
{
int byte1 = UNSAFE.getByte(buffer1, offset1 + index) & 0xFF;
int byte2 = UNSAFE.getByte(buffer2, offset2 + index) & 0xFF;
if (byte1 != byte2)
int b1 = theUnsafe.getByte(buffer1, memoryOffset1 + i) & 0xFF;
int b2 = theUnsafe.getByte(buffer2, memoryOffset2 + i) & 0xFF;
if (b1 != b2)
{
return byte1 - byte2;
return b1 - b2;
}
}

return length1 - length2;
}
}

@SuppressWarnings("unused")
public static final class PureJavaOperations implements ByteOperations
{
public int compare(ByteBuffer buffer1, ByteBuffer buffer2)
{
int end1 = buffer1.limit();
int end2 = buffer2.limit();
for (int index1 = buffer1.position(), index2 = buffer2.position(); index1 < end1 && index2 < end2; index1++, index2++)
for (int i = buffer1.position(), j = buffer2.position(); i < end1 && j < end2; i++, j++)
{
int byte1 = buffer1.get(index1) & 0xFF;
int byte2 = buffer2.get(index2) & 0xFF;
if (byte1 != byte2)
int a = buffer1.get(i) & 0xff;
int b = buffer2.get(j) & 0xff;
if (a != b)
{
return byte1 - byte2;
return a - b;
}
}
return buffer1.remaining() - buffer2.remaining();
Expand Down
Loading