Yicong-Huang commented on code in PR #59279:
URL: https://github.com/apache/spark/pull/59279#discussion_r4213034890
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala:
##########
@@ -248,6 +248,9 @@ class JSONOptions(
parameters.get(ENABLE_STREAMING_TOP_LEVEL_ARRAY).map(_.toBoolean).getOrElse(
SQLConf.get.getConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY))
+ val alternateVariantEncoding: Option[AlternateVariantEncoding] = parameters
+
.get(ALTERNATE_VARIANT_ENCODING).flatMap(AlternateVariantEncoding.fromStringOpt)
Review Comment:
**Non-blocking (P2):** `fromStringOpt` is an exact, case-sensitive match,
and its `None` is indistinguishable from "option not set".
`from_json('{"v":"<z85>"}', 'v variant',
map('alternateVariantEncoding','z85'))` therefore takes the default branch and
returns a variant *string* that holds the raw Z85 text, with no error or
warning. That is a silent wrong result that can be persisted. Consider
resolving the value case-insensitively (as `ParseMode.fromString` does with
`Locale.ROOT`) and failing with a clear invalid-option error when the option is
present but unrecognized, keeping `None` only for an absent option.
**Verification:**
- **Behavior:** from_json with a mis-cased Z85 value decodes Z85 payloads.
- **Regression:** from_json with an unsupported encoding value fails with
the invalid-option error instead of producing a string variant.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -885,3 +900,15 @@ class JacksonParser(
}
}
}
+
+object JacksonParser {
+ def getVariantFromZ85(z85: String): VariantVal = {
+ val decoded = Codec.Base85Codec.decodeBytes(z85, z85.length)
+ val metadataSize = VariantUtil.metadataSize(decoded)
+ val valueWithPadding = decoded.slice(metadataSize, decoded.length)
+ val valueSize = VariantUtil.valueSize(valueWithPadding, 0)
+ val value = valueWithPadding.slice(0, valueSize)
+ val metadata = decoded.slice(0, metadataSize)
+ new VariantVal(value, metadata)
Review Comment:
**Non-blocking (P2):** The decoded bytes go straight into `new
VariantVal(value, metadata)` without validation. `metadataSize` checks only the
metadata header and offsets. `valueSize` for OBJECT/ARRAY/short-string returns
a size taken from the header without checking it against the buffer, and
`slice(0, valueSize)` silently truncates when that size is too large. Nothing
checks the version nibble or `VariantUtil.SIZE_LIMIT`. A truncated,
wrong-version or oversized payload is therefore returned as a normal value and
only fails later, in `variant_get`, cast, `to_json` or a Parquet write, outside
parse-mode handling. Consider making this the validating boundary: require
`valueSize` to fit in the remaining bytes, call `VariantUtil.isValidVariant`,
and map size overflow to `variantSizeLimitError` as the JSON path does.
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -885,3 +900,15 @@ class JacksonParser(
}
}
}
+
+object JacksonParser {
+ def getVariantFromZ85(z85: String): VariantVal = {
+ val decoded = Codec.Base85Codec.decodeBytes(z85, z85.length)
+ val metadataSize = VariantUtil.metadataSize(decoded)
+ val valueWithPadding = decoded.slice(metadataSize, decoded.length)
Review Comment:
**Nit (P3):** `valueWithPadding` is a full copy of the value region, and it
is immediately sliced again. `VariantUtil.valueSize` takes a start position, so
`val valueSize = VariantUtil.valueSize(decoded, metadataSize)` followed by
`decoded.slice(metadataSize, metadataSize + valueSize)` yields the same bytes
with one copy. Similarly, `decodeBlocks` could read `encoded.charAt(i)` instead
of copying the input with `toCharArray`.
**Verification:**
- **Regression:** Z85 round-trip tests produce identical metadata and value
bytes after the change.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -885,3 +900,15 @@ class JacksonParser(
}
}
}
+
+object JacksonParser {
+ def getVariantFromZ85(z85: String): VariantVal = {
+ val decoded = Codec.Base85Codec.decodeBytes(z85, z85.length)
Review Comment:
**Nit (P3):** Malformed Z85 strings fail here with
`IllegalArgumentException` from the codec's `require`: "Input should be 5
character aligned." or `s"Input is not valid Z85: $encoded"`, which embeds the
whole payload in the message. Every other malformed case on this path raises
`MALFORMED_VARIANT`. Consider catching the codec failure here and rethrowing
`QueryExecutionErrors.malformedVariant()`, so one ingest path has one error
class and no unbounded user data in the message.
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -142,13 +145,25 @@ class JacksonParser(
if (parser.getCurrentToken == FIELD_NAME) {
parser.nextToken()
}
- try {
- val v = VariantBuilder.parseJson(
- parser, variantAllowDuplicateKeys, variantValidateUnicodeInJsonParsing)
- new VariantVal(v.getValue, v.getMetadata)
- } catch {
- case _: VariantSizeLimitException =>
- throw
QueryExecutionErrors.variantSizeLimitError(VariantUtil.SIZE_LIMIT,
"JacksonParser")
+ alternateVariantEncoding match {
+ case Some(VariantZ85Encoding) =>
+ // The token should always be a single Z85 string.
+ val token: JsonToken = parser.currentToken()
+ token match {
+ case VALUE_STRING =>
JacksonParser.getVariantFromZ85(parser.getText())
+ case _ => throw QueryExecutionErrors.malformedVariant()
Review Comment:
**Non-blocking (P2):** This arm also catches `VALUE_NULL`. The `VariantType`
converter is invoked directly on null tokens for struct fields, array elements
and map values (and the comment at the top of `parseVariant` says that is
intentional, so a JSON null becomes a variant null). With the option on,
`from_json('{"v":null,"i":1}', 'v variant, i int',
map('alternateVariantEncoding','Z85'))` throws `MALFORMED_VARIANT`. At the root
that becomes a bad record, so PERMISSIVE loses `i` and FAILFAST fails. The new
tests encode null only as a Z85 payload, never as a JSON `null` token. Handling
`VALUE_NULL` the same way as the default branch before rejecting other
non-string tokens would fix this.
**Verification:**
- **Regression:** A JSON null variant field under the Z85 option yields the
same value as without the option, and the rest of the record is preserved.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -142,13 +145,25 @@ class JacksonParser(
if (parser.getCurrentToken == FIELD_NAME) {
parser.nextToken()
}
- try {
- val v = VariantBuilder.parseJson(
- parser, variantAllowDuplicateKeys, variantValidateUnicodeInJsonParsing)
- new VariantVal(v.getValue, v.getMetadata)
- } catch {
- case _: VariantSizeLimitException =>
- throw
QueryExecutionErrors.variantSizeLimitError(VariantUtil.SIZE_LIMIT,
"JacksonParser")
+ alternateVariantEncoding match {
+ case Some(VariantZ85Encoding) =>
+ // The token should always be a single Z85 string.
+ val token: JsonToken = parser.currentToken()
+ token match {
+ case VALUE_STRING =>
JacksonParser.getVariantFromZ85(parser.getText())
+ case _ => throw QueryExecutionErrors.malformedVariant()
Review Comment:
**Nit (P3):** The new suites exercise only valid Z85 round trips. Removing
this `malformedVariant()` arm, reordering the `metadataSize` overflow checks,
or changing how codec errors route through parse modes would not fail any added
test. Please add from_json cases for a non-string token, a JSON `null`,
unaligned and invalid-character strings, a truncated value, and an out-of-range
group, with PERMISSIVE and FAILFAST expectations. Also add direct tests for the
`VariantUtil.metadataSize` guard branches (empty input, overflowing offset
position, size larger than the buffer).
**Verification:**
- **Regression:** Each malformed Z85 input class yields its expected error
or bad-record outcome under PERMISSIVE and FAILFAST.
- **Behavior:** The metadataSize overflow and bounds guards each throw
MALFORMED_VARIANT.
##########
common/utils/src/main/scala/org/apache/spark/util/Codec.scala:
##########
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util
+
+import java.nio.ByteBuffer
+import java.nio.charset.StandardCharsets.US_ASCII
+import java.util.UUID
+
+/** Additional codecs not supported by Apache Commons Codecs. */
+object Codec {
+
+ def uuidToBytes(id: UUID): Array[Byte] = uuidToByteBuffer(id).array()
+
+ def uuidFromBytes(bytes: Array[Byte]): UUID = {
+ require(bytes.length == 16)
+ uuidFromByteBuffer(ByteBuffer.wrap(bytes))
+ }
+
+ def uuidToByteBuffer(id: UUID): ByteBuffer = {
+ val buffer = ByteBuffer.allocate(16)
+ buffer.putLong(id.getMostSignificantBits)
+ buffer.putLong(id.getLeastSignificantBits)
+ buffer.rewind()
+ buffer
+ }
+
+ def uuidFromByteBuffer(buffer: ByteBuffer): UUID = {
+ require(buffer.remaining() >= 16)
+ val highBits = buffer.getLong
+ val lowBits = buffer.getLong
+ new UUID(highBits, lowBits)
+ }
+
+ /**
+ * This implements Base85 using the 4 byte block aligned encoding and
character set from Z85.
+ *
+ * @see https://rfc.zeromq.org/spec/32/
+ */
+ object Base85Codec {
+
+ final val ENCODE_MAP: Array[Byte] = {
+ val chars = ('0' to '9') ++ ('a' to 'z') ++ ('A' to 'Z') ++
".-:+=^!/*?&<>()[]{}@%$#"
+ chars.map(_.toByte).toArray
+ }
+
+ lazy val DECODE_MAP: Array[Byte] = {
+ require(ENCODE_MAP.length - 1 <= Byte.MaxValue)
+ // The bitmask is the same as largest possible value, so the length of
the array must
+ // be one greater.
+ val map: Array[Byte] = Array.fill(ASCII_BITMASK + 1)(-1)
+ for ((b, i) <- ENCODE_MAP.zipWithIndex) {
+ map(b) = i.toByte
+ }
+ map
+ }
+
+ final val BASE: Long = 85L
+ final val BASE_2ND_POWER: Long = 7225L // 85^2
+ final val BASE_3RD_POWER: Long = 614125L // 85^3
+ final val BASE_4TH_POWER: Long = 52200625L // 85^4
+ final val ASCII_BITMASK: Int = 0x7F
+
+ // UUIDs always encode into 20 characters.
+ final val ENCODED_UUID_LENGTH: Int = 20
+
+ /** Encode a 16 byte UUID. */
+ def encodeUUID(id: UUID): String = {
+ val buffer = uuidToByteBuffer(id)
+ encodeBlocks(buffer)
+ }
+
+ /**
+ * Decode a 16 byte UUID. */
+ def decodeUUID(encoded: String): UUID = {
+ val buffer = decodeBlocks(encoded)
+ uuidFromByteBuffer(buffer)
+ }
+
+ /**
+ * Encode an arbitrary byte array.
+ *
+ * Unaligned input will be padded to a multiple of 4 bytes.
+ */
+ def encodeBytes(input: Array[Byte]): String = {
+ if (input.length % 4 == 0) {
+ encodeBlocks(ByteBuffer.wrap(input))
+ } else {
+ val alignedLength = ((input.length + 4) / 4) * 4
+ val buffer = ByteBuffer.allocate(alignedLength)
+ buffer.put(input)
+ while (buffer.hasRemaining) {
+ buffer.put(0.asInstanceOf[Byte])
+ }
+ buffer.rewind()
+ encodeBlocks(buffer)
+ }
+ }
+
+ /**
+ * Encode an arbitrary byte array using 4 byte blocks.
+ *
+ * Expects the input to be 4 byte aligned.
+ */
+ private def encodeBlocks(buffer: ByteBuffer): String = {
+ require(buffer.remaining() % 4 == 0)
+ val numBlocks = buffer.remaining() / 4
+ // Every 4 byte block gets encoded into 5 bytes/chars
+ val outputLength = numBlocks * 5
+ val output: Array[Byte] = Array.ofDim(outputLength)
+ var outputIndex = 0
+
+ while (buffer.hasRemaining) {
+ var sum: Long = buffer.getInt & 0x00000000ffffffffL
+ output(outputIndex) = ENCODE_MAP((sum / BASE_4TH_POWER).toInt)
+ sum %= BASE_4TH_POWER
+ output(outputIndex + 1) = ENCODE_MAP((sum / BASE_3RD_POWER).toInt)
+ sum %= BASE_3RD_POWER
+ output(outputIndex + 2) = ENCODE_MAP((sum / BASE_2ND_POWER).toInt)
+ sum %= BASE_2ND_POWER
+ output(outputIndex + 3) = ENCODE_MAP((sum / BASE).toInt)
+ output(outputIndex + 4) = ENCODE_MAP((sum % BASE).toInt)
+ outputIndex += 5
+ }
+
+ new String(output, US_ASCII)
+ }
+
+ /**
+ * Decode an arbitrary byte array.
+ *
+ * Only `outputLength` bytes will be returned.
+ * Any extra bytes, such as padding added because the input was unaligned,
will be dropped.
+ */
+ def decodeBytes(encoded: String, outputLength: Int): Array[Byte] = {
+ val result = decodeBlocks(encoded)
+ if (result.remaining() > outputLength) {
+ // Only read the expected number of bytes.
+ val output: Array[Byte] = Array.ofDim(outputLength)
+ result.get(output)
+ output
+ } else {
+ result.array()
+ }
+ }
+
+ /**
+ * Decode an arbitrary byte array.
+ *
+ * Output may contain padding bytes, if the input was not 4 byte aligned.
+ * Use [[decodeBytes]] in that case and specify the expected number of
output bytes
+ * without padding.
+ */
+ def decodeAlignedBytes(encoded: String): Array[Byte] =
decodeBlocks(encoded).array()
+
+ /**
+ * Decode an arbitrary byte array.
+ *
+ * Output may contain padding bytes, if the input was not 4 byte aligned.
+ */
+ private def decodeBlocks(encoded: String): ByteBuffer = {
+ val input = encoded.toCharArray
+ require(input.length % 5 == 0, "Input should be 5 character aligned.")
+ val buffer = ByteBuffer.allocate(input.length / 5 * 4)
+
+ // A mechanism to detect invalid characters in the input while decoding,
that only has a
+ // single conditional at the very end, instead of branching for every
character.
+ var canary: Int = 0
+ def decodeInputChar(i: Int): Long = {
+ val c = input(i)
+ canary |= c // non-ascii char has bits outside of ASCII_BITMASK
+ val b = DECODE_MAP(c & ASCII_BITMASK)
+ canary |= b // invalid char maps to -1, which has bits outside
ASCII_BITMASK
+ b.toLong
+ }
+
+ var inputIndex = 0
+ while (buffer.hasRemaining) {
+ var sum = 0L
+ sum += decodeInputChar(inputIndex) * BASE_4TH_POWER
+ sum += decodeInputChar(inputIndex + 1) * BASE_3RD_POWER
+ sum += decodeInputChar(inputIndex + 2) * BASE_2ND_POWER
+ sum += decodeInputChar(inputIndex + 3) * BASE
+ sum += decodeInputChar(inputIndex + 4)
+ buffer.putInt(sum.toInt)
Review Comment:
**Nit (P3):** Nothing rejects a group whose value exceeds `0xFFFFFFFFL`.
Five Z85 digits can reach 85^5-1 = 4437053124. For example, `"#####"` is all
index-84 digits, and `sum.toInt` silently wraps it to a different block, while
the canary only catches out-of-alphabet characters. Consider failing when `sum
> 0xFFFFFFFFL` on the same path as the other invalid-input checks, and adding a
CodecSuite case for it.
**Verification:**
- **Regression:** Out-of-range five-character groups are rejected by the
decoder.
- **Compatibility:** Existing Z85 reference vectors and random round trips
still pass.
##########
core/src/test/scala/org/apache/spark/util/CodecSuite.scala:
##########
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util
+
+import java.nio.charset.StandardCharsets.US_ASCII
+import java.util.UUID
+
+import scala.util.Random
+
+import org.apache.spark.SparkFunSuite
+
+class CodecSuite extends SparkFunSuite with JsonUtils {
+
+ import CodecSuite._
+
+ // Z85 reference strings are generated by
https://cryptii.com/pipes/z85-encoder
+ val testUuids = Seq[(UUID, String)](
+ new UUID(0L, 0L) -> "00000000000000000000",
+ new UUID(Long.MinValue, Long.MinValue) -> "Fb/MH00000Fb/MH00000",
+ new UUID(-1L, -1L) -> "%nSc0%nSc0%nSc0%nSc0",
+ new UUID(0L, Long.MinValue) -> "0000000000Fb/MH00000",
+ new UUID(0L, -1L) -> "0000000000%nSc0%nSc0",
+ new UUID(0L, Long.MaxValue) -> "0000000000Fb/MG%nSc0",
+ new UUID(Long.MinValue, 0L) -> "Fb/MH000000000000000",
+ new UUID(-1L, 0L) -> "%nSc0%nSc00000000000",
+ new UUID(Long.MaxValue, 0L) -> "Fb/MG%nSc00000000000",
+ new UUID(0L, 1L) -> "00000000000000000001",
+ // Just a few random ones, using literals for test determinism
+ new UUID(-4124158004264678669L, -6032951921472435211L) ->
"-(5oirYA.yTvx6v@H:L>",
+ new UUID(6453181356142382984L, 8208554093199893996L) ->
"s=Mlx-0Pp@AQ6uw@k6=D",
+ new UUID(6453181356142382984L, -8208554093199893996L) ->
"s=Mlx-0Pp@JUL=R13LuL",
+ new UUID(-4124158004264678669L, 8208554093199893996L) ->
"-(5oirYA.yAQ6uw@k6=D")
+
+ // From https://rfc.zeromq.org/spec/32/ - Test Case
+ test("Z85 spec reference value") {
+ val inputBytes: Array[Byte] =
+ Array(0x86, 0x4F, 0xD2, 0x6F, 0xB5, 0x59, 0xF7, 0x5B).map(_.toByte)
+ val expectedEncodedString = "HelloWorld"
+ val actualEncodedString = Codec.Base85Codec.encodeBytes(inputBytes)
+ assert(actualEncodedString === expectedEncodedString)
+ val outputBytes = Codec.Base85Codec.decodeAlignedBytes(actualEncodedString)
+ assert(outputBytes sameElements inputBytes)
+ }
+
+ test("Z85 reference implementation values") {
+ for ((id, expectedEncodedString) <- testUuids) {
+ val actualEncodedString = Codec.Base85Codec.encodeUUID(id)
+ assert(actualEncodedString === expectedEncodedString)
+ }
+ }
+
+ test("Z85 spec character map") {
+ assert(Codec.Base85Codec.ENCODE_MAP.length === 85)
+ val referenceBytes = Seq(
+ 0x00, 0x09, 0x98, 0x62, 0x0f, 0xc7, 0x99, 0x43, 0x1f, 0x85,
+ 0x9a, 0x24, 0x2f, 0x43, 0x9b, 0x05, 0x3f, 0x01, 0x9b, 0xe6,
+ 0x4e, 0xbf, 0x9c, 0xc7, 0x5e, 0x7d, 0x9d, 0xa8, 0x6e, 0x3b,
+ 0x9e, 0x89, 0x7d, 0xf9, 0x9f, 0x6a, 0x8d, 0xb7, 0xa0, 0x4b,
+ 0x9d, 0x75, 0xa1, 0x2c, 0xad, 0x33, 0xa2, 0x0d, 0xbc, 0xf1,
+ 0xa2, 0xee, 0xcc, 0xaf, 0xa3, 0xcf, 0xdc, 0x6d, 0xa4, 0xb0,
+ 0xec, 0x2b, 0xa5, 0x91, 0xfb, 0xe9, 0xa6, 0x72)
+ .map(_.toByte).toArray
+ val referenceString = new String(Codec.Base85Codec.ENCODE_MAP, US_ASCII)
+ val encodedString = Codec.Base85Codec.encodeBytes(referenceBytes)
+ assert(encodedString === referenceString)
+ val decodedBytes = Codec.Base85Codec.decodeAlignedBytes(encodedString)
+ assert(decodedBytes sameElements referenceBytes)
+ }
+
+ test("Reject illegal Z85 input - unaligned string") {
+ // Minimum string should 5 characters
Review Comment:
**Nit (P3):** This should read "Minimum string should be 5 characters".
Also, the `forAllEncodedStrings` scaladoc ("Execute `thunk` works for strings
containing ...") would read better as "Check that `thunk` works for strings
containing ...".
**Verification:**
- **Inspection:** The two comments are grammatical, and no code changes.
--
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]
