HeartSaVioR commented on code in PR #47107:
URL: https://github.com/apache/spark/pull/47107#discussion_r1669592518
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateEncoder.scala:
##########
@@ -194,43 +207,35 @@ class PrefixKeyScanStateEncoder(
val prefixKeyEncoded = encodeUnsafeRow(extractPrefixKey(row))
val remainingEncoded = encodeUnsafeRow(remainingKeyProjection(row))
- val encodedBytes = new Array[Byte](prefixKeyEncoded.length +
- remainingEncoded.length + 4 + offsetForColFamilyPrefix)
- encodeColumnFamilyPrefix(encodedBytes, useColumnFamilies)
+ val (encodedBytes, startingOffset) = encodeColumnFamilyPrefix(
+ prefixKeyEncoded.length + remainingEncoded.length + 4
+ )
- Platform.putInt(encodedBytes, Platform.BYTE_ARRAY_OFFSET +
offsetForColFamilyPrefix,
- prefixKeyEncoded.length)
+ Platform.putInt(encodedBytes, startingOffset, prefixKeyEncoded.length)
Platform.copyMemory(prefixKeyEncoded, Platform.BYTE_ARRAY_OFFSET,
- encodedBytes, Platform.BYTE_ARRAY_OFFSET + 4 + offsetForColFamilyPrefix,
- prefixKeyEncoded.length)
+ encodedBytes, startingOffset + 4, prefixKeyEncoded.length)
// NOTE: We don't put the length of remainingEncoded as we can calculate
later
// on deserialization.
Platform.copyMemory(remainingEncoded, Platform.BYTE_ARRAY_OFFSET,
- encodedBytes,
- Platform.BYTE_ARRAY_OFFSET + 4 + offsetForColFamilyPrefix +
prefixKeyEncoded.length,
+ encodedBytes, startingOffset + 4 + prefixKeyEncoded.length,
remainingEncoded.length)
encodedBytes
}
override def decodeKey(keyBytes: Array[Byte]): UnsafeRow = {
- val prefixKeyEncodedLen = Platform.getInt(
- keyBytes, Platform.BYTE_ARRAY_OFFSET + offsetForColFamilyPrefix)
+ val prefixKeyEncodedLen = Platform.getInt(keyBytes, decodeKeyStartOffset)
val prefixKeyEncoded = new Array[Byte](prefixKeyEncodedLen)
- Platform.copyMemory(keyBytes,
- Platform.BYTE_ARRAY_OFFSET + 4 + offsetForColFamilyPrefix,
- prefixKeyEncoded,
- Platform.BYTE_ARRAY_OFFSET, prefixKeyEncodedLen)
+ Platform.copyMemory(keyBytes, decodeKeyStartOffset + 4,
+ prefixKeyEncoded, Platform.BYTE_ARRAY_OFFSET, prefixKeyEncodedLen)
// Here we calculate the remainingKeyEncodedLen leveraging the length of
keyBytes
val remainingKeyEncodedLen = keyBytes.length - 4 - prefixKeyEncodedLen -
Review Comment:
Are you assuming the case of `Platform.BYTE_ARRAY_OFFSET != 0` hence the
length of byte array is misaligned if we take Platform.BYTE_ARRAY_OFFSET into
account? Awesome thought if intended :) Great details.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
