Repository navigation
[SPARK-60058][SQL] Add ALTERNATE_VARIANT_ENCODING JSON Option for from_json and to_json expressions - #59279
[SPARK-60058][SQL] Add ALTERNATE_VARIANT_ENCODING JSON Option for from_json and to_json expressions#59279harshmotw-db wants to merge 1 commit into
Conversation
There was a problem hiding this comment.
Adds support for the Base85Codec which we use for Z85 encoding and decoding.
Yicong-Huang
left a comment
Yicong-Huang
left a comment
There was a problem hiding this comment.
Review summary
The read side of the new option works for well-formed input: valid Z85 payloads round-trip through from_json at the top level and inside structs, arrays and maps. The main gap is on the write side. Despite the title and description, to_json never reads alternateVariantEncoding, and a new test asserts plain-JSON output with the option set, so the advertised from_json(to_json(x)) Z85 round trip does not exist.
On the read path, three behaviors can silently produce wrong data or lose rows. A mis-cased or unknown option value silently falls back to JSON parsing, which turns the Z85 text into a string variant. A JSON null for a variant field becomes a malformed record, which drops the whole row in PERMISSIVE mode. Decoded bytes are wrapped into a VariantVal without version, structure or size validation, so corrupt payloads surface later, outside parse-mode handling.
Smaller issues: codec failures surface as IllegalArgumentException with the full payload in the message instead of MALFORMED_VARIANT; the Base85 decoder silently wraps out-of-range groups; the decode path makes avoidable copies; none of the malformed-input branches has a test; and two test comments need rewording.
Two scope questions are raised separately: whether the option is meant to apply to the JSON file source (including singleVariantColumn), and whether org.apache.spark.util.Codec is intended as public API.
Cross-cutting observations
- The new Z85 read path has no defined malformed-input contract. Null tokens corrupt whole records, invalid bytes are either accepted unvalidated or wrapped silently, decode errors use a non-error-class exception, and no test covers any of these. Treating the Z85 ingest as one well-specified boundary (what is null, what is malformed, and which error is raised) would make the individual fixes consistent.
Separation: Each issue has its own owner and mechanism (token handling, ingest validation, codec range check, tests), so the pattern does not change repair grouping or require coordinated ownership.
Findings
9 total: 0 P0, 0 P1, 4 P2, 5 P3.
Non-blocking (P2)
-
to_json ignores alternateVariantEncoding=Z85, contrary to the PR's stated behavior —
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala:251
The description saysto_jsonencodes variant data as Z85 whenalternateVariantEncodingisZ85, but the option has a single consumer,JacksonParser.parseVariant.JacksonGenerator.makeWriterstill mapsVariantTypetowrite(row.getVariant(ordinal)), which emits JSON, and never reads the option. The new test inJsonExpressionsSuitepins this by assertingStructsToJson(options, ...) == variant.toJson(UTC). Soto_json(named_struct('v', parse_json('1')), map('alternateVariantEncoding','Z85'))returns{"v":1}, and reading that back withfrom_json(..., 'v variant', <same options>)hits the non-string branch and fails withMALFORMED_VARIANT. Either implement the writer side, or narrow the title and description tofrom_jsonand drop the to_json claim.Recommended change: When alternateVariantEncoding is Z85, make JacksonGenerator write each variant as a JSON string holding Base85Codec.encodeBytes(metadata ++ value). Update the from_json/to_json expression test to expect the Z85 string and to assert the from_json(to_json(x)) round trip. Add JacksonGenerator tests covering the option.
Why this works: Read options.alternateVariantEncoding once in JacksonGenerator. In the VariantType writer, branch on it so that Z85 emits gen.writeString(z85) of the concatenated metadata and value bytes, which is the format getVariantFromZ85 decodes.
Scope: Implement Z85 output for variants in the JSON generator and fix the test that pins plain-JSON output.
Compatibility: Default to_json variant rendering as JSON.
Risks: Writer-side support also changes JSON file output when the option is set. That should be consistent with the read-side scope decision.
Constraints: The default (option unset) generator output must remain unchanged.
Success: to_json with alternateVariantEncoding=Z85 emits each variant as a Z85 string from_json(to_json(x, opts), opts) returns the original variant to_json without the option is unchanged
Verification:
- Behavior: to_json with the Z85 option emits Z85 strings for top-level and nested variants that decode to the original metadata and value bytes.
- Regression: Round trip from_json(to_json(v, Z85), Z85) equals v for scalar, string, null, array and object variants.
- Compatibility: to_json output for variants without the option is byte-identical to before.
-
Unrecognized or mis-cased alternateVariantEncoding values are silently ignored —
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala:252— see inline. -
Z85 mode turns a JSON null for a variant field into a malformed record —
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:154— see inline. -
Decoded Z85 variants are not validated before becoming VariantVal —
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:912— see inline.
Nit (P3)
- Codec decode failures surface as IllegalArgumentException that embeds the payload —
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:906— see inline. - getVariantFromZ85 and decodeBlocks make avoidable per-row copies —
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:908— see inline. - Base85 decoder silently wraps out-of-range 5-character groups —
common/utils/src/main/scala/org/apache/spark/util/Codec.scala:199— see inline. - No negative-path tests for the new Z85 parsing behavior —
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:154— see inline. - Two new CodecSuite comments are ungrammatical —
core/src/test/scala/org/apache/spark/util/CodecSuite.scala:86— see inline.
Shared repair plans
Shared repair plan 1
Covered findings:
- Decoded Z85 variants are not validated before becoming VariantVal —
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:912 - Codec decode failures surface as IllegalArgumentException that embeds the payload —
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:906
Verification:
- Regression: Malformed Z85 payloads (unaligned, invalid character, truncated value, bad version, over the size limit) produce the expected error class and parse-mode behavior.
- Behavior: Valid payloads still round-trip unchanged.
Verification
- options.alternateVariantEncoding has no consumer outside JacksonParser. JacksonGenerator's VariantType writer emits JSON and does not read the option.
- The variant converter is invoked directly on VALUE_NULL tokens, and the Z85 branch rejects every non-VALUE_STRING token with MALFORMED_VARIANT.
PR description suggestions
- If the to_json write side is not implemented in this PR, update the title ("for from_json and to_json expressions") and the first paragraph of the description, which currently say to_json encodes variant data as Z85 when the option is set.
Decision challenges
Should org.apache.spark.util.Codec be public API, or private[spark]?
Verified: object Codec is added to org.apache.spark.util in common/utils without private[spark]. It has a nested Base85Codec, UUID helpers, decodeAlignedBytes, and publicly reachable mutable arrays ENCODE_MAP/DECODE_MAP. Most neighbouring objects there are private[spark] (for example SparkStringUtils, SparkErrorUtils, JsonUtils, MavenUtils), though a few are public. The only production caller in this PR is JacksonParser.getVariantFromZ85 -> decodeBytes; the UUID helpers, ENCODED_UUID_LENGTH and decodeAlignedBytes have no production caller. What I could not determine: whether you intend this as a supported public utility, or as an internal helper for follow-up work. If it is internal, consider private[spark] and trimming the unused helpers, so Spark does not take on a compatibility commitment and external code cannot mutate the codec tables.
Is alternateVariantEncoding meant to apply to spark.read.json, including singleVariantColumn?
Verified: the option is registered via newOption, so JSONOptions.isValidOption accepts it for the JSON file source. The Z85 switch lives in the shared parseVariant, which is also the root converter for singleVariantColumn/explodeEmbeddedArray and for every nested variant field. As a result, spark.read.option('singleVariantColumn','v').option('alternateVariantEncoding','Z85').json(path) over ordinary JSON lines turns every row into a corrupt record, because the root token is START_OBJECT. What I could not verify: whether file-source support, and these whole-record modes, are intended, or only the from_json/to_json expressions as the description says. Could you confirm the intended scope? Depending on the answer, the unsupported combinations should be rejected up front, or documented and tested.
| SQLConf.get.getConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY)) | ||
|
|
||
| val alternateVariantEncoding: Option[AlternateVariantEncoding] = parameters | ||
| .get(ALTERNATE_VARIANT_ENCODING).flatMap(AlternateVariantEncoding.fromStringOpt) |
There was a problem hiding this 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.
| val token: JsonToken = parser.currentToken() | ||
| token match { | ||
| case VALUE_STRING => JacksonParser.getVariantFromZ85(parser.getText()) | ||
| case _ => throw QueryExecutionErrors.malformedVariant() |
There was a problem hiding this 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.
| val valueSize = VariantUtil.valueSize(valueWithPadding, 0) | ||
| val value = valueWithPadding.slice(0, valueSize) | ||
| val metadata = decoded.slice(0, metadataSize) | ||
| new VariantVal(value, metadata) |
There was a problem hiding this 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.
|
|
||
| object JacksonParser { | ||
| def getVariantFromZ85(z85: String): VariantVal = { | ||
| val decoded = Codec.Base85Codec.decodeBytes(z85, z85.length) |
There was a problem hiding this 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.
| 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) |
There was a problem hiding this 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.
| sum += decodeInputChar(inputIndex + 2) * BASE_2ND_POWER | ||
| sum += decodeInputChar(inputIndex + 3) * BASE | ||
| sum += decodeInputChar(inputIndex + 4) | ||
| buffer.putInt(sum.toInt) |
There was a problem hiding this 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.
| val token: JsonToken = parser.currentToken() | ||
| token match { | ||
| case VALUE_STRING => JacksonParser.getVariantFromZ85(parser.getText()) | ||
| case _ => throw QueryExecutionErrors.malformedVariant() |
There was a problem hiding this 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.
| } | ||
|
|
||
| test("Reject illegal Z85 input - unaligned string") { | ||
| // Minimum string should 5 characters |
There was a problem hiding this 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 branch has not been deployed
2 participants
What changes were proposed in this pull request?
This PR makes it possible for the from_json expression to interpret variant data encoded as Z85 strings and for the to_json expression to encode variant data as Z85 strings if the
alternateVariantEncodingJSON option is set to Z85. In the future, we could support even more alternateVariantEncoding optionsWhy are the changes needed?
The regular JSON representation of variant data is lossy. Z85 is a solution where variant data can be stored in JSON strings in a lossless format. An example of the Z85 being used to encode variants is the representation of variant file skipping statistics in the Delta protocol where file metadata is stored as JSON.
Does this PR introduce any user-facing change?
Yes, it would allow users to specify the alternateVariantEncoding option to these JSON SQL expressions.
How was this patch tested?
Unit tests
Was this patch authored or co-authored using generative AI tooling?
Yes, generated by: OpenAI Codex (GPT 6.1 Sol)