anishshri-db commented on code in PR #45778:
URL: https://github.com/apache/spark/pull/45778#discussion_r1547174481
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateEncoder.scala:
##########
@@ -283,45 +286,79 @@ class RangeKeyScanStateEncoder(
// If the value is not null, we write the null byte followed by the value.
// Note that setting null for the index on the unsafeRow is not feasible as
it would change
// the sorting order on iteration.
+ // Also note that the same byte is used to indicate whether the value is
negative or not.
private def encodePrefixKeyForRangeScan(row: UnsafeRow): UnsafeRow = {
val writer = new UnsafeRowWriter(numOrderingCols)
writer.resetRowWriter()
rangeScanKeyFieldsWithIdx.foreach { case (field, idx) =>
val value = row.get(idx, field.dataType)
- val isNullCol: Byte = if (value == null) 0x01.toByte else 0x00.toByte
+ // initialize to 0x01 to indicate that the column is not null and
positive
+ var isNullOrSignCol: Byte = 0x01.toByte
+ // Update the isNullOrSignCol byte to indicate null value
+ if (value == null) {
+ isNullOrSignCol = 0x02.toByte
+ }
// Note that we cannot allocate a smaller buffer here even if the value
is null
// because the effective byte array is considered variable size and
needs to have
// the same size across all rows for the ordering to work as expected.
val bbuf = ByteBuffer.allocate(field.dataType.defaultSize + 1)
bbuf.order(ByteOrder.BIG_ENDIAN)
- bbuf.put(isNullCol)
- if (isNullCol == 0x01.toByte) {
+ if (isNullOrSignCol == 0x02.toByte) {
+ bbuf.put(isNullOrSignCol)
writer.write(idx, bbuf.array())
} else {
field.dataType match {
case BooleanType =>
case ByteType =>
+ bbuf.put(isNullOrSignCol)
bbuf.put(value.asInstanceOf[Byte])
writer.write(idx, bbuf.array())
- // for other multi-byte types, we need to convert to big-endian
case ShortType =>
+ if (value.asInstanceOf[Short] < 0) {
+ isNullOrSignCol = 0x00.toByte
+ }
+ bbuf.put(isNullOrSignCol)
bbuf.putShort(value.asInstanceOf[Short])
writer.write(idx, bbuf.array())
case IntegerType =>
+ if (value.asInstanceOf[Int] < 0) {
+ isNullOrSignCol = 0x00.toByte
+ }
+ bbuf.put(isNullOrSignCol)
bbuf.putInt(value.asInstanceOf[Int])
writer.write(idx, bbuf.array())
case LongType =>
+ if (value.asInstanceOf[Long] < 0) {
+ isNullOrSignCol = 0x00.toByte
+ }
+ bbuf.put(isNullOrSignCol)
bbuf.putLong(value.asInstanceOf[Long])
writer.write(idx, bbuf.array())
+ // For floating point types, we cannot support ordering using
additional byte for
+ // negative values. This is because the IEEE 754 floating point
representation
+ // stores exponent first and then the mantissa. So, we cannot simply
prepend a byte
Review Comment:
Updated this to support negative values too
--
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]
