spark-test commented on code in PR #59669:
URL: https://github.com/apache/spark/pull/59669#discussion_r4481813531
##########
common/variant/src/main/java/org/apache/spark/types/variant/VariantBuilder.java:
##########
@@ -205,6 +205,225 @@ public static Variant pickAtPaths(Variant v, PickNode
tree) {
return builder.result();
}
+ /**
+ * Return {@code v} in canonical form, or {@code v} itself if it is already
canonical.
+ * <p>
+ * The variant binary encoding allows the same logical value to be encoded
in several ways
+ * (dictionary order, unused keys, integer and offset widths, decimal scale,
...). Canonical form
+ * picks one encoding per value, so that values which are equal under the
variant equality rules
+ * have identical {@code value} and {@code metadata} bytes and can be
compared, hashed and
+ * grouped bytewise. It is a Spark-internal normalization, not part of the
variant spec, and
+ * nothing outside Spark is expected to produce or require it.
+ * <p>
+ * A variant is canonical when all of the following hold.
+ * <ul>
+ * <li>Metadata header: version {@link VariantUtil#VERSION} and the
minimal offset size for
Review Comment:
**Nit (P3):** The Javadoc further down says `isCanonical` checks exactly
these rules, but the header rules are incomplete. `isCanonical` compares the
whole metadata header byte with `metadataHeader(canonicalOffsetSize)` (line
281), which leaves `sorted_strings` and the unused bit 5 clear. It also
compares whole container header bytes with `objectHeader`/`arrayHeader` (lines
352-353 and 370-371), which leave the reserved bits clear: bit 7 for objects,
and bits 5-7 for arrays. So metadata that meets every rule listed here but sets
`sorted_strings`, as a spec-conformant writer may for exactly such a sorted,
unique dictionary, still gets rebuilt. The 'metadata flags and trailing bytes'
and 'reserved container header bits' tests pin both behaviors, but the
documented rules predict neither. A maintainer following the list, for example
by setting `sorted_strings` for canonical dictionaries in `result()`, would
break the `isCanonical`/`doCanonicalize` agreement that this Javadoc says must
be kept. Stating that canonical headers keep these bits clear would also
cover the earlier request to document every header requirement.
This extends an [existing
discussion](https://github.com/apache/spark/pull/59669#discussion_r4419570081).
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/VariantExpressionEvalUtils.scala:
##########
@@ -76,6 +76,21 @@ object VariantExpressionEvalUtils {
def isValidVariant(input: VariantVal): Boolean =
VariantUtil.isValidVariant(input.getValue, input.getMetadata)
+ def canonicalizeVariant(input: VariantVal): VariantVal = {
+ val variant = new Variant(input.getValue, input.getMetadata)
+ val canonicalized = try {
+ VariantBuilder.canonicalize(variant)
+ } catch {
+ case _: VariantSizeLimitException =>
+ throw QueryExecutionErrors.variantConstructorSizeLimitError()
Review Comment:
**Nit (P3):** `VARIANT_CONSTRUCTOR_SIZE_LIMIT` has a fixed message, 'Cannot
construct a Variant larger than 16 MiB. The maximum allowed size of a Variant
value is 16 MiB.', but outside tests `SIZE_LIMIT` is 128 MiB
(`VariantUtil.java:150-153`). The canonical rebuild can outgrow an input the
constructor already accepted, either by expanding shared payloads or by
widening field IDs after the keys are re-sorted. In production, that overflow
would tell a user whose input may already be larger than 16 MiB that no variant
may exceed 16 MiB. The comment on `variantConstructorSizeLimit` also reserves
this error for oversized bytes from another tool reaching the `Variant`
constructor. The constructor's own message has said 16 MiB since before this
PR. What's new is that a Spark-side rebuild overflow now lands there too.
`parse_json`, `JacksonParser`, `variant_insert`, `variant_set` and
`variant_array_append` all report `VARIANT_SIZE_LIMIT` with the enforced limit.
The 'canonicalize_variant
size limit' test runs at the 16 MiB testing limit, so it can't see the
difference.
A dedicated error condition could keep the internal function name out of the
error, as requested earlier in review, and still report the real limit. It
would take a `sizeLimit` parameter, formatted the way `variantSizeLimitError`
formats it, use SQLSTATE 22023, and replace `variantConstructorSizeLimitError`
here.
**Verification:**
- **Behavior:** Canonicalizing a value whose shared payload expands past the
size limit raises the new error condition, with the enforced limit as its
sizeLimit parameter and no function name.
- **Compatibility:** Oversized value or metadata passed to the Variant
constructor still raises VARIANT_CONSTRUCTOR_SIZE_LIMIT, and the builder-backed
variant functions still raise VARIANT_SIZE_LIMIT with their own function names.
##########
common/variant/src/test/scala/org/apache/spark/types/variant/VariantCanonicalizeSuite.scala:
##########
@@ -0,0 +1,860 @@
+/*
+ * 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.types.variant
+
+import java.nio.charset.StandardCharsets
+
+import scala.util.Random
+
+import org.scalatest.funsuite.AnyFunSuite // scalastyle:ignore funsuite
+
+import org.apache.spark.SparkRuntimeException
+import org.apache.spark.types.variant.VariantUtil._
+
+class VariantCanonicalizeSuite extends AnyFunSuite { // scalastyle:ignore
funsuite
+ private val emptyMetadata = Array[Byte](VERSION, 0, 0)
+ private val nullValue = Array(primitiveHeader(NULL))
+ private val trueValue = Array(primitiveHeader(TRUE))
+ private val falseValue = Array(primitiveHeader(FALSE))
+
+ private def bytes(values: Int*): Array[Byte] = values.map(_.toByte).toArray
+
+ private def utf8(value: String): Array[Byte] =
value.getBytes(StandardCharsets.UTF_8)
+
+ private def littleEndian(value: BigInt, width: Int): Array[Byte] =
+ Array.tabulate(width)(index => ((value >> (8 * index)) & 0xff).toByte)
+
+ private def concatenate(parts: Seq[Array[Byte]]): Array[Byte] =
+ parts.iterator.flatMap(_.iterator).toArray
+
+ private def integer(value: Long, typeInfo: Int = INT1, width: Int = 1):
Array[Byte] =
+ Array(primitiveHeader(typeInfo)) ++ littleEndian(BigInt(value), width)
+
+ private def decimal(
+ unscaled: String,
+ scale: Int,
+ typeInfo: Int,
+ width: Int): Array[Byte] = {
+ Array(primitiveHeader(typeInfo), scale.toByte) ++
littleEndian(BigInt(unscaled), width)
+ }
+
+ private def floating(rawBits: Long, typeInfo: Int, width: Int): Array[Byte] =
+ Array(primitiveHeader(typeInfo)) ++ littleEndian(BigInt(rawBits), width)
+
+ private def string(value: Array[Byte], longEncoding: Boolean = false):
Array[Byte] = {
+ if (longEncoding) {
+ Array(primitiveHeader(LONG_STR)) ++ littleEndian(BigInt(value.length),
4) ++ value
+ } else {
+ Array(shortStrHeader(value.length)) ++ value
+ }
+ }
+
+ private def metadata(
+ keys: Seq[Array[Byte]],
+ offsetSize: Int = 1,
+ flags: Int = 0): Array[Byte] = {
+ val offsets = keys.scanLeft(0)((offset, key) => offset + key.length)
+ bytes(metadataHeader(offsetSize) | flags) ++
littleEndian(BigInt(keys.size), offsetSize) ++
+ concatenate(offsets.map(offset => littleEndian(BigInt(offset),
offsetSize))) ++
+ concatenate(keys)
+ }
+
+ private def dictionary(names: String*): Array[Byte] =
metadata(names.map(utf8))
+
+ private def array(
+ elements: Seq[Array[Byte]],
+ offsetSize: Int = 1,
+ largeSize: Boolean = false): Array[Byte] = {
+ val offsets = elements.scanLeft(0)((offset, element) => offset +
element.length)
+ Array(arrayHeader(largeSize, offsetSize)) ++
+ littleEndian(BigInt(elements.size), if (largeSize) 4 else 1) ++
+ concatenate(offsets.map(offset => littleEndian(BigInt(offset),
offsetSize))) ++
+ concatenate(elements)
+ }
+
+ private def obj(
+ fields: Seq[(Int, Array[Byte])],
+ idSize: Int = 1,
+ offsetSize: Int = 1,
+ largeSize: Boolean = false,
+ physicalOrder: Seq[Int] = Seq.empty): Array[Byte] = {
+ val order = if (physicalOrder.isEmpty) fields.indices else physicalOrder
+ val offsets = new Array[Int](fields.size)
+ var dataSize = 0
+ order.foreach { fieldIndex =>
+ offsets(fieldIndex) = dataSize
+ dataSize += fields(fieldIndex)._2.length
+ }
+ Array(objectHeader(largeSize, idSize, offsetSize)) ++
+ littleEndian(BigInt(fields.size), if (largeSize) 4 else 1) ++
+ concatenate(fields.map { case (keyId, _) => littleEndian(BigInt(keyId),
idSize) }) ++
+ concatenate((offsets.toSeq :+ dataSize).map(offset =>
+ littleEndian(BigInt(offset), offsetSize))) ++
+ concatenate(order.map(fieldIndex => fields(fieldIndex)._2))
+ }
+
+ private def assertBytes(actual: Array[Byte], expected: Array[Byte], label:
String): Unit = {
+ assert(actual.sameElements(expected),
+ s"$label: expected ${expected.length} bytes, got ${actual.length} bytes;
" +
+ s"expected prefix ${expected.take(32).mkString(",")}, " +
+ s"actual prefix ${actual.take(32).mkString(",")}")
+ }
+
+ // The input is canonical exactly when it already has the expected bytes.
+ private def check(
+ input: Variant,
+ expectedValue: Array[Byte],
+ expectedMetadata: Array[Byte] = emptyMetadata): Variant = {
+ val canonicalInput = input.pos == 0 &&
input.value.sameElements(expectedValue) &&
+ input.metadata.sameElements(expectedMetadata)
+ val originalValue = input.value.clone()
+ val originalMetadata = input.metadata.clone()
+ assert(VariantBuilder.isCanonical(input) == canonicalInput)
+ val actual = VariantBuilder.canonicalize(input)
+ assert((actual eq input) == canonicalInput)
+ assert(actual.pos == 0)
+ assertBytes(actual.value, expectedValue, "canonical value")
+ assertBytes(actual.metadata, expectedMetadata, "canonical metadata")
+ assert(VariantBuilder.isCanonical(actual))
+ assert(VariantBuilder.canonicalize(actual) eq actual)
+ Seq(input, actual).foreach { source =>
+ val rebuilt = VariantBuilder.doCanonicalize(source)
+ assertBytes(rebuilt.value, expectedValue, "unconditional rebuild value")
+ assertBytes(rebuilt.metadata, expectedMetadata, "unconditional rebuild
metadata")
+ }
+ assertBytes(input.value, originalValue, "unchanged input value")
+ assertBytes(input.metadata, originalMetadata, "unchanged input metadata")
+ actual
+ }
+
+ private def checkKeyless(inputValue: Array[Byte], expectedValue:
Array[Byte]): Unit =
+ check(new Variant(inputValue, emptyMetadata), expectedValue)
+
+ private def checkError(condition: String)(body: => Any): Unit = {
+ assert(intercept[SparkRuntimeException](body).getCondition == condition)
+ }
+
+ test("canonical encoding has pinned value and metadata bytes") {
+ checkKeyless(bytes(24, 1, 0, 0, 0, 0, 0, 0, 0), bytes(12, 1))
+ checkKeyless(bytes(64, 1, 0, 0, 0, 120), bytes(5, 120))
+ check(VariantBuilder.parseJson("{\"b\":2,\"a\":1}", false),
+ bytes(2, 2, 0, 1, 0, 2, 4, 12, 1, 12, 2), bytes(1, 2, 0, 1, 2, 97, 98))
+ checkKeyless(bytes(2, 0, 0), bytes(2, 0, 0))
+ checkKeyless(bytes(3, 0, 0), bytes(3, 0, 0))
+ }
+
+ test("metadata helpers encode and decode every offset-width header") {
+ Seq((1, 0x01), (2, 0x41), (3, 0x81), (4, 0xc1)).foreach {
+ case (width, header) =>
+ val input = bytes(header) ++ littleEndian(BigInt(2), width) ++
+ concatenate(Seq(0, 1, 2).map(offset => littleEndian(BigInt(offset),
width))) ++
+ bytes(97, 98)
+ assert(metadataHeader(width) == header.toByte)
+ assert(getMetadataOffsetSize(input) == width)
+ assert(getMetadataNumKeys(input, width) == 2)
+ assert(getMetadataStringStart(input, 2, width) == 1 + 4 * width)
+ assert(getMetadataKey(input, 0) == "a")
+ assert(getMetadataKey(input, 1) == "b")
+ }
+ }
+
+ test("isCanonical rejects a metadata last offset past the available string
bytes") {
+ val value = obj(Seq(0 -> integer(1)))
+ val validMetadata = dictionary("ab")
+ assert(VariantBuilder.isCanonical(new Variant(value, validMetadata)))
+
+ Seq(1, 2).foreach { missingBytes =>
+ val truncatedMetadata = validMetadata.dropRight(missingBytes)
+ assert(!VariantBuilder.isCanonical(new Variant(value,
truncatedMetadata)))
+ }
+ }
+
+ test("isCanonical rejects decreasing or out-of-bounds dictionary offsets") {
+ val value = obj(Seq(0 -> integer(1), 1 -> integer(2), 2 -> integer(3)))
+ assert(VariantBuilder.isCanonical(new Variant(value, dictionary("a", "b",
"c"))))
+
+ val decreasingOffsets = bytes(VERSION, 3, 0, 2, 1, 3) ++ utf8("abc")
+ val outOfBoundsOffset = bytes(VERSION, 3, 0, 4, 2, 3) ++ utf8("abc")
Review Comment:
**Nit (P3):** The bound `nextOffset > lastOffset` rejects this fixture at
the first key, but the test doesn't depend on it. Without the bound, the drop
from 4 to 2 fails the monotonicity check one key later, before any key
comparison runs, so the test still passes. No other fixture reaches the bound
either: the malformed-structure metadata cases fail the exact-length check
first, and the fuzz metadata is always well formed. As explained in the thread
on this check, the bound is what keeps the `Arrays.compareUnsigned` key
comparison inside the metadata. Without it, offsets `[0, 4, 4, 3]` over `abc`
reach that comparison with `toIndex` 10 on 9-byte metadata, and `isCanonical`,
and with it `canonicalize`, throws `ArrayIndexOutOfBoundsException` instead of
returning false. Any dictionary whose offsets stay non-decreasing up to an
out-of-range offset ending the second or a later key behaves the same way, for
example `[0, 1, 5, 3]`. To pin the bound, add a `[0, 4, 4, 3]` fixture paired
with this three-field object, so the test can also assert that `canonicalize`
fails with `MALFORMED_VARIANT`.
This extends existing discussions
[1](https://github.com/apache/spark/pull/59669#discussion_r4351197407),
[2](https://github.com/apache/spark/pull/59669#discussion_r4419505117).
See **Shared repair plan 2** in the review body.
##########
common/variant/src/main/java/org/apache/spark/types/variant/VariantBuilder.java:
##########
@@ -205,6 +205,225 @@ public static Variant pickAtPaths(Variant v, PickNode
tree) {
return builder.result();
}
+ /**
+ * Return {@code v} in canonical form, or {@code v} itself if it is already
canonical.
+ * <p>
+ * The variant binary encoding allows the same logical value to be encoded
in several ways
+ * (dictionary order, unused keys, integer and offset widths, decimal scale,
...). Canonical form
+ * picks one encoding per value, so that values which are equal under the
variant equality rules
+ * have identical {@code value} and {@code metadata} bytes and can be
compared, hashed and
+ * grouped bytewise. It is a Spark-internal normalization, not part of the
variant spec, and
+ * nothing outside Spark is expected to produce or require it.
+ * <p>
+ * A variant is canonical when all of the following hold.
+ * <ul>
+ * <li>Metadata header: version {@link VariantUtil#VERSION} and the
minimal offset size for
+ * {@code max(lastOffset, numKeys)}, i.e. 1, 2, 3 or 4 bytes as given
by
+ * {@code getIntegerSize}.</li>
+ * <li>Metadata dictionary: the first offset is 0, offsets are
non-decreasing, the key bytes
+ * end exactly at the last offset (no trailing bytes), keys are
strictly increasing under
+ * unsigned byte comparison (so they are unique), and every key is
referenced by at least
+ * one object field. Key bytes are preserved as is, including invalid
UTF-8.</li>
+ * <li>Value root: starts at position 0 and spans exactly {@code
value.length} bytes.</li>
+ * <li>Object: field ids strictly increasing (fields are stored in key
order and no key is
+ * repeated), field data stored contiguously in field order, the end
offset equal to the
+ * data size, and the header using the minimal id size for the largest
id, the minimal
+ * offset size for the data size, and {@code is_large} only for more
than 255 fields.</li>
+ * <li>Array: element order preserved, element data stored contiguously in
element order, the
+ * end offset equal to the data size, and the header using the minimal
offset size for the
+ * data size and {@code is_large} only for more than 255 elements.</li>
+ * <li>Integer: the smallest of INT1, INT2, INT4 and INT8 that holds the
value.</li>
+ * <li>Decimal: trailing zeros stripped and a negative scale raised to 0.
An integral decimal
+ * that fits in a signed 64-bit integer is stored as an integer
instead. Otherwise the
+ * smallest of DECIMAL4 (precision and scale at most 9), DECIMAL8 (at
most 18) and
+ * DECIMAL16 that holds the value.</li>
+ * <li>Float and double: -0.0 is stored as 0.0 and NaN as the canonical
NaN bit pattern of
+ * {@link Float#floatToIntBits} and {@link Double#doubleToLongBits}.
The type is kept, so
+ * a float and a double are never identical.</li>
+ * <li>String: at most {@link VariantUtil#MAX_SHORT_STR_SIZE} bytes uses
the short-string
+ * encoding, anything longer the long-string encoding. Bytes are
preserved as is, including
+ * invalid UTF-8.</li>
+ * <li>Every other type (null, boolean, date, timestamps, binary, uuid,
...) has a single
+ * encoding in the spec and is copied unchanged.</li>
+ * </ul>
+ * {@code isCanonical} checks exactly these rules and {@code doCanonicalize}
produces them; the
+ * two must be changed together when a rule changes or a type with
alternative encodings is
+ * added to the spec.
+ */
+ public static Variant canonicalize(Variant v) {
+ if (isCanonical(v)) {
+ return v;
+ }
+ return doCanonicalize(v);
+ }
+
+ /**
+ * Check whether {@code v} already satisfies every rule listed in {@link
#canonicalize}. Return
+ * false at the first non-canonical encoding. Malformed bytes also return
false so that the
Review Comment:
**Nit (P3):** Some malformed content throws from inside the check instead of
returning false. The DECIMAL branch calls `getDecimalWithOriginalScale` (line
376), whose `checkDecimal` throws `MALFORMED_VARIANT` for an out-of-range
precision or scale, even though the header is readable and the root `valueSize`
check passed. The suite's own malformed inputs `decimal("1", 39, DECIMAL16,
16)` and `decimal("1000000000", 0, DECIMAL4, 4)` do exactly that with the empty
metadata. The test only asserts that `canonicalize` fails, which it does either
way. A container whose offset list runs past the value also throws, from
`valueSize`. Behavior is fine today, because `canonicalize` is the only caller
and fails with the same error in the rebuild. But a caller that relies on this
Javadoc and uses `isCanonical` as a predicate would get an exception instead of
`false`. The Javadoc should say that malformed content the check has to decode,
such as decimals or container offset lists, can raise `MALF
ORMED_VARIANT`, and that an unknown type ID raises
`UNKNOWN_PRIMITIVE_TYPE_IN_VARIANT`.
See **Shared repair plan 1** in the review body.
##########
common/variant/src/test/scala/org/apache/spark/types/variant/VariantCanonicalizeSuite.scala:
##########
@@ -0,0 +1,860 @@
+/*
+ * 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.types.variant
+
+import java.nio.charset.StandardCharsets
+
+import scala.util.Random
+
+import org.scalatest.funsuite.AnyFunSuite // scalastyle:ignore funsuite
+
+import org.apache.spark.SparkRuntimeException
+import org.apache.spark.types.variant.VariantUtil._
+
+class VariantCanonicalizeSuite extends AnyFunSuite { // scalastyle:ignore
funsuite
+ private val emptyMetadata = Array[Byte](VERSION, 0, 0)
+ private val nullValue = Array(primitiveHeader(NULL))
+ private val trueValue = Array(primitiveHeader(TRUE))
+ private val falseValue = Array(primitiveHeader(FALSE))
+
+ private def bytes(values: Int*): Array[Byte] = values.map(_.toByte).toArray
+
+ private def utf8(value: String): Array[Byte] =
value.getBytes(StandardCharsets.UTF_8)
+
+ private def littleEndian(value: BigInt, width: Int): Array[Byte] =
+ Array.tabulate(width)(index => ((value >> (8 * index)) & 0xff).toByte)
+
+ private def concatenate(parts: Seq[Array[Byte]]): Array[Byte] =
+ parts.iterator.flatMap(_.iterator).toArray
+
+ private def integer(value: Long, typeInfo: Int = INT1, width: Int = 1):
Array[Byte] =
+ Array(primitiveHeader(typeInfo)) ++ littleEndian(BigInt(value), width)
+
+ private def decimal(
+ unscaled: String,
+ scale: Int,
+ typeInfo: Int,
+ width: Int): Array[Byte] = {
+ Array(primitiveHeader(typeInfo), scale.toByte) ++
littleEndian(BigInt(unscaled), width)
+ }
+
+ private def floating(rawBits: Long, typeInfo: Int, width: Int): Array[Byte] =
+ Array(primitiveHeader(typeInfo)) ++ littleEndian(BigInt(rawBits), width)
+
+ private def string(value: Array[Byte], longEncoding: Boolean = false):
Array[Byte] = {
+ if (longEncoding) {
+ Array(primitiveHeader(LONG_STR)) ++ littleEndian(BigInt(value.length),
4) ++ value
+ } else {
+ Array(shortStrHeader(value.length)) ++ value
+ }
+ }
+
+ private def metadata(
+ keys: Seq[Array[Byte]],
+ offsetSize: Int = 1,
+ flags: Int = 0): Array[Byte] = {
+ val offsets = keys.scanLeft(0)((offset, key) => offset + key.length)
+ bytes(metadataHeader(offsetSize) | flags) ++
littleEndian(BigInt(keys.size), offsetSize) ++
+ concatenate(offsets.map(offset => littleEndian(BigInt(offset),
offsetSize))) ++
+ concatenate(keys)
+ }
+
+ private def dictionary(names: String*): Array[Byte] =
metadata(names.map(utf8))
+
+ private def array(
+ elements: Seq[Array[Byte]],
+ offsetSize: Int = 1,
+ largeSize: Boolean = false): Array[Byte] = {
+ val offsets = elements.scanLeft(0)((offset, element) => offset +
element.length)
+ Array(arrayHeader(largeSize, offsetSize)) ++
+ littleEndian(BigInt(elements.size), if (largeSize) 4 else 1) ++
+ concatenate(offsets.map(offset => littleEndian(BigInt(offset),
offsetSize))) ++
+ concatenate(elements)
+ }
+
+ private def obj(
+ fields: Seq[(Int, Array[Byte])],
+ idSize: Int = 1,
+ offsetSize: Int = 1,
+ largeSize: Boolean = false,
+ physicalOrder: Seq[Int] = Seq.empty): Array[Byte] = {
+ val order = if (physicalOrder.isEmpty) fields.indices else physicalOrder
+ val offsets = new Array[Int](fields.size)
+ var dataSize = 0
+ order.foreach { fieldIndex =>
+ offsets(fieldIndex) = dataSize
+ dataSize += fields(fieldIndex)._2.length
+ }
+ Array(objectHeader(largeSize, idSize, offsetSize)) ++
+ littleEndian(BigInt(fields.size), if (largeSize) 4 else 1) ++
+ concatenate(fields.map { case (keyId, _) => littleEndian(BigInt(keyId),
idSize) }) ++
+ concatenate((offsets.toSeq :+ dataSize).map(offset =>
+ littleEndian(BigInt(offset), offsetSize))) ++
+ concatenate(order.map(fieldIndex => fields(fieldIndex)._2))
+ }
+
+ private def assertBytes(actual: Array[Byte], expected: Array[Byte], label:
String): Unit = {
+ assert(actual.sameElements(expected),
+ s"$label: expected ${expected.length} bytes, got ${actual.length} bytes;
" +
+ s"expected prefix ${expected.take(32).mkString(",")}, " +
+ s"actual prefix ${actual.take(32).mkString(",")}")
+ }
+
+ // The input is canonical exactly when it already has the expected bytes.
+ private def check(
+ input: Variant,
+ expectedValue: Array[Byte],
+ expectedMetadata: Array[Byte] = emptyMetadata): Variant = {
+ val canonicalInput = input.pos == 0 &&
input.value.sameElements(expectedValue) &&
+ input.metadata.sameElements(expectedMetadata)
+ val originalValue = input.value.clone()
+ val originalMetadata = input.metadata.clone()
+ assert(VariantBuilder.isCanonical(input) == canonicalInput)
+ val actual = VariantBuilder.canonicalize(input)
+ assert((actual eq input) == canonicalInput)
+ assert(actual.pos == 0)
+ assertBytes(actual.value, expectedValue, "canonical value")
+ assertBytes(actual.metadata, expectedMetadata, "canonical metadata")
+ assert(VariantBuilder.isCanonical(actual))
+ assert(VariantBuilder.canonicalize(actual) eq actual)
+ Seq(input, actual).foreach { source =>
+ val rebuilt = VariantBuilder.doCanonicalize(source)
+ assertBytes(rebuilt.value, expectedValue, "unconditional rebuild value")
+ assertBytes(rebuilt.metadata, expectedMetadata, "unconditional rebuild
metadata")
+ }
+ assertBytes(input.value, originalValue, "unchanged input value")
+ assertBytes(input.metadata, originalMetadata, "unchanged input metadata")
+ actual
+ }
+
+ private def checkKeyless(inputValue: Array[Byte], expectedValue:
Array[Byte]): Unit =
+ check(new Variant(inputValue, emptyMetadata), expectedValue)
+
+ private def checkError(condition: String)(body: => Any): Unit = {
+ assert(intercept[SparkRuntimeException](body).getCondition == condition)
+ }
+
+ test("canonical encoding has pinned value and metadata bytes") {
+ checkKeyless(bytes(24, 1, 0, 0, 0, 0, 0, 0, 0), bytes(12, 1))
+ checkKeyless(bytes(64, 1, 0, 0, 0, 120), bytes(5, 120))
+ check(VariantBuilder.parseJson("{\"b\":2,\"a\":1}", false),
+ bytes(2, 2, 0, 1, 0, 2, 4, 12, 1, 12, 2), bytes(1, 2, 0, 1, 2, 97, 98))
+ checkKeyless(bytes(2, 0, 0), bytes(2, 0, 0))
+ checkKeyless(bytes(3, 0, 0), bytes(3, 0, 0))
+ }
+
+ test("metadata helpers encode and decode every offset-width header") {
+ Seq((1, 0x01), (2, 0x41), (3, 0x81), (4, 0xc1)).foreach {
+ case (width, header) =>
+ val input = bytes(header) ++ littleEndian(BigInt(2), width) ++
+ concatenate(Seq(0, 1, 2).map(offset => littleEndian(BigInt(offset),
width))) ++
+ bytes(97, 98)
+ assert(metadataHeader(width) == header.toByte)
+ assert(getMetadataOffsetSize(input) == width)
+ assert(getMetadataNumKeys(input, width) == 2)
+ assert(getMetadataStringStart(input, 2, width) == 1 + 4 * width)
+ assert(getMetadataKey(input, 0) == "a")
+ assert(getMetadataKey(input, 1) == "b")
+ }
+ }
+
+ test("isCanonical rejects a metadata last offset past the available string
bytes") {
+ val value = obj(Seq(0 -> integer(1)))
+ val validMetadata = dictionary("ab")
+ assert(VariantBuilder.isCanonical(new Variant(value, validMetadata)))
+
+ Seq(1, 2).foreach { missingBytes =>
+ val truncatedMetadata = validMetadata.dropRight(missingBytes)
+ assert(!VariantBuilder.isCanonical(new Variant(value,
truncatedMetadata)))
+ }
+ }
+
+ test("isCanonical rejects decreasing or out-of-bounds dictionary offsets") {
+ val value = obj(Seq(0 -> integer(1), 1 -> integer(2), 2 -> integer(3)))
+ assert(VariantBuilder.isCanonical(new Variant(value, dictionary("a", "b",
"c"))))
+
+ val decreasingOffsets = bytes(VERSION, 3, 0, 2, 1, 3) ++ utf8("abc")
+ val outOfBoundsOffset = bytes(VERSION, 3, 0, 4, 2, 3) ++ utf8("abc")
+ Seq(decreasingOffsets, outOfBoundsOffset).foreach { invalidMetadata =>
+ assert(!VariantBuilder.isCanonical(new Variant(value, invalidMetadata)))
+ }
+ }
+
+ test("getMetadataKey rejects a dictionary size whose string start overflows
an int") {
+ val forged = Array(metadataHeader(4)) ++ littleEndian(BigInt(Int.MaxValue
- 1), 4) ++
+ littleEndian(BigInt(0), 4) ++ littleEndian(BigInt(1), 4) ++ utf8("a")
+ checkError("MALFORMED_VARIANT") { getMetadataKey(forged, 0) }
+ }
+
+ test("dictionary permutation and object payload order canonicalize
independently") {
+ val expectedValue = obj(Seq(0 -> integer(1), 1 -> integer(2)))
+ Seq(Seq("a", "b"), Seq("b", "a")).foreach { keys =>
+ Seq(Seq(0, 1), Seq(1, 0)).foreach { order =>
+ val value = obj(Seq(keys.indexOf("a") -> integer(1), keys.indexOf("b")
-> integer(2)),
+ physicalOrder = order)
+ check(new Variant(value, dictionary(keys: _*)), expectedValue,
dictionary("a", "b"))
+ }
+ }
+ }
+
+ test("unused dictionary entries are removed and repeated names share an ID")
{
+ val sourceMetadata = dictionary("b", "unused", "a", "a")
+ val sourceValue = array(Seq(
+ obj(Seq(2 -> integer(1))), obj(Seq(3 -> integer(2))), obj(Seq(0 ->
integer(3)))))
+ val expectedValue = array(Seq(
+ obj(Seq(0 -> integer(1))), obj(Seq(0 -> integer(2))), obj(Seq(1 ->
integer(3)))))
+ check(new Variant(sourceValue, sourceMetadata), expectedValue,
dictionary("a", "b"))
+ check(new Variant(integer(1), sourceMetadata), integer(1))
+ check(new Variant(integer(1), metadata(Seq(bytes(0xff)))), integer(1))
+ check(new Variant(obj(Seq(0 -> integer(1))), bytes(1, 1, 1, 2, 120, 97)),
+ obj(Seq(0 -> integer(1))), dictionary("a"))
+ val emptyKey = obj(Seq(0 -> nullValue, 1 -> integer(1)))
+ check(new Variant(emptyKey, dictionary("", "a")), emptyKey, dictionary("",
"a"))
+ }
+
+ test("duplicate dictionary entries and repeated field IDs are never
canonical") {
+ check(new Variant(array(Seq(obj(Seq(0 -> integer(1))), obj(Seq(1 ->
integer(2))))),
+ dictionary("a", "a")),
+ array(Seq(obj(Seq(0 -> integer(1))), obj(Seq(0 -> integer(2))))),
dictionary("a"))
+ val repeatedId = new Variant(obj(Seq(0 -> integer(1), 0 -> integer(2))),
dictionary("a"))
+ assert(!VariantBuilder.isCanonical(repeatedId))
+ checkError("VARIANT_DUPLICATE_KEY") {
VariantBuilder.canonicalize(repeatedId) }
+ }
+
+ // scalastyle:off nonascii
+ test("dictionary and fields use unsigned UTF-8 rather than UTF-16 ordering")
{
+ val bmpKey = utf8("\ue000")
+ val supplementaryKey = utf8("\ud800\udc00")
+ val expectedMetadata = metadata(Seq(bmpKey, supplementaryKey))
+ val expectedValue = obj(Seq(0 -> integer(10), 1 -> integer(20)))
+ Seq(Seq(0 -> integer(20), 1 -> integer(10)),
+ Seq(1 -> integer(10), 0 -> integer(20))).foreach { fields =>
+ check(new Variant(obj(fields), metadata(Seq(supplementaryKey, bmpKey))),
+ expectedValue, expectedMetadata)
+ }
+ }
+
+ test("valid multi-byte UTF-8 keys are kept and sorted by their bytes") {
+ check(new Variant(obj(Seq(2 -> integer(1), 1 -> integer(2), 0 ->
integer(3))),
+ dictionary("\u4f60", "caf\u00e9", "cafe")),
+ obj(Seq(0 -> integer(1), 1 -> integer(2), 2 -> integer(3))),
+ dictionary("cafe", "caf\u00e9", "\u4f60"))
+ }
+ // scalastyle:on nonascii
+
+ test("all independent metadata ID offset and count widths converge") {
+ val fields = Seq(0 -> trueValue, 1 -> integer(258, INT2, 2), 2 ->
nullValue)
+ val elements = fields.map(_._2)
+ val keys = Seq("a", "b", "c").map(utf8)
+ for {
+ metadataWidth <- 1 to 4
+ idWidth <- 1 to 4
+ offsetWidth <- 1 to 4
+ largeSize <- Seq(false, true)
+ } {
+ withClue(s"metadata=$metadataWidth id=$idWidth offset=$offsetWidth
large=$largeSize: ") {
+ check(new Variant(obj(fields, idWidth, offsetWidth, largeSize),
+ metadata(keys, metadataWidth)), obj(fields), metadata(keys))
+ }
+ }
+ for {
+ metadataWidth <- 1 to 4
+ offsetWidth <- 1 to 4
+ largeSize <- Seq(false, true)
+ } {
+ val keylessMetadata = metadata(Seq.empty, metadataWidth)
+ check(new Variant(array(elements, offsetWidth, largeSize),
keylessMetadata),
+ array(elements))
+ check(new Variant(array(Seq.empty, offsetWidth, largeSize),
keylessMetadata),
+ array(Seq.empty))
+ check(new Variant(obj(Seq.empty, 4, offsetWidth, largeSize),
keylessMetadata),
+ obj(Seq.empty))
+ }
+ }
+
+ test("metadata flags and trailing bytes are not part of canonical output") {
+ Seq(1 << 4, 1 << 5, (1 << 4) | (1 << 5)).foreach { flags =>
+ check(new Variant(integer(1), metadata(Seq.empty, flags = flags)),
integer(1))
+ }
+ checkKeyless(integer(1) ++ bytes(0, 0), integer(1))
+ check(new Variant(integer(1), emptyMetadata ++ bytes(0, 0)), integer(1))
+ }
+
+ test("reserved container header bits are cleared during normalization") {
+ val objectValue = obj(Seq(0 -> integer(1)))
+ val objectWithReservedBit = objectValue.clone()
+ objectWithReservedBit(0) = (objectWithReservedBit(0) | 0x80).toByte
+ check(new Variant(objectWithReservedBit, dictionary("a")), objectValue,
dictionary("a"))
+ val arrayValue = array(Seq(integer(1)))
+ Seq(0x20, 0x40, 0x80, 0xe0).foreach { reservedBits =>
+ val arrayWithReservedBits = arrayValue.clone()
+ arrayWithReservedBits(0) = (arrayWithReservedBits(0) |
reservedBits).toByte
+ checkKeyless(arrayWithReservedBits, arrayValue)
+ }
+ }
+
+ test("padding and shared payloads inside containers become contiguous") {
+ val paddedArray = Array(arrayHeader(false, 1)) ++ bytes(1, 0, 3) ++
integer(1) ++ bytes(0)
+ checkKeyless(paddedArray, array(Seq(integer(1))))
+ val paddedObject =
+ Array(objectHeader(false, 1, 1)) ++ bytes(1, 0, 0, 3) ++ integer(1) ++
bytes(0)
+ check(new Variant(paddedObject, dictionary("a")), obj(Seq(0 ->
integer(1))), dictionary("a"))
+ val shared = Array(objectHeader(false, 1, 1)) ++ bytes(2, 0, 1, 0, 0, 2)
++ integer(7)
+ check(new Variant(shared, dictionary("a", "b")),
+ obj(Seq(0 -> integer(7), 1 -> integer(7))), dictionary("a", "b"))
+ }
+
+ test("container count and payload widths change at 255 and 256") {
+ Seq((255, false, 1, 1), (256, true, 1, 2), (257, true, 2, 2)).foreach {
+ case (count, largeSize, idWidth, offsetWidth) =>
+ val keys = (0 until count).map(index => utf8(f"key$index%03d"))
+ val fields = (0 until count).map(index => index -> nullValue)
+ val expectedMetadata = metadata(keys, 2)
+ val expectedObject = obj(fields, idWidth, offsetWidth, largeSize)
+ check(new Variant(obj(fields, 4, 4, largeSize = true), metadata(keys,
4)),
+ expectedObject, expectedMetadata)
+ check(new Variant(expectedObject, expectedMetadata), expectedObject,
expectedMetadata)
+ val elements = Seq.fill(count)(nullValue)
+ checkKeyless(array(elements, 4, largeSize = true), array(elements,
offsetWidth, largeSize))
+ }
+ }
+
+ test("object ID width follows the largest referenced global ID") {
+ val keys = (0 to 256).map(index => utf8(f"key$index%03d"))
+ val sourceElements = keys.indices.map(index => obj(Seq(index ->
nullValue), 4, 4, true))
+ val expectedElements = keys.indices.map { index =>
+ obj(Seq(index -> nullValue), idSize = if (index <= U8_MAX) 1 else 2)
+ }
+ val expectedValue = array(expectedElements, 2, largeSize = true)
+ check(new Variant(array(sourceElements, 4, largeSize = true),
metadata(keys, 4)),
+ expectedValue, metadata(keys, 2))
+ check(new Variant(obj(Seq(256 -> integer(1)), 2), metadata(keys, 2)),
+ obj(Seq(0 -> integer(1))), metadata(Seq(keys.last)))
+ }
+
+ test("metadata and container offsets have real one two and three byte
boundaries") {
+ Seq((255, 1), (256, 2), (65535, 2), (65536, 3)).foreach {
+ case (size, expectedWidth) =>
+ withClue(s"size=$size width=$expectedWidth: ") {
+ val key = Array.fill[Byte](size)('a'.toByte)
+ val objectValue = obj(Seq(0 -> nullValue))
+ val expectedMetadata = metadata(Seq(key), expectedWidth)
+ check(new Variant(objectValue, metadata(Seq(key), 4)), objectValue,
expectedMetadata)
+ check(new Variant(objectValue, expectedMetadata), objectValue,
expectedMetadata)
+ val payload = Array(primitiveHeader(BINARY)) ++
littleEndian(BigInt(size - 5), 4) ++
+ Array.fill[Byte](size - 5)(7)
+ checkKeyless(array(Seq(payload), 4), array(Seq(payload),
expectedWidth))
+ val expectedObject = obj(Seq(0 -> payload), offsetSize =
expectedWidth)
+ check(new Variant(obj(Seq(0 -> payload), offsetSize = 4),
dictionary("a")),
+ expectedObject, dictionary("a"))
+ }
+ }
+ }
+
+ test("signed integer boundaries choose the smallest encoding") {
+ val boundaries = Seq(
+ (Long.MinValue, INT8, 8), (Int.MinValue.toLong - 1, INT8, 8),
+ (Int.MinValue.toLong, INT4, 4), (Short.MinValue.toLong - 1, INT4, 4),
+ (Short.MinValue.toLong, INT2, 2), (Byte.MinValue.toLong - 1, INT2, 2),
+ (Byte.MinValue.toLong, INT1, 1), (-1L, INT1, 1), (0L, INT1, 1), (1L,
INT1, 1),
+ (Byte.MaxValue.toLong, INT1, 1), (Byte.MaxValue.toLong + 1, INT2, 2),
+ (Short.MaxValue.toLong, INT2, 2), (Short.MaxValue.toLong + 1, INT4, 4),
+ (Int.MaxValue.toLong, INT4, 4), (Int.MaxValue.toLong + 1, INT8, 8),
+ (Long.MaxValue, INT8, 8))
+ boundaries.foreach { case (number, expectedType, expectedWidth) =>
+ Seq((INT1, 1), (INT2, 2), (INT4, 4), (INT8, 8)).foreach { case
(typeInfo, width) =>
+ if (width >= expectedWidth) {
+ withClue(s"integer=$number width=$width: ") {
+ checkKeyless(integer(number, typeInfo, width),
+ integer(number, expectedType, expectedWidth))
+ }
+ }
+ }
+ }
+ }
+
+ test("decimal normalization respects precision scale and signed long
promotion limits") {
+ val cases = Seq(
Review Comment:
**Nit (P3):** Every promotable case here carries trailing zeros: `("0",
30)`, `("100", 2)`, `("100000", 2)` and the long-boundary values at scale 1.
The later `original.scale() == canonical.scale()` check in `isValueCanonical`
would reject those even without `decimalPromotesToLong`. The only scale-0 case,
1E20, doesn't fit a long. The fuzz inputs come from `parseJson`, which never
emits a long-fitting scale-0 decimal, and `randomizeBinaryEncoding` copies
decimals unchanged. So the promotion check in `isCanonical`
(`VariantBuilder.java:378-380`), the only rule that rejects a value like
DECIMAL4 scale 0 unscaled 5, has no failure signal. If it were removed or
narrowed, `isCanonical` would accept that value as canonical, although `INT1 5`
is the canonical form. Equal values would then keep different bytes, and every
test would still pass. Spark writes exactly this shape when casting
`DECIMAL(p,0)` values to variant, because `castToVariant` passes each decimal
to `appendDecimal` with
its own scale. A scale-0 case such as `decimal("5", 0, DECIMAL4, 4)` expecting
`integer(5)`, plus negative and long-boundary variants, would let `check()`
catch it.
See **Shared repair plan 2** in the review body.
##########
common/variant/src/main/java/org/apache/spark/types/variant/VariantBuilder.java:
##########
@@ -555,18 +806,163 @@ private void appendVariantImpl(byte[] value, byte[]
metadata, int pos) {
int offset = readUnsigned(value, offsetStart + offsetSize * i,
offsetSize);
int elementPos = dataStart + offset;
offsets.add(writePos - start);
- appendVariantImpl(value, metadata, elementPos);
+ appendVariantImpl(value, metadata, elementPos, canonicalIds);
}
finishWritingArray(start, offsets);
return null;
});
break;
+ default:
+ if (canonicalIds != null) {
+ appendCanonicalizedScalar(value, pos);
+ } else {
+ shallowAppendVariantImpl(value, pos);
+ }
+ break;
+ }
+ }
+
+ // Append the scalar at `value[pos...]` in canonical form. Keep in sync with
the scalar cases
+ // of `isValueCanonical`.
+ private void appendCanonicalizedScalar(byte[] value, int pos) {
+ switch (getType(value, pos)) {
+ case LONG:
+ appendLong(getLong(value, pos));
+ break;
+ case DECIMAL: {
+ BigDecimal stripped = getDecimal(value, pos);
+ if (decimalPromotesToLong(stripped)) {
+ appendLong(stripped.longValue());
+ } else {
+ appendDecimal(canonicalDecimalForm(stripped));
+ }
+ break;
+ }
+ case FLOAT:
+ appendFloat(normalizeNegativeZero(getFloat(value, pos)));
+ break;
+ case DOUBLE:
+ appendDouble(normalizeNegativeZero(getDouble(value, pos)));
+ break;
+ case STRING: {
+ int size = valueSize(value, pos);
+ if (size <= 0 || (long) pos + size > value.length) {
+ throw malformedVariant();
+ }
+ int headerSize = (value[pos] & BASIC_TYPE_MASK) == SHORT_STR ? 1 : 1 +
U32_SIZE;
+ appendString(value, pos + headerSize, size - headerSize);
+ break;
+ }
default:
shallowAppendVariantImpl(value, pos);
break;
}
}
+ private void buildCanonicalized(byte[] value, byte[] metadata, int pos) {
+ int offsetSize = getMetadataOffsetSize(metadata);
+ int numKeys = getMetadataNumKeys(metadata, offsetSize);
+ // Validate metadata table bounds before allocating the key-ID mapping.
+ getMetadataStringStart(metadata, numKeys, offsetSize);
+ int[] canonicalIds = new int[numKeys];
+ Arrays.fill(canonicalIds, -1);
+ TreeMap<byte[], Integer> keys = new TreeMap<>(VariantUtil::compareKeys);
+ collectAllObjectKeys(value, metadata, pos, canonicalIds, keys);
+ // Remap the IDs assigned during collection to unsigned byte order without
decoding keys.
+ int[] sortedIds = new int[keys.size()];
+ for (java.util.Map.Entry<byte[], Integer> key : keys.entrySet()) {
+ sortedIds[key.getValue()] = dictionaryKeys.size();
+ dictionaryKeys.add(key.getKey());
+ }
+ for (int i = 0; i < canonicalIds.length; ++i) {
+ if (canonicalIds[i] >= 0) {
+ canonicalIds[i] = sortedIds[canonicalIds[i]];
+ }
+ }
+ appendVariantImpl(value, metadata, pos, canonicalIds);
+ }
+
+ private static void collectAllObjectKeys(
+ byte[] value, byte[] metadata, int pos, int[] canonicalIds,
TreeMap<byte[], Integer> keys) {
+ checkIndex(pos, value.length);
+ switch (value[pos] & BASIC_TYPE_MASK) {
+ case OBJECT:
+ handleObject(value, pos, (size, idSize, offsetSize, idStart,
offsetStart, dataStart) -> {
+ for (int fieldIndex = 0; fieldIndex < size; ++fieldIndex) {
+ int keyId = readUnsigned(value, idStart + idSize * fieldIndex,
idSize);
+ int offset = readUnsigned(value, offsetStart + offsetSize *
fieldIndex, offsetSize);
+ if (keyId >= canonicalIds.length) {
+ throw malformedVariant();
+ }
+ if (canonicalIds[keyId] < 0) {
+ byte[] key = getMetadataKeyBytes(metadata, keyId);
+ Integer id = keys.get(key);
+ if (id == null) {
+ id = keys.size();
+ keys.put(key, id);
+ }
+ canonicalIds[keyId] = id;
+ }
+ collectAllObjectKeys(value, metadata, dataStart + offset,
canonicalIds, keys);
+ }
+ return null;
+ });
+ break;
+ case ARRAY:
+ handleArray(value, pos, (size, offsetSize, offsetStart, dataStart) -> {
+ for (int elementIndex = 0; elementIndex < size; ++elementIndex) {
+ int offset = readUnsigned(value, offsetStart + offsetSize *
elementIndex, offsetSize);
+ collectAllObjectKeys(value, metadata, dataStart + offset,
canonicalIds, keys);
Review Comment:
**Non-blocking (P2):** This recursion follows every element reference once
per path that reaches it and never writes, so `checkCapacity` doesn't bound it.
The size limit only applies once the rebuild in `appendVariantImpl` starts
writing. The change accepts container offsets that alias one child and expands
them (see the 'padding and shared payloads inside containers become contiguous'
test), and the `Variant` constructor checks only the version and the size. So a
value of a few hundred bytes made of 60 nested two-element arrays, each with
offsets `[0, 0, childSize]`, gets through. `isCanonical` gives up after one
path, and this pre-pass then makes about 2^61 calls before anything is written.
Instead of failing with the size-limit error, `canonicalize` spins effectively
forever. The variant code has no interruption check, so cancelling the task
doesn't stop it. Only direct callers can reach this today. Once the expression
is injected before GROUP BY/DISTINCT, though, it would run
on every stored row, including variant bytes written by other tools.
One way to bound it at no cost to ordinary values: count the values this
walk visits, and throw `VariantSizeLimitException` once the count exceeds
`SIZE_LIMIT`. The rebuild writes every visited value as at least one byte, so a
value past that count would overflow the write pass anyway. A value that never
references the same position twice can't reach the count.
**Verification:**
- **Regression:** Canonicalizing a value of several dozen nested arrays
whose elements all alias one child fails quickly with the size-limit error
under the test size limit, instead of hanging while keys are collected.
- **Behavior:** The existing shared-payload, padding, permutation and fuzz
canonicalization cases produce the same canonical bytes as before.
--
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]
