xiangfu0 commented on code in PR #19101: URL: https://github.com/apache/pinot/pull/19101#discussion_r3920903302
########## docs/design/VARIANT.md: ########## @@ -0,0 +1,344 @@ +<!-- + + 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. + +--> +# Apache Pinot VARIANT — Design Document Review Comment: Moved the spec to `docs/design/VARIANT.md`, updated the quickstart and PR links, and corrected section 8 to say 22/23 are already assigned to UUID/UUID_ARRAY. ########## pinot-common/src/main/proto/expressions.proto: ########## @@ -46,6 +46,7 @@ enum ColumnDataType { BIG_DECIMAL_ARRAY = 21; UUID = 22; UUID_ARRAY = 23; + VARIANT = 24; Review Comment: Kept wire value 24 and the frozen legacy-proto `UNRECOGNIZED` assertion. The PR body now calls this out as an explicit independent sign-off requirement, and I requested reviews from you and `raghavyadav01`. I am leaving this thread open for that sign-off rather than self-resolving it. ########## pinot-spi/src/main/java/org/apache/pinot/spi/utils/VariantEnvelope.java: ########## @@ -0,0 +1,243 @@ +/** + * 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.pinot.spi.utils; + +import java.nio.ByteBuffer; +import java.util.Objects; +import javax.annotation.Nullable; + + +/// Pinot-owned framing for the two buffers that make up a Parquet Variant value. +/// +/// The version-1 wire format is: +/// ``` +/// 0 4 bytes ASCII magic "PVAR" +/// 4 1 byte envelope version (1) +/// 5 1 byte flags (0) +/// 6 2 bytes reserved (0) +/// 8 4 bytes metadata length, unsigned range restricted to Java array sizes +/// 12 4 bytes value length, unsigned range restricted to Java array sizes +/// 16 M bytes Parquet Variant metadata +/// 16 + M V bytes Parquet Variant value +/// ``` +/// +/// An empty byte array is deliberately not an envelope. Pinot reserves it as the default null value for a +/// `VARIANT` field, allowing the null-value vector to distinguish SQL null from an encoded Variant null. +/// +/// This class validates only Pinot's stable outer framing. Producers and consumers remain responsible for validating +/// the Parquet Variant metadata and value payloads. +public final class VariantEnvelope { + public static final int HEADER_SIZE = 16; + public static final byte VERSION = 1; + public static final byte FLAGS = 0; + + private static final int MAGIC = 0x50564152; // ASCII "PVAR" + + private VariantEnvelope() { + } + + /// Encodes the remaining bytes of the supplied metadata and value buffers without changing their positions or + /// limits. + /// + /// Array-backed buffers are copied directly from their backing arrays. Other buffers, including direct and + /// read-only buffers, are read through independent duplicate views. + public static byte[] encode(ByteBuffer metadata, ByteBuffer value) { + Objects.requireNonNull(metadata, "metadata must not be null"); + Objects.requireNonNull(value, "value must not be null"); + + int metadataLength = metadata.remaining(); + int valueLength = value.remaining(); + byte[] envelope = allocate(metadataLength, valueLength); + copyRemaining(metadata, envelope, HEADER_SIZE); + copyRemaining(value, envelope, HEADER_SIZE + metadataLength); + return envelope; + } + + /// Encodes slices of the supplied arrays without allocating intermediate buffer views. + public static byte[] encode(byte[] metadata, int metadataOffset, int metadataLength, byte[] value, int valueOffset, + int valueLength) { + Objects.requireNonNull(metadata, "metadata must not be null"); + Objects.requireNonNull(value, "value must not be null"); + requireRange(metadata, metadataOffset, metadataLength, "metadata"); + requireRange(value, valueOffset, valueLength, "value"); + + byte[] envelope = allocate(metadataLength, valueLength); + System.arraycopy(metadata, metadataOffset, envelope, HEADER_SIZE, metadataLength); + System.arraycopy(value, valueOffset, envelope, HEADER_SIZE + metadataLength, valueLength); + return envelope; + } + + /// Decodes and validates an envelope, returning zero-copy, read-only views over its metadata and value buffers. + /// + /// The returned views alias `envelope`; this method does not copy either payload. The decoded object and any views + /// obtained from it keep the backing array alive, so the caller does not need to retain a separate reference to + /// `envelope`. Mutations made to the input array after this method returns are visible through the views and can + /// corrupt the decoded payload. Callers must therefore treat the input array as immutable for as long as the decoded + /// object or any returned view may be used. + /// + /// The decoded holder is safe for concurrent reads when the aliased input array is not mutated. Each accessor returns + /// a read-only view with independent position and limit, so cursor movement by one reader does not affect another + /// reader. + public static Decoded decode(byte[] envelope) { Review Comment: Renamed the public factory to `allocateUnfilled` and aligned the class/spec wording to big-endian signed int restricted to >= 0. Golden and validation tests cover the unfilled/header contract. ########## pinot-common/src/main/java/org/apache/pinot/common/utils/VariantUtils.java: ########## @@ -0,0 +1,2060 @@ +/** + * 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.pinot.common.utils; + +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.JsonToken; +import java.io.IOException; +import java.math.BigDecimal; +import java.math.BigInteger; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.sql.Timestamp; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.time.ZoneOffset; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Locale; +import java.util.Objects; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import javax.annotation.Nullable; +import org.apache.calcite.sql.type.SqlTypeName; +import org.apache.commons.io.output.StringBuilderWriter; +import org.apache.parquet.variant.Variant; +import org.apache.parquet.variant.VariantArrayBuilder; +import org.apache.parquet.variant.VariantBuilder; +import org.apache.parquet.variant.VariantObjectBuilder; +import org.apache.pinot.spi.data.FieldSpec.DataType; +import org.apache.pinot.spi.utils.ByteArray; +import org.apache.pinot.spi.utils.UuidUtils; +import org.apache.pinot.spi.utils.VariantEnvelope; + + +/// Query-side operations for Pinot {@code VARIANT} values. +/// +/// <p>The utility navigates the Parquet Variant binary representation directly. It never materializes a JSON tree. +/// Instances are not required, and stateless convenience methods are thread-safe. Overloads that accept a +/// caller-provided {@link ReusableResult} require that result to be thread-confined and not shared by concurrent calls. +/// An empty byte array is Pinot's SQL-null placeholder and is never decoded as an envelope. +public final class VariantUtils { + public static final String RAW_VARIANT_REQUIRES_NULL_HANDLING_ERROR = + "Raw VARIANT projection requires query null handling to be enabled; set enableNullHandling=true"; + + private static final JsonFactory JSON_FACTORY = new JsonFactory(); + private static final BigDecimal MIN_INT_DECIMAL = BigDecimal.valueOf(Integer.MIN_VALUE); + private static final BigDecimal MAX_INT_DECIMAL = BigDecimal.valueOf(Integer.MAX_VALUE); + private static final BigDecimal MIN_LONG_DECIMAL = BigDecimal.valueOf(Long.MIN_VALUE); + private static final BigDecimal MAX_LONG_DECIMAL = BigDecimal.valueOf(Long.MAX_VALUE); + private static final int MAX_JSON_NESTING_DEPTH = 100; + private static final int MAX_VARIANT_DECIMAL_PRECISION = 38; + private static final int MAX_VARIANT_DECIMAL_SCALE = 38; + private static final int MAX_VARIANT_DECIMAL_BYTES = 16; + private static final long MICROS_PER_SECOND = TimeUnit.SECONDS.toMicros(1); + private static final long NANOS_PER_MICRO = TimeUnit.MICROSECONDS.toNanos(1); + private static final long NANOS_PER_DAY = TimeUnit.DAYS.toNanos(1); + private static final int VARIANT_BASIC_TYPE_MASK = 0x03; + private static final int VARIANT_PRIMITIVE_TYPE_MASK = 0x3F; + private static final int VARIANT_PRIMITIVE = 0; + private static final int VARIANT_SHORT_STRING = 1; + private static final int VARIANT_OBJECT = 2; + private static final int VARIANT_ARRAY = 3; + private static final int VARIANT_NULL = 0; + private static final int VARIANT_TRUE = 1; + private static final int VARIANT_FALSE = 2; + private static final int VARIANT_INT8 = 3; + private static final int VARIANT_INT16 = 4; + private static final int VARIANT_INT32 = 5; + private static final int VARIANT_INT64 = 6; + private static final int VARIANT_DOUBLE = 7; + private static final int VARIANT_DECIMAL4 = 8; + private static final int VARIANT_DECIMAL8 = 9; + private static final int VARIANT_DECIMAL16 = 10; + private static final int VARIANT_DATE = 11; + private static final int VARIANT_TIMESTAMP_TZ = 12; + private static final int VARIANT_TIMESTAMP_NTZ = 13; + private static final int VARIANT_FLOAT = 14; + private static final int VARIANT_BINARY = 15; + private static final int VARIANT_LONG_STRING = 16; + private static final int VARIANT_TIME = 17; + private static final int VARIANT_TIMESTAMP_NANOS_TZ = 18; + private static final int VARIANT_TIMESTAMP_NANOS_NTZ = 19; + private static final int VARIANT_UUID = 20; + private static final int VARIANT_METADATA_VERSION_MASK = 0x0F; + private static final int VARIANT_METADATA_VERSION = 1; + private static final int OBJECT_BINARY_SEARCH_THRESHOLD = 32; + private static final int INVALID_UTF8_COMPARISON = Integer.MIN_VALUE; + private static final VariantPath ROOT_PATH = new VariantPath(new PathElement[0]); + + private VariantUtils() { + } + + /// Returns whether a final result containing raw VARIANT values requires query null handling. Without a null bitmap, + /// Pinot's reserved empty-byte SQL-null placeholder cannot be distinguished from a logical Variant value. + public static boolean requiresNullHandlingForRawVariantResult(DataSchema resultSchema, + boolean nullHandlingEnabled) { + if (nullHandlingEnabled) { + return false; + } + for (DataSchema.ColumnDataType dataType : resultSchema.getColumnDataTypes()) { + if (dataType == DataSchema.ColumnDataType.VARIANT) { + return true; + } + } + return false; + } + + /// Statically supported result types for {@code variantGet} and {@code tryVariantGet}. + public enum ResultType { + BOOLEAN(DataType.BOOLEAN, SqlTypeName.BOOLEAN), + INT(DataType.INT, SqlTypeName.INTEGER), + LONG(DataType.LONG, SqlTypeName.BIGINT), + FLOAT(DataType.FLOAT, SqlTypeName.REAL), + DOUBLE(DataType.DOUBLE, SqlTypeName.DOUBLE), + BIG_DECIMAL(DataType.BIG_DECIMAL, SqlTypeName.DECIMAL), + STRING(DataType.STRING, SqlTypeName.VARCHAR), + BYTES(DataType.BYTES, SqlTypeName.VARBINARY), + UUID(DataType.UUID, SqlTypeName.UUID), + TIMESTAMP(DataType.TIMESTAMP, SqlTypeName.TIMESTAMP), + VARIANT(DataType.VARIANT, SqlTypeName.VARIANT), + JSON(DataType.JSON, SqlTypeName.VARCHAR); + + private final DataType _dataType; + private final SqlTypeName _sqlTypeName; + + ResultType(DataType dataType, SqlTypeName sqlTypeName) { + _dataType = dataType; + _sqlTypeName = sqlTypeName; + } + + public DataType getDataType() { + return _dataType; + } + + public SqlTypeName getSqlTypeName() { + return _sqlTypeName; + } + } + + /// An immutable, pre-parsed Variant path. The v1 grammar supports {@code $}, dot-separated object fields, and + /// non-negative array subscripts. + public static final class VariantPath { + private final PathElement[] _elements; + + private VariantPath(PathElement[] elements) { + _elements = elements; + } + } + + /// Reusable, unboxed destination for vectorized Variant extraction. + /// + /// <p>Only the getter corresponding to the requested {@link ResultType} is defined after a successful extraction. + /// The instance is mutable and not thread-safe; callers should retain one per transform-function instance. Every + /// extraction may replace its state. Each successful byte-valued extraction installs a newly materialized array. + /// Values returned as {@code byte[]} or as a {@link ByteArray} may be retained after this result is reused, but they + /// are read-only by contract and must be copied before mutation. + public static final class ReusableResult { + private final Cursor _cursor = new Cursor(); + private int _intValue; + private long _longValue; + private float _floatValue; + private double _doubleValue; + private BigDecimal _bigDecimalValue; + private String _stringValue; + private byte[] _bytesValue; + + public int getIntValue() { + return _intValue; + } + + public long getLongValue() { + return _longValue; + } + + public float getFloatValue() { + return _floatValue; + } + + public double getDoubleValue() { + return _doubleValue; + } + + public BigDecimal getBigDecimalValue() { + return _bigDecimalValue; + } + + public String getStringValue() { + return _stringValue; + } + + /// Returns the extracted BYTES, VARIANT, or direct 16-byte UUID representation. + /// + /// <p>The returned array is replaced, but not mutated, by the next byte-valued extraction. It may be retained after + /// this result is reused, but must be treated as immutable and copied before mutation. + public byte[] getBytesValue() { + return _bytesValue; + } + + public UUID getUuidValue() { + return UuidUtils.toUUID(_bytesValue); + } + + /// Materializes the extracted value in the external representation used by scalar functions and ingestion. + /// + /// <p>For BYTES and VARIANT, the returned {@code byte[]} may be retained after this result is reused. It must be + /// treated as immutable and copied before mutation. + public Object toExternalValue(ResultType resultType) { + switch (resultType) { + case BOOLEAN: + return _intValue != 0; + case INT: + return _intValue; + case LONG: + return _longValue; + case FLOAT: + return _floatValue; + case DOUBLE: + return _doubleValue; + case BIG_DECIMAL: + return _bigDecimalValue; + case STRING: + case JSON: + return _stringValue; + case BYTES: + case VARIANT: + return _bytesValue; + case UUID: + return UuidUtils.toUUID(_bytesValue); + case TIMESTAMP: + return new Timestamp(_longValue); + default: + throw new IllegalStateException("Unhandled Variant target type: " + resultType); + } + } + + /// Materializes the extracted value in {@link DataSchema}'s internal representation. + /// + /// <p>TIMESTAMP remains epoch milliseconds and UUID wraps the directly copied 16-byte value, avoiding an + /// external-object round trip in the multi-stage engine. For BYTES, UUID, and VARIANT, the returned + /// {@link ByteArray} wraps a newly materialized array that may be retained after this result is reused. Neither the + /// wrapper nor its array may be mutated; callers must copy the array before mutation. + public Object toInternalValue(ResultType resultType) { + switch (resultType) { + case BOOLEAN: + return _intValue; + case INT: + return _intValue; + case LONG: + case TIMESTAMP: + return _longValue; + case FLOAT: + return _floatValue; + case DOUBLE: + return _doubleValue; + case BIG_DECIMAL: + return _bigDecimalValue; + case STRING: + case JSON: + return _stringValue; + case BYTES: + case UUID: + case VARIANT: + return new ByteArray(_bytesValue); + default: + throw new IllegalStateException("Unhandled Variant target type: " + resultType); + } + } + } + + /// Parses a target type literal once for reuse by a transform function. + public static ResultType parseResultType(String targetType) { + if (targetType == null) { + throw new IllegalArgumentException("Variant target type must not be null"); + } + try { + return ResultType.valueOf(targetType.trim().toUpperCase(Locale.ROOT)); + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException("Unsupported Variant target type: " + targetType, e); + } + } + + /// Compiles a v1 Variant path. + public static VariantPath compilePath(String path) { + if (path == null || path.isEmpty() || path.charAt(0) != '$') { + throw new IllegalArgumentException("Variant path must start with '$': " + path); + } + List<PathElement> elements = new ArrayList<>(); + int index = 1; + while (index < path.length()) { + char current = path.charAt(index); + if (current == '.') { + int fieldStart = ++index; + while (index < path.length()) { + char next = path.charAt(index); + if (next == '.' || next == '[') { + break; + } + index++; + } + if (fieldStart == index) { + throw new IllegalArgumentException("Variant path contains an empty field: " + path); + } + elements.add(PathElement.forField(path.substring(fieldStart, index))); + } else if (current == '[') { + int subscriptStart = ++index; + while (index < path.length() && Character.isDigit(path.charAt(index))) { + index++; + } + if (subscriptStart == index || index >= path.length() || path.charAt(index) != ']') { + throw new IllegalArgumentException("Invalid Variant array subscript in path: " + path); + } + try { + elements.add(PathElement.forIndex(Integer.parseInt(path.substring(subscriptStart, index)))); + } catch (NumberFormatException e) { + throw new IllegalArgumentException("Variant array subscript is too large in path: " + path, e); + } + index++; + } else { + throw new IllegalArgumentException("Unexpected character at offset " + index + " in Variant path: " + path); + } + } + return new VariantPath(elements.toArray(new PathElement[0])); + } + + /// Extracts a Variant value. A missing path or SQL null returns Java null; a Variant null remains an encoded Variant + /// value. + @Nullable + public static byte[] variantGet(@Nullable byte[] envelope, String path) { + return (byte[]) variantGet(envelope, compilePath(path), ResultType.VARIANT); + } + + /// Strictly extracts and converts a value. A missing path or SQL null returns Java null. A Variant null remains + /// encoded when the target type is {@link ResultType#VARIANT}, and returns Java null for other target types. An + /// incompatible non-null value throws. + @Nullable + public static Object variantGet(@Nullable byte[] envelope, String path, String targetType) { + return variantGet(envelope, compilePath(path), parseResultType(targetType)); + } + + /// Strictly extracts using pre-parsed path and type values. + @Nullable + public static Object variantGet(@Nullable byte[] envelope, VariantPath path, ResultType targetType) { + ReusableResult result = new ReusableResult(); + return extractInto(envelope, path, targetType, result) ? result.toExternalValue(targetType) : null; + } + + /// Strictly extracts into a reusable, unboxed result. + /// + /// @return {@code false} for SQL null, a missing path, or Variant null converted to a non-Variant target + public static boolean extractInto(@Nullable byte[] envelope, VariantPath path, ResultType targetType, + ReusableResult result) { + Objects.requireNonNull(result, "result must not be null"); + if (isSqlNull(envelope)) { + return false; + } + Objects.requireNonNull(path, "path must not be null"); + Objects.requireNonNull(targetType, "targetType must not be null"); + Cursor cursor = result._cursor; + if (!cursor.navigate(envelope, path)) { + return false; + } + if (cursor.getType() == Variant.Type.NULL && targetType != ResultType.VARIANT) { + return false; + } + convert(cursor, targetType, result); + return true; + } + + /// Tolerant Variant extraction. Malformed input returns Java null. + @Nullable + public static byte[] tryVariantGet(@Nullable byte[] envelope, String path) { + return (byte[]) tryVariantGet(envelope, compilePath(path), ResultType.VARIANT); + } + + /// Tolerant typed extraction. Malformed input and incompatible types return Java null. + @Nullable + public static Object tryVariantGet(@Nullable byte[] envelope, String path, String targetType) { + try { + return tryVariantGet(envelope, compilePath(path), parseResultType(targetType)); + } catch (RuntimeException e) { + return null; + } + } + + /// Tolerant extraction using pre-parsed path and type values. + @Nullable + public static Object tryVariantGet(@Nullable byte[] envelope, VariantPath path, ResultType targetType) { + try { + ReusableResult result = new ReusableResult(); + return tryExtractInto(envelope, path, targetType, result) ? result.toExternalValue(targetType) : null; + } catch (RuntimeException e) { + return null; + } + } + + /// Tolerantly extracts into a reusable, unboxed result. + /// + /// @return {@code false} for SQL null, missing paths, Variant null converted to a non-Variant target, + /// malformed input, or an incompatible conversion + public static boolean tryExtractInto(@Nullable byte[] envelope, VariantPath path, ResultType targetType, + ReusableResult result) { + Objects.requireNonNull(result, "result must not be null"); + if (isSqlNull(envelope)) { + return false; + } + Objects.requireNonNull(path, "path must not be null"); + Objects.requireNonNull(targetType, "targetType must not be null"); + Cursor cursor = result._cursor; + try { + if (!cursor.navigate(envelope, path)) { + return false; + } + if (cursor.getType() == Variant.Type.NULL && targetType != ResultType.VARIANT) { + return false; + } + return tryConvert(cursor, targetType, result); + } catch (IllegalArgumentException | IllegalStateException | UnsupportedOperationException + | IndexOutOfBoundsException e) { + // Cursor operations use these exceptions only for malformed or unsupported Variant encodings. + return false; + } + } + + /// Returns whether the path is present. A present Variant null counts as present. + @Nullable + public static Boolean variantExists(@Nullable byte[] envelope, String path) { + return variantExists(envelope, compilePath(path)); + } + + /// Returns whether a compiled path is present. A present Variant null counts as present. + @Nullable + public static Boolean variantExists(@Nullable byte[] envelope, VariantPath path) { + return variantExists(envelope, path, new ReusableResult()); + } + + /// Allocation-free compiled-path form of {@link #variantExists(byte[], VariantPath)} when the caller retains the + /// supplied result between rows. + @Nullable + public static Boolean variantExists(@Nullable byte[] envelope, VariantPath path, ReusableResult result) { + Objects.requireNonNull(result, "result must not be null"); + if (isSqlNull(envelope)) { + return null; + } + return result._cursor.navigate(envelope, Objects.requireNonNull(path, "path must not be null")); + } + + /// Returns whether the root value is a Variant null. SQL null is not a Variant null. + public static boolean isVariantNull(@Nullable byte[] envelope) { + return isVariantNull(envelope, ROOT_PATH, new ReusableResult()); + } + + /// Returns whether a present value at the path is a Variant null. SQL null and missing paths return false. + public static boolean isVariantNull(@Nullable byte[] envelope, String path) { + return isVariantNull(envelope, compilePath(path)); + } + + /// Returns whether a present value at a compiled path is a Variant null. SQL null and missing paths return false. + public static boolean isVariantNull(@Nullable byte[] envelope, VariantPath path) { + return isVariantNull(envelope, path, new ReusableResult()); + } + + /// Allocation-free compiled-path form of {@link #isVariantNull(byte[], VariantPath)} when the caller retains the + /// supplied result between rows. + public static boolean isVariantNull(@Nullable byte[] envelope, VariantPath path, ReusableResult result) { + Objects.requireNonNull(result, "result must not be null"); + if (isSqlNull(envelope)) { + return false; + } + Cursor cursor = result._cursor; + return cursor.navigate(envelope, Objects.requireNonNull(path, "path must not be null")) + && cursor.getType() == Variant.Type.NULL; + } + + /// Returns the Variant type name at the root, or Java null for SQL null. + @Nullable + public static String variantTypeOf(@Nullable byte[] envelope) { + return variantTypeOf(envelope, ROOT_PATH, new ReusableResult()); + } + + /// Returns the Variant type name at a path, or Java null for SQL null or a missing path. + @Nullable + public static String variantTypeOf(@Nullable byte[] envelope, String path) { + return variantTypeOf(envelope, compilePath(path)); + } + + /// Returns the Variant type name at a compiled path, or Java null for SQL null or a missing path. + @Nullable + public static String variantTypeOf(@Nullable byte[] envelope, VariantPath path) { + return variantTypeOf(envelope, path, new ReusableResult()); + } + + /// Allocation-free compiled-path form of {@link #variantTypeOf(byte[], VariantPath)} when the caller retains the + /// supplied result between rows. + @Nullable + public static String variantTypeOf(@Nullable byte[] envelope, VariantPath path, ReusableResult result) { + Objects.requireNonNull(result, "result must not be null"); + if (isSqlNull(envelope)) { + return null; + } + Cursor cursor = result._cursor; + return cursor.navigate(envelope, Objects.requireNonNull(path, "path must not be null")) + ? typeName(cursor.getType()) : null; + } + + /// Renders the Variant value as canonical JSON text without constructing a JSON tree. + @Nullable + public static String variantToJson(@Nullable byte[] envelope) { + return variantToJson(envelope, new ReusableResult()); + } + + /// Allocation-reduced form of [#variantToJson(byte[])] when the caller retains the supplied result between rows. + @Nullable + public static String variantToJson(@Nullable byte[] envelope, ReusableResult result) { + Objects.requireNonNull(result, "result must not be null"); + if (isSqlNull(envelope)) { + return null; + } + Cursor cursor = result._cursor; + cursor.navigate(envelope, ROOT_PATH); + return variantToJson(cursor.asVariant()); + } + + /// Parses JSON text into a Pinot Variant envelope without constructing a JSON tree. + @Nullable + public static byte[] parseJsonToVariant(@Nullable String json) { + if (json == null) { + return null; + } + try (JsonParser parser = JSON_FACTORY.createParser(json)) { + JsonToken token = parser.nextToken(); + if (token == null) { + throw new IllegalArgumentException("Cannot parse empty text as Variant"); + } + VariantBuilder builder = new VariantBuilder(); + appendJsonValue(parser, token, builder, 0); + if (parser.nextToken() != null) { + throw new IllegalArgumentException("Unexpected trailing token after Variant JSON value"); + } + Variant variant = builder.build(); + return VariantEnvelope.encode(variant.getMetadataBuffer(), variant.getValueBuffer()); + } catch (IOException | RuntimeException e) { + throw new IllegalArgumentException("Cannot parse JSON as Variant", e); + } + } + + /// Tolerant JSON parser. Malformed or unsupported input returns Java null. + @Nullable + public static byte[] tryParseJsonToVariant(@Nullable String json) { + try { + return parseJsonToVariant(json); + } catch (RuntimeException e) { + return null; + } + } + + private static boolean isSqlNull(@Nullable byte[] envelope) { + return envelope == null || envelope.length == 0; + } + + private static void convert(Cursor value, ResultType targetType, ReusableResult result) { + switch (targetType) { + case BOOLEAN: + requireType(value, Variant.Type.BOOLEAN, targetType); + result._intValue = value.getBoolean() ? 1 : 0; + break; + case INT: + result._intValue = toInt(value, targetType); + break; + case LONG: + result._longValue = toLong(value, targetType); + break; + case FLOAT: + result._floatValue = toFloat(value, targetType); + break; + case DOUBLE: + result._doubleValue = toDouble(value, targetType); + break; + case BIG_DECIMAL: + result._bigDecimalValue = toBigDecimal(value, targetType); + break; + case STRING: + requireType(value, Variant.Type.STRING, targetType); + result._stringValue = value.getString(); + break; + case BYTES: + requireType(value, Variant.Type.BINARY, targetType); + result._bytesValue = value.getBinary(); + break; + case UUID: + requireType(value, Variant.Type.UUID, targetType); + result._bytesValue = value.getUuidBytes(); + break; + case TIMESTAMP: + result._longValue = toTimestampMillis(value, targetType); + break; + case VARIANT: + result._bytesValue = value.copyEnvelope(); + break; + case JSON: + result._stringValue = variantToJson(value.asVariant()); + break; + default: + throw new IllegalStateException("Unhandled Variant target type: " + targetType); + } + } + + private static boolean tryConvert(Cursor value, ResultType targetType, ReusableResult result) { + Variant.Type valueType = value.getType(); + switch (targetType) { + case BOOLEAN: + if (valueType != Variant.Type.BOOLEAN) { + return false; + } + result._intValue = value.getBoolean() ? 1 : 0; + return true; + case INT: + return tryConvertToInt(value, valueType, result); + case LONG: + return tryConvertToLong(value, valueType, result); + case FLOAT: + switch (valueType) { + case BYTE: + case SHORT: + case INT: + case LONG: + result._floatValue = value.getInteger(); + return true; + case FLOAT: + result._floatValue = value.getFloat(); + return true; + case DOUBLE: + result._floatValue = (float) value.getDouble(); + return true; + case DECIMAL4: + case DECIMAL8: + case DECIMAL16: + result._floatValue = value.getDecimal().floatValue(); + return true; + default: + return false; + } + case DOUBLE: + switch (valueType) { + case BYTE: + case SHORT: + case INT: + case LONG: + result._doubleValue = value.getInteger(); + return true; + case FLOAT: + result._doubleValue = value.getFloat(); + return true; + case DOUBLE: + result._doubleValue = value.getDouble(); + return true; + case DECIMAL4: + case DECIMAL8: + case DECIMAL16: + result._doubleValue = value.getDecimal().doubleValue(); + return true; + default: + return false; + } + case BIG_DECIMAL: + switch (valueType) { + case BYTE: + case SHORT: + case INT: + case LONG: + result._bigDecimalValue = BigDecimal.valueOf(value.getInteger()); + return true; + case FLOAT: + float floatValue = value.getFloat(); + if (!Float.isFinite(floatValue)) { + return false; + } + result._bigDecimalValue = BigDecimal.valueOf(floatValue); + return true; + case DOUBLE: + double doubleValue = value.getDouble(); + if (!Double.isFinite(doubleValue)) { + return false; + } + result._bigDecimalValue = BigDecimal.valueOf(doubleValue); + return true; + case DECIMAL4: + case DECIMAL8: + case DECIMAL16: + result._bigDecimalValue = value.getDecimal(); + return true; + default: + return false; + } + case STRING: + if (valueType != Variant.Type.STRING) { + return false; + } + result._stringValue = value.getString(); + return true; + case BYTES: + if (valueType != Variant.Type.BINARY) { + return false; + } + result._bytesValue = value.getBinary(); + return true; + case UUID: + if (valueType != Variant.Type.UUID) { + return false; + } + result._bytesValue = value.getUuidBytes(); + return true; + case TIMESTAMP: + switch (valueType) { + case DATE: + result._longValue = value.getInteger() * TimeUnit.DAYS.toMillis(1); + return true; + case TIMESTAMP_TZ: + case TIMESTAMP_NTZ: + result._longValue = Math.floorDiv(value.getInteger(), TimeUnit.MILLISECONDS.toMicros(1)); + return true; + case TIMESTAMP_NANOS_TZ: + case TIMESTAMP_NANOS_NTZ: + result._longValue = Math.floorDiv(value.getInteger(), TimeUnit.MILLISECONDS.toNanos(1)); + return true; + default: + return false; + } + case VARIANT: + result._bytesValue = value.copyEnvelope(); + return true; + case JSON: + result._stringValue = variantToJson(value.asVariant()); + return true; + default: + throw new AssertionError("Unhandled Variant target type: " + targetType); + } + } + + private static boolean tryConvertToInt(Cursor value, Variant.Type valueType, ReusableResult result) { + switch (valueType) { + case BYTE: + case SHORT: + case INT: + result._intValue = (int) value.getInteger(); + return true; + case LONG: + long longValue = value.getInteger(); + if (longValue < Integer.MIN_VALUE || longValue > Integer.MAX_VALUE) { + return false; + } + result._intValue = (int) longValue; + return true; + case DECIMAL4: + case DECIMAL8: + case DECIMAL16: + BigDecimal decimalValue = value.getDecimal(); + if (!isIntegralInRange(decimalValue, MIN_INT_DECIMAL, MAX_INT_DECIMAL)) { + return false; + } + result._intValue = decimalValue.intValue(); + return true; + default: + return false; + } + } + + private static boolean tryConvertToLong(Cursor value, Variant.Type valueType, ReusableResult result) { + switch (valueType) { + case BYTE: + case SHORT: + case INT: + case LONG: + result._longValue = value.getInteger(); + return true; + case DECIMAL4: + case DECIMAL8: + case DECIMAL16: + BigDecimal decimalValue = value.getDecimal(); + if (!isIntegralInRange(decimalValue, MIN_LONG_DECIMAL, MAX_LONG_DECIMAL)) { + return false; + } + result._longValue = decimalValue.longValue(); + return true; + default: + return false; + } + } + + private static boolean isIntegralInRange(BigDecimal value, BigDecimal minimum, BigDecimal maximum) { + return value.compareTo(minimum) >= 0 && value.compareTo(maximum) <= 0 + && (value.scale() <= 0 || value.stripTrailingZeros().scale() <= 0); + } + + private static int toInt(Cursor value, ResultType targetType) { + switch (value.getType()) { + case BYTE: + case SHORT: + case INT: + return (int) value.getInteger(); + case LONG: + return Math.toIntExact(value.getInteger()); + case DECIMAL4: + case DECIMAL8: + case DECIMAL16: + return value.getDecimal().intValueExact(); + default: + throw typeMismatch(value, targetType); + } + } + + private static long toLong(Cursor value, ResultType targetType) { + switch (value.getType()) { + case BYTE: + case SHORT: + case INT: + case LONG: + return value.getInteger(); + case DECIMAL4: + case DECIMAL8: + case DECIMAL16: + return value.getDecimal().longValueExact(); + default: + throw typeMismatch(value, targetType); + } + } + + private static float toFloat(Cursor value, ResultType targetType) { + switch (value.getType()) { + case BYTE: + case SHORT: + case INT: + case LONG: + return value.getInteger(); + case FLOAT: + return value.getFloat(); + case DOUBLE: + return (float) value.getDouble(); + case DECIMAL4: + case DECIMAL8: + case DECIMAL16: + return value.getDecimal().floatValue(); + default: + throw typeMismatch(value, targetType); + } + } + + private static double toDouble(Cursor value, ResultType targetType) { + switch (value.getType()) { + case BYTE: + case SHORT: + case INT: + case LONG: + return value.getInteger(); + case FLOAT: + return value.getFloat(); + case DOUBLE: + return value.getDouble(); + case DECIMAL4: + case DECIMAL8: + case DECIMAL16: + return value.getDecimal().doubleValue(); + default: + throw typeMismatch(value, targetType); + } + } + + private static BigDecimal toBigDecimal(Cursor value, ResultType targetType) { + switch (value.getType()) { + case BYTE: + case SHORT: + case INT: + case LONG: + return BigDecimal.valueOf(value.getInteger()); + case FLOAT: + return BigDecimal.valueOf(value.getFloat()); + case DOUBLE: + return BigDecimal.valueOf(value.getDouble()); + case DECIMAL4: + case DECIMAL8: + case DECIMAL16: + return value.getDecimal(); + default: + throw typeMismatch(value, targetType); + } + } + + private static long toTimestampMillis(Cursor value, ResultType targetType) { + switch (value.getType()) { + case DATE: + return Math.multiplyExact(value.getInteger(), TimeUnit.DAYS.toMillis(1)); + case TIMESTAMP_TZ: + case TIMESTAMP_NTZ: + return Math.floorDiv(value.getInteger(), TimeUnit.MILLISECONDS.toMicros(1)); + case TIMESTAMP_NANOS_TZ: + case TIMESTAMP_NANOS_NTZ: + return Math.floorDiv(value.getInteger(), TimeUnit.MILLISECONDS.toNanos(1)); + default: + throw typeMismatch(value, targetType); + } + } + + private static void requireType(Cursor value, Variant.Type expected, ResultType targetType) { + if (value.getType() != expected) { + throw typeMismatch(value, targetType); + } + } + + private static IllegalArgumentException typeMismatch(Cursor value, ResultType targetType) { + return new IllegalArgumentException( + "Cannot convert Variant " + typeName(value.getType()) + " to " + targetType.name()); + } + + private static String typeName(Variant.Type type) { + switch (type) { + case DECIMAL4: + case DECIMAL8: + case DECIMAL16: + return "DECIMAL"; + default: + return type.name(); + } + } + + private static String variantToJson(Variant variant) { + try { + // StringBuilder-backed writer: StringWriter wraps a synchronized StringBuffer and would pay a monitor + // acquisition per append on this per-row rendering path. + StringBuilderWriter writer = new StringBuilderWriter(); + try (JsonGenerator generator = JSON_FACTORY.createGenerator(writer)) { + writeJsonValue(generator, variant); + } + return writer.toString(); + } catch (IOException e) { + throw new IllegalStateException("Cannot render Variant as JSON", e); + } + } + + + private static void writeJsonValue(JsonGenerator generator, Variant variant) + throws IOException { + switch (variant.getType()) { + case OBJECT: + generator.writeStartObject(); + for (int i = 0; i < variant.numObjectElements(); i++) { + Variant.ObjectField field = variant.getFieldAtIndex(i); + generator.writeFieldName(field.key); + writeJsonValue(generator, field.value); + } + generator.writeEndObject(); + break; + case ARRAY: + generator.writeStartArray(); + for (int i = 0; i < variant.numArrayElements(); i++) { + writeJsonValue(generator, variant.getElementAtIndex(i)); + } + generator.writeEndArray(); + break; + case NULL: + generator.writeNull(); + break; + case BOOLEAN: + generator.writeBoolean(variant.getBoolean()); + break; + case BYTE: + generator.writeNumber(variant.getByte()); + break; + case SHORT: + generator.writeNumber(variant.getShort()); + break; + case INT: + generator.writeNumber(variant.getInt()); + break; + case LONG: + generator.writeNumber(variant.getLong()); + break; + case FLOAT: + generator.writeNumber(variant.getFloat()); + break; + case DOUBLE: + generator.writeNumber(variant.getDouble()); + break; + case DECIMAL4: + case DECIMAL8: + case DECIMAL16: + generator.writeNumber(variant.getDecimal()); + break; + case STRING: + generator.writeString(variant.getString()); + break; + case BINARY: + generator.writeBinary(toBytes(variant.getBinary())); + break; + case UUID: + generator.writeString(variant.getUUID().toString()); + break; + case DATE: + generator.writeString(LocalDate.ofEpochDay(variant.getInt()).toString()); + break; + case TIMESTAMP_TZ: + generator.writeString(instantFromMicros(variant.getLong()).toString()); + break; + case TIMESTAMP_NTZ: + generator.writeString(LocalDateTime.ofInstant(instantFromMicros(variant.getLong()), ZoneOffset.UTC).toString()); + break; + case TIMESTAMP_NANOS_TZ: + generator.writeString(instantFromNanos(variant.getLong()).toString()); + break; + case TIMESTAMP_NANOS_NTZ: + generator.writeString(LocalDateTime.ofInstant(instantFromNanos(variant.getLong()), ZoneOffset.UTC).toString()); + break; + case TIME: + generator.writeString(LocalTime.ofNanoOfDay(Math.floorMod(variant.getLong() * NANOS_PER_MICRO, NANOS_PER_DAY)) + .toString()); + break; + default: + throw new IllegalStateException("Unsupported Variant type: " + variant.getType()); + } + } + + private static Instant instantFromMicros(long micros) { + long seconds = Math.floorDiv(micros, MICROS_PER_SECOND); + long nanos = Math.floorMod(micros, MICROS_PER_SECOND) * NANOS_PER_MICRO; + return Instant.ofEpochSecond(seconds, nanos); + } + + private static Instant instantFromNanos(long nanos) { + return Instant.ofEpochSecond(Math.floorDiv(nanos, TimeUnit.SECONDS.toNanos(1)), + Math.floorMod(nanos, TimeUnit.SECONDS.toNanos(1))); + } + + private static byte[] toBytes(ByteBuffer buffer) { + ByteBuffer view = buffer.slice(); + byte[] bytes = new byte[view.remaining()]; + view.get(bytes); + return bytes; + } + + private static void appendJsonValue(JsonParser parser, JsonToken token, VariantBuilder builder, int depth) + throws IOException { + if (depth > MAX_JSON_NESTING_DEPTH) { + throw new IllegalArgumentException("Variant JSON exceeds maximum nesting depth " + MAX_JSON_NESTING_DEPTH); + } + switch (token) { + case START_OBJECT: + VariantObjectBuilder objectBuilder = builder.startObject(); + while (parser.nextToken() != JsonToken.END_OBJECT) { + if (parser.currentToken() != JsonToken.FIELD_NAME) { + throw new IllegalArgumentException("Expected a JSON object field name"); + } + objectBuilder.appendKey(parser.currentName()); + JsonToken fieldValue = parser.nextToken(); + if (fieldValue == null) { + throw new IllegalArgumentException("Unexpected end of JSON object"); + } + appendJsonValue(parser, fieldValue, objectBuilder, depth + 1); + } + builder.endObject(); + break; + case START_ARRAY: + VariantArrayBuilder arrayBuilder = builder.startArray(); + while (true) { + JsonToken element = parser.nextToken(); + if (element == JsonToken.END_ARRAY) { + break; + } + if (element == null) { + throw new IllegalArgumentException("Unexpected end of JSON array"); + } + appendJsonValue(parser, element, arrayBuilder, depth + 1); + } + builder.endArray(); + break; + case VALUE_NULL: + builder.appendNull(); + break; + case VALUE_TRUE: + builder.appendBoolean(true); + break; + case VALUE_FALSE: + builder.appendBoolean(false); + break; + case VALUE_STRING: + builder.appendString(parser.getText()); + break; + case VALUE_NUMBER_INT: + appendInteger(parser, builder); + break; + case VALUE_NUMBER_FLOAT: + appendDecimal(parser.getDecimalValue(), builder); + break; + default: + throw new IllegalArgumentException("Unsupported JSON token for Variant: " + token); + } + } + + private static void appendInteger(JsonParser parser, VariantBuilder builder) + throws IOException { + switch (parser.getNumberType()) { + case INT: + builder.appendInt(parser.getIntValue()); + break; + case LONG: + builder.appendLong(parser.getLongValue()); + break; + case BIG_INTEGER: + appendBigInteger(parser.getBigIntegerValue(), builder); + break; + default: + throw new IllegalArgumentException("Unsupported JSON integer representation: " + parser.getNumberType()); + } + } + + private static void appendBigInteger(BigInteger value, VariantBuilder builder) { + if (value.bitLength() < Integer.SIZE) { + builder.appendInt(value.intValue()); + } else if (value.bitLength() < Long.SIZE) { + builder.appendLong(value.longValue()); + } else { + appendDecimal(new BigDecimal(value), builder); + } + } + + private static void appendDecimal(BigDecimal value, VariantBuilder builder) { + BigDecimal normalized = value; + if (normalized.scale() < 0) { + // Parquet Variant stores scale as an unsigned byte. Expand exponent notation exactly instead of allowing a + // negative scale to wrap during encoding. + long expandedPrecision = (long) normalized.precision() - normalized.scale(); + if (normalized.signum() != 0 && expandedPrecision > MAX_VARIANT_DECIMAL_PRECISION) { + throw unsupportedVariantDecimal(value); + } + normalized = normalized.signum() == 0 ? BigDecimal.ZERO : normalized.setScale(0); + } else if (normalized.scale() > MAX_VARIANT_DECIMAL_SCALE) { + // Accept values whose excessive lexical scale consists only of insignificant trailing zeros. + normalized = normalized.stripTrailingZeros(); + if (normalized.scale() < 0) { + normalized = normalized.setScale(0); + } + } + byte[] unscaledBytes = normalized.unscaledValue().toByteArray(); + if (normalized.scale() > MAX_VARIANT_DECIMAL_SCALE + || normalized.precision() > MAX_VARIANT_DECIMAL_PRECISION + || unscaledBytes.length > MAX_VARIANT_DECIMAL_BYTES) { + throw unsupportedVariantDecimal(value); + } + builder.appendDecimal(normalized); + } + + private static IllegalArgumentException unsupportedVariantDecimal(BigDecimal value) { + return new IllegalArgumentException( + "JSON decimal exceeds Parquet Variant decimal(38) encoding: precision=" + value.precision() + + ", scale=" + value.scale()); + } + + /// Mutable zero-copy view over one selected value in a Pinot envelope. + /// + /// <p>The constants and layouts used here mirror Parquet Variant encoding version 1. Keeping this cursor on + /// {@link ReusableResult} avoids allocating envelope views, Variant wrappers, and navigation wrappers for every row. + private static final class Cursor { Review Comment: Added an explicit pinned sync policy in `VariantUtils` and a reflection test comparing the cursor masks/tags and every `Variant.Type` with parquet-variant 1.18.0. Unknown metadata versions and primitive tags now fail closed with actionable strict errors and tolerant nulls. I deliberately did not fall back through an old decoder for an unknown encoding because that risks silent misinterpretation. ########## pinot-common/src/main/java/org/apache/pinot/common/utils/VariantUtils.java: ########## @@ -0,0 +1,2060 @@ +/** + * 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.pinot.common.utils; + +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.JsonToken; +import java.io.IOException; +import java.math.BigDecimal; +import java.math.BigInteger; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.sql.Timestamp; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.time.ZoneOffset; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Locale; +import java.util.Objects; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import javax.annotation.Nullable; +import org.apache.calcite.sql.type.SqlTypeName; +import org.apache.commons.io.output.StringBuilderWriter; +import org.apache.parquet.variant.Variant; +import org.apache.parquet.variant.VariantArrayBuilder; +import org.apache.parquet.variant.VariantBuilder; +import org.apache.parquet.variant.VariantObjectBuilder; +import org.apache.pinot.spi.data.FieldSpec.DataType; +import org.apache.pinot.spi.utils.ByteArray; +import org.apache.pinot.spi.utils.UuidUtils; +import org.apache.pinot.spi.utils.VariantEnvelope; + + +/// Query-side operations for Pinot {@code VARIANT} values. +/// +/// <p>The utility navigates the Parquet Variant binary representation directly. It never materializes a JSON tree. +/// Instances are not required, and stateless convenience methods are thread-safe. Overloads that accept a +/// caller-provided {@link ReusableResult} require that result to be thread-confined and not shared by concurrent calls. +/// An empty byte array is Pinot's SQL-null placeholder and is never decoded as an envelope. +public final class VariantUtils { + public static final String RAW_VARIANT_REQUIRES_NULL_HANDLING_ERROR = + "Raw VARIANT projection requires query null handling to be enabled; set enableNullHandling=true"; + + private static final JsonFactory JSON_FACTORY = new JsonFactory(); + private static final BigDecimal MIN_INT_DECIMAL = BigDecimal.valueOf(Integer.MIN_VALUE); + private static final BigDecimal MAX_INT_DECIMAL = BigDecimal.valueOf(Integer.MAX_VALUE); + private static final BigDecimal MIN_LONG_DECIMAL = BigDecimal.valueOf(Long.MIN_VALUE); + private static final BigDecimal MAX_LONG_DECIMAL = BigDecimal.valueOf(Long.MAX_VALUE); + private static final int MAX_JSON_NESTING_DEPTH = 100; + private static final int MAX_VARIANT_DECIMAL_PRECISION = 38; + private static final int MAX_VARIANT_DECIMAL_SCALE = 38; + private static final int MAX_VARIANT_DECIMAL_BYTES = 16; + private static final long MICROS_PER_SECOND = TimeUnit.SECONDS.toMicros(1); + private static final long NANOS_PER_MICRO = TimeUnit.MICROSECONDS.toNanos(1); + private static final long NANOS_PER_DAY = TimeUnit.DAYS.toNanos(1); + private static final int VARIANT_BASIC_TYPE_MASK = 0x03; + private static final int VARIANT_PRIMITIVE_TYPE_MASK = 0x3F; + private static final int VARIANT_PRIMITIVE = 0; + private static final int VARIANT_SHORT_STRING = 1; + private static final int VARIANT_OBJECT = 2; + private static final int VARIANT_ARRAY = 3; + private static final int VARIANT_NULL = 0; + private static final int VARIANT_TRUE = 1; + private static final int VARIANT_FALSE = 2; + private static final int VARIANT_INT8 = 3; + private static final int VARIANT_INT16 = 4; + private static final int VARIANT_INT32 = 5; + private static final int VARIANT_INT64 = 6; + private static final int VARIANT_DOUBLE = 7; + private static final int VARIANT_DECIMAL4 = 8; + private static final int VARIANT_DECIMAL8 = 9; + private static final int VARIANT_DECIMAL16 = 10; + private static final int VARIANT_DATE = 11; + private static final int VARIANT_TIMESTAMP_TZ = 12; + private static final int VARIANT_TIMESTAMP_NTZ = 13; + private static final int VARIANT_FLOAT = 14; + private static final int VARIANT_BINARY = 15; + private static final int VARIANT_LONG_STRING = 16; + private static final int VARIANT_TIME = 17; + private static final int VARIANT_TIMESTAMP_NANOS_TZ = 18; + private static final int VARIANT_TIMESTAMP_NANOS_NTZ = 19; + private static final int VARIANT_UUID = 20; + private static final int VARIANT_METADATA_VERSION_MASK = 0x0F; + private static final int VARIANT_METADATA_VERSION = 1; + private static final int OBJECT_BINARY_SEARCH_THRESHOLD = 32; + private static final int INVALID_UTF8_COMPARISON = Integer.MIN_VALUE; + private static final VariantPath ROOT_PATH = new VariantPath(new PathElement[0]); + + private VariantUtils() { + } + + /// Returns whether a final result containing raw VARIANT values requires query null handling. Without a null bitmap, + /// Pinot's reserved empty-byte SQL-null placeholder cannot be distinguished from a logical Variant value. + public static boolean requiresNullHandlingForRawVariantResult(DataSchema resultSchema, + boolean nullHandlingEnabled) { + if (nullHandlingEnabled) { + return false; + } + for (DataSchema.ColumnDataType dataType : resultSchema.getColumnDataTypes()) { + if (dataType == DataSchema.ColumnDataType.VARIANT) { + return true; + } + } + return false; + } + + /// Statically supported result types for {@code variantGet} and {@code tryVariantGet}. + public enum ResultType { + BOOLEAN(DataType.BOOLEAN, SqlTypeName.BOOLEAN), + INT(DataType.INT, SqlTypeName.INTEGER), + LONG(DataType.LONG, SqlTypeName.BIGINT), + FLOAT(DataType.FLOAT, SqlTypeName.REAL), + DOUBLE(DataType.DOUBLE, SqlTypeName.DOUBLE), + BIG_DECIMAL(DataType.BIG_DECIMAL, SqlTypeName.DECIMAL), + STRING(DataType.STRING, SqlTypeName.VARCHAR), + BYTES(DataType.BYTES, SqlTypeName.VARBINARY), + UUID(DataType.UUID, SqlTypeName.UUID), + TIMESTAMP(DataType.TIMESTAMP, SqlTypeName.TIMESTAMP), + VARIANT(DataType.VARIANT, SqlTypeName.VARIANT), + JSON(DataType.JSON, SqlTypeName.VARCHAR); + + private final DataType _dataType; + private final SqlTypeName _sqlTypeName; + + ResultType(DataType dataType, SqlTypeName sqlTypeName) { + _dataType = dataType; + _sqlTypeName = sqlTypeName; + } + + public DataType getDataType() { + return _dataType; + } + + public SqlTypeName getSqlTypeName() { + return _sqlTypeName; + } + } + + /// An immutable, pre-parsed Variant path. The v1 grammar supports {@code $}, dot-separated object fields, and + /// non-negative array subscripts. + public static final class VariantPath { + private final PathElement[] _elements; + + private VariantPath(PathElement[] elements) { + _elements = elements; + } + } + + /// Reusable, unboxed destination for vectorized Variant extraction. + /// + /// <p>Only the getter corresponding to the requested {@link ResultType} is defined after a successful extraction. + /// The instance is mutable and not thread-safe; callers should retain one per transform-function instance. Every + /// extraction may replace its state. Each successful byte-valued extraction installs a newly materialized array. + /// Values returned as {@code byte[]} or as a {@link ByteArray} may be retained after this result is reused, but they + /// are read-only by contract and must be copied before mutation. + public static final class ReusableResult { + private final Cursor _cursor = new Cursor(); + private int _intValue; + private long _longValue; + private float _floatValue; + private double _doubleValue; + private BigDecimal _bigDecimalValue; + private String _stringValue; + private byte[] _bytesValue; + + public int getIntValue() { + return _intValue; + } + + public long getLongValue() { + return _longValue; + } + + public float getFloatValue() { + return _floatValue; + } + + public double getDoubleValue() { + return _doubleValue; + } + + public BigDecimal getBigDecimalValue() { + return _bigDecimalValue; + } + + public String getStringValue() { + return _stringValue; + } + + /// Returns the extracted BYTES, VARIANT, or direct 16-byte UUID representation. + /// + /// <p>The returned array is replaced, but not mutated, by the next byte-valued extraction. It may be retained after + /// this result is reused, but must be treated as immutable and copied before mutation. + public byte[] getBytesValue() { + return _bytesValue; + } + + public UUID getUuidValue() { + return UuidUtils.toUUID(_bytesValue); + } + + /// Materializes the extracted value in the external representation used by scalar functions and ingestion. + /// + /// <p>For BYTES and VARIANT, the returned {@code byte[]} may be retained after this result is reused. It must be + /// treated as immutable and copied before mutation. + public Object toExternalValue(ResultType resultType) { + switch (resultType) { + case BOOLEAN: + return _intValue != 0; + case INT: + return _intValue; + case LONG: + return _longValue; + case FLOAT: + return _floatValue; + case DOUBLE: + return _doubleValue; + case BIG_DECIMAL: + return _bigDecimalValue; + case STRING: + case JSON: + return _stringValue; + case BYTES: + case VARIANT: + return _bytesValue; + case UUID: + return UuidUtils.toUUID(_bytesValue); + case TIMESTAMP: + return new Timestamp(_longValue); + default: + throw new IllegalStateException("Unhandled Variant target type: " + resultType); + } + } + + /// Materializes the extracted value in {@link DataSchema}'s internal representation. + /// + /// <p>TIMESTAMP remains epoch milliseconds and UUID wraps the directly copied 16-byte value, avoiding an + /// external-object round trip in the multi-stage engine. For BYTES, UUID, and VARIANT, the returned + /// {@link ByteArray} wraps a newly materialized array that may be retained after this result is reused. Neither the + /// wrapper nor its array may be mutated; callers must copy the array before mutation. + public Object toInternalValue(ResultType resultType) { + switch (resultType) { + case BOOLEAN: + return _intValue; + case INT: + return _intValue; + case LONG: + case TIMESTAMP: + return _longValue; + case FLOAT: + return _floatValue; + case DOUBLE: + return _doubleValue; + case BIG_DECIMAL: + return _bigDecimalValue; + case STRING: + case JSON: + return _stringValue; + case BYTES: + case UUID: + case VARIANT: + return new ByteArray(_bytesValue); + default: + throw new IllegalStateException("Unhandled Variant target type: " + resultType); + } + } + } + + /// Parses a target type literal once for reuse by a transform function. + public static ResultType parseResultType(String targetType) { + if (targetType == null) { + throw new IllegalArgumentException("Variant target type must not be null"); + } + try { + return ResultType.valueOf(targetType.trim().toUpperCase(Locale.ROOT)); + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException("Unsupported Variant target type: " + targetType, e); + } + } + + /// Compiles a v1 Variant path. + public static VariantPath compilePath(String path) { + if (path == null || path.isEmpty() || path.charAt(0) != '$') { + throw new IllegalArgumentException("Variant path must start with '$': " + path); + } + List<PathElement> elements = new ArrayList<>(); + int index = 1; + while (index < path.length()) { + char current = path.charAt(index); + if (current == '.') { + int fieldStart = ++index; + while (index < path.length()) { + char next = path.charAt(index); + if (next == '.' || next == '[') { + break; + } + index++; + } + if (fieldStart == index) { + throw new IllegalArgumentException("Variant path contains an empty field: " + path); + } + elements.add(PathElement.forField(path.substring(fieldStart, index))); + } else if (current == '[') { + int subscriptStart = ++index; + while (index < path.length() && Character.isDigit(path.charAt(index))) { + index++; + } + if (subscriptStart == index || index >= path.length() || path.charAt(index) != ']') { + throw new IllegalArgumentException("Invalid Variant array subscript in path: " + path); + } + try { + elements.add(PathElement.forIndex(Integer.parseInt(path.substring(subscriptStart, index)))); + } catch (NumberFormatException e) { + throw new IllegalArgumentException("Variant array subscript is too large in path: " + path, e); + } + index++; + } else { + throw new IllegalArgumentException("Unexpected character at offset " + index + " in Variant path: " + path); + } + } + return new VariantPath(elements.toArray(new PathElement[0])); + } + + /// Extracts a Variant value. A missing path or SQL null returns Java null; a Variant null remains an encoded Variant + /// value. + @Nullable + public static byte[] variantGet(@Nullable byte[] envelope, String path) { + return (byte[]) variantGet(envelope, compilePath(path), ResultType.VARIANT); + } + + /// Strictly extracts and converts a value. A missing path or SQL null returns Java null. A Variant null remains + /// encoded when the target type is {@link ResultType#VARIANT}, and returns Java null for other target types. An + /// incompatible non-null value throws. + @Nullable + public static Object variantGet(@Nullable byte[] envelope, String path, String targetType) { + return variantGet(envelope, compilePath(path), parseResultType(targetType)); + } + + /// Strictly extracts using pre-parsed path and type values. + @Nullable + public static Object variantGet(@Nullable byte[] envelope, VariantPath path, ResultType targetType) { + ReusableResult result = new ReusableResult(); + return extractInto(envelope, path, targetType, result) ? result.toExternalValue(targetType) : null; + } + + /// Strictly extracts into a reusable, unboxed result. + /// + /// @return {@code false} for SQL null, a missing path, or Variant null converted to a non-Variant target + public static boolean extractInto(@Nullable byte[] envelope, VariantPath path, ResultType targetType, + ReusableResult result) { + Objects.requireNonNull(result, "result must not be null"); + if (isSqlNull(envelope)) { + return false; + } + Objects.requireNonNull(path, "path must not be null"); + Objects.requireNonNull(targetType, "targetType must not be null"); + Cursor cursor = result._cursor; + if (!cursor.navigate(envelope, path)) { + return false; + } + if (cursor.getType() == Variant.Type.NULL && targetType != ResultType.VARIANT) { + return false; + } + convert(cursor, targetType, result); + return true; + } + + /// Tolerant Variant extraction. Malformed input returns Java null. + @Nullable + public static byte[] tryVariantGet(@Nullable byte[] envelope, String path) { + return (byte[]) tryVariantGet(envelope, compilePath(path), ResultType.VARIANT); + } + + /// Tolerant typed extraction. Malformed input and incompatible types return Java null. + @Nullable + public static Object tryVariantGet(@Nullable byte[] envelope, String path, String targetType) { + try { + return tryVariantGet(envelope, compilePath(path), parseResultType(targetType)); + } catch (RuntimeException e) { + return null; + } + } + + /// Tolerant extraction using pre-parsed path and type values. + @Nullable + public static Object tryVariantGet(@Nullable byte[] envelope, VariantPath path, ResultType targetType) { + try { + ReusableResult result = new ReusableResult(); + return tryExtractInto(envelope, path, targetType, result) ? result.toExternalValue(targetType) : null; + } catch (RuntimeException e) { + return null; + } + } + + /// Tolerantly extracts into a reusable, unboxed result. + /// + /// @return {@code false} for SQL null, missing paths, Variant null converted to a non-Variant target, + /// malformed input, or an incompatible conversion + public static boolean tryExtractInto(@Nullable byte[] envelope, VariantPath path, ResultType targetType, + ReusableResult result) { + Objects.requireNonNull(result, "result must not be null"); + if (isSqlNull(envelope)) { + return false; + } + Objects.requireNonNull(path, "path must not be null"); + Objects.requireNonNull(targetType, "targetType must not be null"); + Cursor cursor = result._cursor; + try { + if (!cursor.navigate(envelope, path)) { + return false; + } + if (cursor.getType() == Variant.Type.NULL && targetType != ResultType.VARIANT) { + return false; + } + return tryConvert(cursor, targetType, result); + } catch (IllegalArgumentException | IllegalStateException | UnsupportedOperationException + | IndexOutOfBoundsException e) { + // Cursor operations use these exceptions only for malformed or unsupported Variant encodings. + return false; + } + } + + /// Returns whether the path is present. A present Variant null counts as present. + @Nullable + public static Boolean variantExists(@Nullable byte[] envelope, String path) { + return variantExists(envelope, compilePath(path)); + } + + /// Returns whether a compiled path is present. A present Variant null counts as present. + @Nullable + public static Boolean variantExists(@Nullable byte[] envelope, VariantPath path) { + return variantExists(envelope, path, new ReusableResult()); + } + + /// Allocation-free compiled-path form of {@link #variantExists(byte[], VariantPath)} when the caller retains the + /// supplied result between rows. + @Nullable + public static Boolean variantExists(@Nullable byte[] envelope, VariantPath path, ReusableResult result) { + Objects.requireNonNull(result, "result must not be null"); + if (isSqlNull(envelope)) { + return null; + } + return result._cursor.navigate(envelope, Objects.requireNonNull(path, "path must not be null")); + } + + /// Returns whether the root value is a Variant null. SQL null is not a Variant null. + public static boolean isVariantNull(@Nullable byte[] envelope) { + return isVariantNull(envelope, ROOT_PATH, new ReusableResult()); + } + + /// Returns whether a present value at the path is a Variant null. SQL null and missing paths return false. + public static boolean isVariantNull(@Nullable byte[] envelope, String path) { + return isVariantNull(envelope, compilePath(path)); + } + + /// Returns whether a present value at a compiled path is a Variant null. SQL null and missing paths return false. + public static boolean isVariantNull(@Nullable byte[] envelope, VariantPath path) { + return isVariantNull(envelope, path, new ReusableResult()); + } + + /// Allocation-free compiled-path form of {@link #isVariantNull(byte[], VariantPath)} when the caller retains the + /// supplied result between rows. + public static boolean isVariantNull(@Nullable byte[] envelope, VariantPath path, ReusableResult result) { + Objects.requireNonNull(result, "result must not be null"); + if (isSqlNull(envelope)) { + return false; + } + Cursor cursor = result._cursor; + return cursor.navigate(envelope, Objects.requireNonNull(path, "path must not be null")) + && cursor.getType() == Variant.Type.NULL; + } + + /// Returns the Variant type name at the root, or Java null for SQL null. + @Nullable + public static String variantTypeOf(@Nullable byte[] envelope) { + return variantTypeOf(envelope, ROOT_PATH, new ReusableResult()); + } + + /// Returns the Variant type name at a path, or Java null for SQL null or a missing path. + @Nullable + public static String variantTypeOf(@Nullable byte[] envelope, String path) { + return variantTypeOf(envelope, compilePath(path)); + } + + /// Returns the Variant type name at a compiled path, or Java null for SQL null or a missing path. + @Nullable + public static String variantTypeOf(@Nullable byte[] envelope, VariantPath path) { + return variantTypeOf(envelope, path, new ReusableResult()); + } + + /// Allocation-free compiled-path form of {@link #variantTypeOf(byte[], VariantPath)} when the caller retains the + /// supplied result between rows. + @Nullable + public static String variantTypeOf(@Nullable byte[] envelope, VariantPath path, ReusableResult result) { + Objects.requireNonNull(result, "result must not be null"); + if (isSqlNull(envelope)) { + return null; + } + Cursor cursor = result._cursor; + return cursor.navigate(envelope, Objects.requireNonNull(path, "path must not be null")) + ? typeName(cursor.getType()) : null; + } + + /// Renders the Variant value as canonical JSON text without constructing a JSON tree. + @Nullable + public static String variantToJson(@Nullable byte[] envelope) { + return variantToJson(envelope, new ReusableResult()); + } + + /// Allocation-reduced form of [#variantToJson(byte[])] when the caller retains the supplied result between rows. + @Nullable + public static String variantToJson(@Nullable byte[] envelope, ReusableResult result) { + Objects.requireNonNull(result, "result must not be null"); + if (isSqlNull(envelope)) { + return null; + } + Cursor cursor = result._cursor; + cursor.navigate(envelope, ROOT_PATH); + return variantToJson(cursor.asVariant()); + } + + /// Parses JSON text into a Pinot Variant envelope without constructing a JSON tree. + @Nullable + public static byte[] parseJsonToVariant(@Nullable String json) { + if (json == null) { + return null; + } + try (JsonParser parser = JSON_FACTORY.createParser(json)) { + JsonToken token = parser.nextToken(); + if (token == null) { + throw new IllegalArgumentException("Cannot parse empty text as Variant"); + } + VariantBuilder builder = new VariantBuilder(); + appendJsonValue(parser, token, builder, 0); + if (parser.nextToken() != null) { + throw new IllegalArgumentException("Unexpected trailing token after Variant JSON value"); + } + Variant variant = builder.build(); + return VariantEnvelope.encode(variant.getMetadataBuffer(), variant.getValueBuffer()); + } catch (IOException | RuntimeException e) { + throw new IllegalArgumentException("Cannot parse JSON as Variant", e); + } + } + + /// Tolerant JSON parser. Malformed or unsupported input returns Java null. + @Nullable + public static byte[] tryParseJsonToVariant(@Nullable String json) { + try { + return parseJsonToVariant(json); + } catch (RuntimeException e) { + return null; + } + } + + private static boolean isSqlNull(@Nullable byte[] envelope) { + return envelope == null || envelope.length == 0; + } + + private static void convert(Cursor value, ResultType targetType, ReusableResult result) { + switch (targetType) { + case BOOLEAN: + requireType(value, Variant.Type.BOOLEAN, targetType); + result._intValue = value.getBoolean() ? 1 : 0; + break; + case INT: + result._intValue = toInt(value, targetType); + break; + case LONG: + result._longValue = toLong(value, targetType); + break; + case FLOAT: + result._floatValue = toFloat(value, targetType); + break; + case DOUBLE: + result._doubleValue = toDouble(value, targetType); + break; + case BIG_DECIMAL: + result._bigDecimalValue = toBigDecimal(value, targetType); + break; + case STRING: + requireType(value, Variant.Type.STRING, targetType); + result._stringValue = value.getString(); + break; + case BYTES: + requireType(value, Variant.Type.BINARY, targetType); + result._bytesValue = value.getBinary(); + break; + case UUID: + requireType(value, Variant.Type.UUID, targetType); + result._bytesValue = value.getUuidBytes(); + break; + case TIMESTAMP: + result._longValue = toTimestampMillis(value, targetType); + break; + case VARIANT: + result._bytesValue = value.copyEnvelope(); + break; + case JSON: + result._stringValue = variantToJson(value.asVariant()); + break; + default: + throw new IllegalStateException("Unhandled Variant target type: " + targetType); + } + } + + private static boolean tryConvert(Cursor value, ResultType targetType, ReusableResult result) { + Variant.Type valueType = value.getType(); + switch (targetType) { + case BOOLEAN: + if (valueType != Variant.Type.BOOLEAN) { + return false; + } + result._intValue = value.getBoolean() ? 1 : 0; + return true; + case INT: + return tryConvertToInt(value, valueType, result); + case LONG: + return tryConvertToLong(value, valueType, result); + case FLOAT: + switch (valueType) { + case BYTE: + case SHORT: + case INT: + case LONG: + result._floatValue = value.getInteger(); + return true; + case FLOAT: + result._floatValue = value.getFloat(); + return true; + case DOUBLE: + result._floatValue = (float) value.getDouble(); + return true; + case DECIMAL4: + case DECIMAL8: + case DECIMAL16: + result._floatValue = value.getDecimal().floatValue(); + return true; + default: + return false; + } + case DOUBLE: + switch (valueType) { + case BYTE: + case SHORT: + case INT: + case LONG: + result._doubleValue = value.getInteger(); + return true; + case FLOAT: + result._doubleValue = value.getFloat(); + return true; + case DOUBLE: + result._doubleValue = value.getDouble(); + return true; + case DECIMAL4: + case DECIMAL8: + case DECIMAL16: + result._doubleValue = value.getDecimal().doubleValue(); + return true; + default: + return false; + } + case BIG_DECIMAL: + switch (valueType) { + case BYTE: + case SHORT: + case INT: + case LONG: + result._bigDecimalValue = BigDecimal.valueOf(value.getInteger()); + return true; + case FLOAT: + float floatValue = value.getFloat(); + if (!Float.isFinite(floatValue)) { + return false; + } + result._bigDecimalValue = BigDecimal.valueOf(floatValue); + return true; + case DOUBLE: + double doubleValue = value.getDouble(); + if (!Double.isFinite(doubleValue)) { + return false; + } + result._bigDecimalValue = BigDecimal.valueOf(doubleValue); + return true; + case DECIMAL4: + case DECIMAL8: + case DECIMAL16: + result._bigDecimalValue = value.getDecimal(); + return true; + default: + return false; + } + case STRING: + if (valueType != Variant.Type.STRING) { + return false; + } + result._stringValue = value.getString(); + return true; + case BYTES: + if (valueType != Variant.Type.BINARY) { + return false; + } + result._bytesValue = value.getBinary(); + return true; + case UUID: + if (valueType != Variant.Type.UUID) { + return false; + } + result._bytesValue = value.getUuidBytes(); + return true; + case TIMESTAMP: + switch (valueType) { + case DATE: + result._longValue = value.getInteger() * TimeUnit.DAYS.toMillis(1); Review Comment: The tolerant path now uses `Math.multiplyExact` too. One correction to the reproduction: Parquet DATE is signed int32, and `Integer.MAX_VALUE * 86,400,000` still fits in a long, so a legal DATE cannot overflow. The max-int32 boundary now returns the exact millis in both strict and tolerant paths; malformed wider payloads are rejected. ########## pinot-common/src/test/java/org/apache/pinot/common/utils/VariantUtilsTest.java: ########## @@ -0,0 +1,1153 @@ +/** + * 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.pinot.common.utils; + +import java.math.BigDecimal; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.sql.Timestamp; +import java.util.Arrays; +import java.util.EnumSet; +import java.util.List; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; +import org.apache.calcite.sql.type.SqlTypeName; +import org.apache.parquet.variant.Variant; +import org.apache.parquet.variant.VariantBuilder; +import org.apache.parquet.variant.VariantObjectBuilder; +import org.apache.pinot.common.utils.DataSchema.ColumnDataType; +import org.apache.pinot.common.utils.VariantUtils.ResultType; +import org.apache.pinot.common.utils.VariantUtils.ReusableResult; +import org.apache.pinot.common.utils.VariantUtils.VariantPath; +import org.apache.pinot.spi.data.FieldSpec.DataType; +import org.apache.pinot.spi.utils.ByteArray; +import org.apache.pinot.spi.utils.UuidUtils; +import org.apache.pinot.spi.utils.VariantEnvelope; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNull; +import static org.testng.Assert.assertSame; +import static org.testng.Assert.assertThrows; +import static org.testng.Assert.assertTrue; + + +public class VariantUtilsTest { + @Test + public void testRawVariantResultRequiresNullHandling() { + DataSchema variantSchema = + new DataSchema(new String[]{"payload"}, new ColumnDataType[]{ColumnDataType.VARIANT}); + DataSchema typedSchema = + new DataSchema(new String[]{"eventType"}, new ColumnDataType[]{ColumnDataType.STRING}); + + assertTrue(VariantUtils.requiresNullHandlingForRawVariantResult(variantSchema, false)); + assertFalse(VariantUtils.requiresNullHandlingForRawVariantResult(variantSchema, true)); + assertFalse(VariantUtils.requiresNullHandlingForRawVariantResult(typedSchema, false)); + } + + @Test + public void testResultTypeContract() { + assertEquals(ResultType.BOOLEAN.getDataType(), DataType.BOOLEAN); + assertEquals(ResultType.BOOLEAN.getSqlTypeName(), SqlTypeName.BOOLEAN); + assertEquals(ResultType.INT.getDataType(), DataType.INT); + assertEquals(ResultType.INT.getSqlTypeName(), SqlTypeName.INTEGER); + assertEquals(ResultType.LONG.getDataType(), DataType.LONG); + assertEquals(ResultType.LONG.getSqlTypeName(), SqlTypeName.BIGINT); + assertEquals(ResultType.FLOAT.getDataType(), DataType.FLOAT); + assertEquals(ResultType.FLOAT.getSqlTypeName(), SqlTypeName.REAL); + assertEquals(ResultType.DOUBLE.getDataType(), DataType.DOUBLE); + assertEquals(ResultType.DOUBLE.getSqlTypeName(), SqlTypeName.DOUBLE); + assertEquals(ResultType.BIG_DECIMAL.getDataType(), DataType.BIG_DECIMAL); + assertEquals(ResultType.BIG_DECIMAL.getSqlTypeName(), SqlTypeName.DECIMAL); + assertEquals(ResultType.STRING.getDataType(), DataType.STRING); + assertEquals(ResultType.STRING.getSqlTypeName(), SqlTypeName.VARCHAR); + assertEquals(ResultType.BYTES.getDataType(), DataType.BYTES); + assertEquals(ResultType.BYTES.getSqlTypeName(), SqlTypeName.VARBINARY); + assertEquals(ResultType.UUID.getDataType(), DataType.UUID); + assertEquals(ResultType.UUID.getSqlTypeName(), SqlTypeName.UUID); + assertEquals(ResultType.TIMESTAMP.getDataType(), DataType.TIMESTAMP); + assertEquals(ResultType.TIMESTAMP.getSqlTypeName(), SqlTypeName.TIMESTAMP); + assertEquals(ResultType.VARIANT.getDataType(), DataType.VARIANT); + assertEquals(ResultType.VARIANT.getSqlTypeName(), SqlTypeName.VARIANT); + assertEquals(ResultType.JSON.getDataType(), DataType.JSON); + assertEquals(ResultType.JSON.getSqlTypeName(), SqlTypeName.VARCHAR); + } + + @Test + public void testDirectBinaryPathExtractionAndPredicates() { + byte[] variant = VariantUtils.parseJsonToVariant( + "{\"eventType\":\"click\",\"items\":[{\"price\":12.5},null],\"active\":true}"); + + assertEquals(VariantUtils.variantGet(variant, "$.eventType", "STRING"), "click"); + assertEquals((double) VariantUtils.variantGet(variant, "$.items[0].price", "DOUBLE"), 12.5); + assertEquals(VariantUtils.variantGet(variant, "$.active", "BOOLEAN"), true); + assertTrue(VariantUtils.variantExists(variant, "$.items[1]")); + assertFalse(VariantUtils.variantExists(variant, "$.missing")); + assertTrue(VariantUtils.isVariantNull(variant, "$.items[1]")); + assertFalse(VariantUtils.isVariantNull(variant, "$.missing")); + assertEquals(VariantUtils.variantTypeOf(variant, "$.items[0]"), "OBJECT"); + assertEquals(VariantUtils.variantTypeOf(variant, "$.items[0].price"), "DECIMAL"); + } + + @Test + public void testStrictAndTolerantExtraction() { + byte[] variant = VariantUtils.parseJsonToVariant("{\"eventType\":\"click\",\"score\":\"not-a-number\"}"); + + assertNull(VariantUtils.variantGet(variant, "$.missing", "STRING")); + assertThrows(IllegalArgumentException.class, () -> VariantUtils.variantGet(variant, "$.score", "DOUBLE")); + assertNull(VariantUtils.tryVariantGet(variant, "$.missing", "STRING")); + assertNull(VariantUtils.tryVariantGet(variant, "$.score", "DOUBLE")); + } + + @Test + public void testReusableResultStrictAndTolerantExtraction() { + byte[] variant = VariantUtils.parseJsonToVariant("{\"value\":7,\"null\":null,\"text\":\"x\"}"); + VariantPath valuePath = VariantUtils.compilePath("$.value"); + ReusableResult result = new ReusableResult(); + + assertTrue(VariantUtils.extractInto(variant, valuePath, ResultType.INT, result)); + assertEquals(result.getIntValue(), 7); + assertFalse(VariantUtils.extractInto(variant, VariantUtils.compilePath("$.missing"), ResultType.INT, result)); + assertFalse(VariantUtils.extractInto(variant, VariantUtils.compilePath("$.null"), ResultType.INT, result)); + assertThrows(IllegalArgumentException.class, + () -> VariantUtils.extractInto(variant, VariantUtils.compilePath("$.text"), ResultType.DOUBLE, result)); + assertFalse( + VariantUtils.tryExtractInto(variant, VariantUtils.compilePath("$.text"), ResultType.DOUBLE, result)); + assertFalse(VariantUtils.tryExtractInto(new byte[]{1}, valuePath, ResultType.INT, result)); + assertThrows(NullPointerException.class, () -> VariantUtils.extractInto( + variant, valuePath, ResultType.INT, null)); + assertThrows(NullPointerException.class, () -> VariantUtils.tryExtractInto( + variant, valuePath, ResultType.INT, null)); + } + + @Test + public void testReusableTolerantHeterogeneousMismatchesAndNumericRange() { + byte[][] rows = { + VariantUtils.parseJsonToVariant("{\"value\":\"not-an-int\"}"), + VariantUtils.parseJsonToVariant("{\"value\":true}"), + VariantUtils.parseJsonToVariant("{\"value\":{}}"), + VariantUtils.parseJsonToVariant("{\"value\":[]}"), + VariantUtils.parseJsonToVariant("{\"value\":2147483648}"), + VariantUtils.parseJsonToVariant("{\"value\":-2147483649}"), + VariantUtils.parseJsonToVariant("{\"value\":1.5}"), + VariantUtils.parseJsonToVariant("{\"value\":9223372036854775808}"), + VariantUtils.parseJsonToVariant("{\"value\":-9223372036854775809}") + }; + ResultType[] resultTypes = { + ResultType.INT, + ResultType.INT, + ResultType.INT, + ResultType.INT, + ResultType.INT, + ResultType.INT, + ResultType.INT, + ResultType.LONG, + ResultType.LONG + }; + VariantPath path = VariantUtils.compilePath("$.value"); + ReusableResult result = new ReusableResult(); + + for (int i = 0; i < rows.length; i++) { + assertFalse(VariantUtils.tryExtractInto(rows[i], path, resultTypes[i], result), + "Expected tolerant conversion to reject row " + i); + } + + assertThrows(IllegalArgumentException.class, + () -> VariantUtils.extractInto(rows[0], path, ResultType.INT, result)); + assertThrows(ArithmeticException.class, + () -> VariantUtils.extractInto(rows[4], path, ResultType.INT, result)); + assertThrows(ArithmeticException.class, + () -> VariantUtils.extractInto(rows[7], path, ResultType.LONG, result)); + + byte[] valid = VariantUtils.parseJsonToVariant("{\"value\":17}"); + assertTrue(VariantUtils.tryExtractInto(valid, path, ResultType.INT, result)); + assertEquals(result.getIntValue(), 17); + } + + @Test + public void testReusableResultParityForEveryResultType() { + VariantBuilder builder = new VariantBuilder(); + builder.appendBoolean(true); + assertReusableParity(encode(builder), ResultType.BOOLEAN); + + builder = new VariantBuilder(); + builder.appendInt(-17); + assertReusableParity(encode(builder), ResultType.INT); + + builder = new VariantBuilder(); + builder.appendLong(9_876_543_210L); + assertReusableParity(encode(builder), ResultType.LONG); + + builder = new VariantBuilder(); + builder.appendFloat(1.25F); + assertReusableParity(encode(builder), ResultType.FLOAT); + + builder = new VariantBuilder(); + builder.appendDouble(-123.5D); + assertReusableParity(encode(builder), ResultType.DOUBLE); + + builder = new VariantBuilder(); + builder.appendDecimal(new BigDecimal("12345678901234567890.1234")); + assertReusableParity(encode(builder), ResultType.BIG_DECIMAL); + + builder = new VariantBuilder(); + builder.appendString("a UTF-8 value \uD83D\uDE00"); + assertReusableParity(encode(builder), ResultType.STRING); + + builder = new VariantBuilder(); + builder.appendBinary(ByteBuffer.wrap(new byte[]{0, 1, -1, 42})); + assertReusableParity(encode(builder), ResultType.BYTES); + + builder = new VariantBuilder(); + builder.appendUUID(UUID.fromString("00112233-4455-6677-8899-aabbccddeeff")); + assertReusableParity(encode(builder), ResultType.UUID); + + builder = new VariantBuilder(); + builder.appendTimestampNanosTz(-1_234_567_890L); + assertReusableParity(encode(builder), ResultType.TIMESTAMP); + + byte[] nested = VariantUtils.parseJsonToVariant("{\"payload\":{\"count\":7}}"); + assertReusableParity(nested, VariantUtils.compilePath("$.payload"), ResultType.VARIANT); + assertReusableParity(nested, VariantUtils.compilePath("$.payload"), ResultType.JSON); + } + + @Test + public void testReusableNumericAndTemporalEncodingParity() { + VariantBuilder builder = new VariantBuilder(); + builder.appendByte((byte) -8); + byte[] byteValue = encode(builder); + assertReusableParity(byteValue, ResultType.INT); + assertReusableParity(byteValue, ResultType.LONG); + assertReusableParity(byteValue, ResultType.FLOAT); + assertReusableParity(byteValue, ResultType.DOUBLE); + assertReusableParity(byteValue, ResultType.BIG_DECIMAL); + + builder = new VariantBuilder(); + builder.appendShort((short) 32_000); + assertReusableParity(encode(builder), ResultType.INT); + + builder = new VariantBuilder(); + builder.appendDecimal(new BigDecimal("123.45")); + assertReusableParity(encode(builder), ResultType.BIG_DECIMAL); + + builder = new VariantBuilder(); + builder.appendDecimal(new BigDecimal("123.00")); + byte[] integralDecimal = encode(builder); + assertReusableParity(integralDecimal, ResultType.INT); + assertReusableParity(integralDecimal, ResultType.LONG); + + builder = new VariantBuilder(); + builder.appendDecimal(new BigDecimal("1234567890123.45")); + assertReusableParity(encode(builder), ResultType.BIG_DECIMAL); + + builder = new VariantBuilder(); + builder.appendString("x".repeat(128)); + assertReusableParity(encode(builder), ResultType.STRING); + + builder = new VariantBuilder(); + builder.appendDate(-1); + byte[] dateValue = encode(builder); + assertReusableParity(dateValue, ResultType.TIMESTAMP); Review Comment: Added `testDateToTimestampInt32BoundaryDoesNotOverflow`, covering `variantGet`, `tryVariantGet`, and reusable strict/tolerant extraction at `Integer.MAX_VALUE`. It asserts exact success because the full legal int32 DATE range fits in long milliseconds. ########## pinot-spi/src/main/java/org/apache/pinot/spi/data/FieldSpec.java: ########## @@ -909,6 +991,9 @@ public int hashCode(Object value) { /// return -1 if value1 is less than value2 /// return 1 if value1 is greater than value2 public int compare(Object value1, Object value2) { + if (!supportsOrdering()) { + throw new UnsupportedOperationException(this + " does not support ordering"); Review Comment: Narrowed the throwing branches to `this == VARIANT`. STRUCT/MAP/OPEN_STRUCT/LIST/UNKNOWN retain their previous exception behavior, and the PR body explicitly records that non-Variant behavior is preserved. ########## pinot-spi/src/main/java/org/apache/pinot/spi/data/FieldSpec.java: ########## @@ -880,6 +952,10 @@ public Object convert(String value) { return value; case BYTES: return BytesUtils.toBytes(value); + case VARIANT: + byte[] envelope = BytesUtils.toBytes(value); + VariantEnvelope.decode(envelope); Review Comment: Special-cased the zero-length Variant SQL-null sentinel in both conversion paths. An explicit empty hex/default now round-trips, while non-empty values are still envelope-validated; schema and FieldSpec regressions cover it. ########## pinot-spi/src/main/java/org/apache/pinot/spi/data/FieldSpec.java: ########## @@ -684,6 +691,9 @@ protected void appendDefaultNullValue(ObjectNode jsonNode) { case BYTES: jsonNode.put(key, BytesUtils.toHexString((byte[]) _defaultNullValue)); break; + case VARIANT: Review Comment: Collapsed BYTES and VARIANT into the shared serialization switch arm. The conversion arms remain separate because only VARIANT validates non-empty envelopes. -- 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]
