gortiz commented on code in PR #19101: URL: https://github.com/apache/pinot/pull/19101#discussion_r3914270630
########## pinot-spi/VARIANT_DESIGN.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: Read this first: it is the real spec for the PR, and it is unusually good. Sections 5.1 (capability policy) and 7.1 (the four null states) are what everything else derives from. Two housekeeping points. (1) The doc lives in pinot-spi/ root, which is a source module, not a docs location - it will be shipped inside the pinot-spi source tree and will not be found by anyone reading pinot-docs. Move it to pinot-docs, or to a top-level design/ or docs/design/ folder. (2) Section 8 says numbers 22 and 23 'remain reserved for the separately allocated UUID contract'; they are not reserved, they are already assigned to UUID and UUID_ARRAY. That sentence is in the paragraph about permanent wire allocations, so it should be exact. ########## 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: The framing layer is the strongest part of the change: magic, version, flags, reserved bytes, negative lengths and length agreement are all validated, and decode() aliases the input without copying. Two small notes. (1) allocate() is public and returns an array with a valid header but an uninitialised payload, so isEnvelope() returns true for it. The javadoc says callers must fill both regions before publishing, but a public factory that can hand out a 'valid' envelope full of zeros is easy to misuse - consider package-private, or a name that says so (allocateUnfilled). (2) The class javadoc describes the two length fields as 'unsigned range restricted to Java array sizes' while VARIANT_DESIGN.md section 4 describes them as 'big-endian signed int restricted to >= 0'. Same behaviour, two descriptions of a format that is explicitly frozen - pick one wording and use it in both places. ########## 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: Permanent wire number 24 - the one line in this PR that can never be taken back. Worth an explicit sign-off on the number alone, separate from the rest of the review. Everything else here is revertible. An enum number that has shipped in a release is not: any future reuse of 24 silently misinterprets in-flight plans between mixed-version nodes. The frozen legacy_expressions.proto test that asserts an old peer reads 24 as UNRECOGNIZED is the right guard and is well done. ########## 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: Test gap for the DATE -> TIMESTAMP overflow above: this suite covers DATE conversion parity but never an epoch-day value large enough to overflow the millis multiply. assertReusableParity() compares strict and tolerant paths for well-behaved values, so it cannot catch the case where strict throws and tolerant returns a wrapped long - the two paths genuinely disagree and the harness never asks them to. Please add a case with epochDay near Integer.MAX_VALUE asserting variantGet throws and tryVariantGet returns null. The suite is otherwise excellent - 72 cases covering every physical encoding, the wide-object threshold, and the parquet-java-vs-Arrow-RS key ordering divergence is real diligence. ########## 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: The only legal default null value for a VARIANT column cannot be expressed in schema JSON. convert("") -> BytesUtils.toBytes("") = empty array -> VariantEnvelope.decode(empty) throws, and the user sees an opaque 'Cannot convert value' message. SchemaUtils.validateVariantFieldSpec requires the default to equal DEFAULT_DIMENSION_NULL_VALUE_OF_BYTES - i.e. exactly the empty array that convert() refuses to produce. So a schema that spells out its (correct, and only allowed) default is rejected, while the identical schema that omits the field is accepted. Special-case the empty array here - it is the documented SQL-null sentinel, deliberately not an envelope, and the class javadoc already says so - or at minimum make the error message say 'VARIANT default null value must be omitted'. ########## pinot-common/src/main/java/org/apache/pinot/common/utils/DataSchema.java: ########## @@ -465,6 +473,56 @@ public ColumnDataType getStoredType() { return _storedColumnDataType; } + public boolean supportsEquality() { Review Comment: This is the mechanism behind my next comment: every capability predicate here returns false for arrays and OBJECT (and MAP for pattern matching), not just for VARIANT. Everything that consults them therefore widens beyond VARIANT. The delegation to toCapabilityDataType() is clean and the array/OBJECT short-circuits are defensible on their own terms - hashing a Java array by identity was never correct. The issue is only that the new callers are query-path guards that previously admitted these types. See TypeCapabilityValidationVisitor.visitSetOp. ########## pinot-core/src/main/java/org/apache/pinot/core/query/utils/OrderByComparatorFactory.java: ########## @@ -51,6 +52,12 @@ public static Comparator<Object[]> getComparator(List<OrderByExpressionContext> throw new BadQueryRequestException("MV expression: " + orderByExpressions.get(i) + " should not be included in the ORDER-BY clause"); } + FieldSpec.DataType dataType = orderByColumnContexts[i].getDataType(); + if (!dataType.supportsOrdering()) { Review Comment: Same widening on the single-stage side, and here UNKNOWN is the reachable one: DataType.UNKNOWN.supportsOrdering() is false, so any ORDER BY over a NULL-typed expression now fails with BadQueryRequestException. A literal null expression gets DataType.UNKNOWN from its ColumnContext, so a generated 'ORDER BY NULL' (not exotic in BI-tool SQL) is the concrete risk. Please verify it still works and add a test either way. If UNKNOWN should stay permissive - and I think it should, since ordering nulls is well defined - gate this on == DataType.VARIANT rather than on !supportsOrdering(), or add UNKNOWN to supportsOrdering()'s true set. The same question applies to PredicateEvaluatorProvider, where RANGE now requires supportsOrdering() and would reject an UNKNOWN-typed data source. ########## pinot-core/src/test/java/org/apache/pinot/core/query/utils/OrderByComparatorFactoryTest.java: ########## @@ -104,4 +109,32 @@ public void testTwoNullsCompareNextColumn() { assertEquals(extractColumn(_rows, COLUMN2_INDEX), Arrays.asList(1, 2, 3)); } + + @Test + public void testRejectsRawVariant() { Review Comment: Test gap that mirrors the risk exactly: the new tests cover VARIANT and MAP, but MAP can barely reach an ORDER BY in practice while UNKNOWN can (ORDER BY on a null literal) - and UNKNOWN is untested. Please add a third case with DataType.UNKNOWN asserting the intended behaviour. If the answer is 'UNKNOWN must still be orderable', that test will fail today and shows the guard needs narrowing to VARIANT. ########## pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/operands/FilterOperand.java: ########## @@ -193,7 +197,12 @@ public Predicate(List<RexExpression> operands, DataSchema dataSchema, IntPredica ColumnDataType lhsType = _lhs.getResultType(); ColumnDataType rhsType = _rhs.getResultType(); - if (lhsType == rhsType) { + // Reject raw VARIANT operands only; other non-orderable types (OBJECT, arrays, MAP) keep their existing + // best-effort comparison behavior. VARIANT is opaque because its PVAR byte encoding is not a canonical + // semantic ordering, so a comparison must extract a typed scalar first. + Preconditions.checkArgument(lhsType != ColumnDataType.VARIANT && rhsType != ColumnDataType.VARIANT, + "Raw VARIANT values do not support comparison; extract a typed path with variantGet first"); + if (lhsType == ColumnDataType.UNKNOWN || rhsType == ColumnDataType.UNKNOWN || lhsType == rhsType) { Review Comment: Unrelated to VARIANT but semantically real: this short-circuits comparison casting whenever either side is UNKNOWN, changing NULL-literal comparison behaviour for all types. BinaryOperatorTransformFunction:116 does the equivalent on the single-stage side via _alwaysNull. Both look like genuine fixes - previously 'NULL = intCol' computed a common cast type and could fail on an UNKNOWN stored type - and both are tested (FilterOperandTest.testComparisonWithNullLiteral, BinaryOperatorTransformFunctionTest.testLeftNullRightLiteral), which is good. The only gap is disclosure: these are SQL null-comparison semantics changes to shared code, landing inside a VARIANT feature PR under a commit message ('Preserve SQL null comparison semantics') that reads like a no-op restoration. Please call them out in the PR description so the release notes pick them up. ########## pinot-common/src/main/java/org/apache/pinot/common/function/TransformFunctionType.java: ########## @@ -117,6 +122,17 @@ public enum TransformFunctionType { JSON_EXTRACT_KEY("jsonExtractKey", ReturnTypes.TO_ARRAY, OperandTypes.family( List.of(SqlTypeFamily.CHARACTER, SqlTypeFamily.CHARACTER, SqlTypeFamily.CHARACTER), i -> i > 1)), + VARIANT_GET("variantGet", TransformFunctionType::variantGetReturnTypeInference, variantGetOperandTypeChecker()), + TRY_VARIANT_GET("tryVariantGet", TransformFunctionType::variantGetReturnTypeInference, + variantGetOperandTypeChecker()), + VARIANT_EXISTS("variantExists", ReturnTypes.BOOLEAN_NULLABLE, variantPathOperandTypeChecker()), + IS_VARIANT_NULL("isVariantNull", ReturnTypes.BOOLEAN, optionalVariantPathOperandTypeChecker()), + VARIANT_TYPE_OF("variantTypeOf", ReturnTypes.VARCHAR_2000_NULLABLE, optionalVariantPathOperandTypeChecker()), + VARIANT_TO_JSON("variantToJson", ReturnTypes.VARCHAR_2000_NULLABLE, OperandTypes.ANY), + PARSE_JSON_TO_VARIANT("parseJson", TransformFunctionType::nullableVariantReturnTypeInference, Review Comment: Naming: parseJson is a very generic name to claim for a function that returns VARIANT rather than Pinot's JSON type, and it sits directly among jsonExtractScalar / jsonExtractKey / jsonFormat, which all mean Pinot JSON. Nothing named parseJson exists today so there is no conflict, and the Spark precedent (parse_json -> variant) is a fair argument. But a user reading the function list will reasonably expect parseJson to produce the JSON type, and the alias parseJsonToVariant is the self-documenting spelling. Suggest making parseJsonToVariant (and tryParseJsonToVariant) the primary names and parseJson / tryParseJson the aliases, so the generic name is not the one people find first and the return type is obvious at the call site. Also, unrelated but on this block: VARIANT_TO_JSON uses OperandTypes.ANY, so variantToJson(intCol) passes planning and only fails at runtime - worth a real operand checker. ########## pinot-common/pom.xml: ########## @@ -200,6 +200,18 @@ <groupId>org.apache.pinot</groupId> <artifactId>pinot-timeseries-spi</artifactId> </dependency> + <dependency> + <groupId>org.apache.parquet</groupId> + <artifactId>parquet-variant</artifactId> Review Comment: The parquet-column exclusion is unsound: VariantBuilder, the class parseJsonToVariant() instantiates, has a member referencing org.apache.parquet.io.api.Binary, which lives in the excluded artifact. Drop the exclusion - and note that the decoder half of the library is provably clean, so a narrower boundary is available if the 3.3 MB matters. Verified against parquet-variant 1.18.0. Of the classes Pinot touches, exactly one is 'dirty': VariantBuilder has `void appendAsString(org.apache.parquet.io.api.Binary)`, and Binary ships in parquet-column, which this exclusion removes. Everything on the decode side - Variant, Variant$Type, VariantUtil, Metadata, ImmutableMetadata, MetadataBuilder - and also VariantObjectBuilder and VariantArrayBuilder have zero references outside parquet-common. It works today only because the JVM resolves method descriptors lazily and appendJsonValue() calls only appendString/appendInt/appendLong/appendDecimal/appendBoolean/appendNull/startObject/startArray; it would become a runtime NoClassDefFoundError, with no build-time signal, if anything reflected over VariantBuilder's declared methods or a future parquet release routed a called method through appendAsString. Recommendation: drop the exclusion. You cannot exclude an artifact you compile against, and a lazily-unresolved descriptor is not a safety property to rely on. The cost to be aware of is that this is exactly what the exclusion was for - parquet-column is ~3.3 MB and pulls parquet-generator at compile scope, so pinot-common's closure grows by roughly that (worth a `mvn dependency:tree -pl pinot-common` in the PR to state the real number). If that cost is unwelcome, there is a narrower boundary that keeps the original intent AND is sound, because the dirty/clean split falls almost exactly where the exclusion was aiming: only the JSON-to-Variant construction path (parseJsonToVariant / tryParseJsonToVariant, i.e. the VariantBuilder hierarchy) needs parquet-column. Move just that pair out of pinot-common - its consumers are ingestion (InbuiltFunctionEvaluator) and the MSE literal-folding path (LiteralParseJsonOperand), neither of which is pinot-common-only - and pinot-common retains only parquet-common-clean classes, at which point the exclusion becomes correct rather than accidental. Two things to fix regardless of which route you take. (a) parquet-jackson is declared at runtime scope and is NOT excluded - a 2.4 MB shaded-Jackson jar in pinot-common's closure for no benefit, since Pinot uses its own Jackson. One-line win. (b) The reason a pinot-common dependency hurts disproportionately today is the global shade, which as apache/pinot#18459 argues buys us no real isolation in return. Worth a comment right here recording that, e.g. 'Dependencies added to pinot-common are amplified by the global shade; see apache/pinot#18459 (Replace global shading with classloader-realm isolation for plugins), which removes it - pushing that through reduces the blast radius of additions like this one.' That keeps the tradeoff visible without blocking this PR on an unrelated build change. ########## pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/HashJoinOperator.java: ########## @@ -79,18 +99,55 @@ public HashJoinOperator(OpChainExecutionContext context, MultiStageOperator left _nullKeyRightRows = needUnmatchedRightRows() ? new ArrayList<>() : null; } - /// Constructor that takes the schema for NonEquiEvaluator as an argument + /// Constructor that takes the schema for NonEquiEvaluator as an argument. + /// + /// <p>For SEMI and ANTI joins whose node does not carry its inputs, the result schema contains only left columns, so + /// this legacy constructor cannot validate the right key's logical type. New callers that need right-side VARIANT + /// validation must use the overload that accepts {@code rightSchema}. public HashJoinOperator(OpChainExecutionContext context, MultiStageOperator leftInput, DataSchema leftSchema, MultiStageOperator rightInput, JoinNode node, DataSchema nonEquiEvaluationSchema) { + this(context, leftInput, leftSchema, rightInput, tryInferRightSchema(leftSchema, node), node, + nonEquiEvaluationSchema, false); + } + + /// Constructor that takes the schema for NonEquiEvaluator as an argument + public HashJoinOperator(OpChainExecutionContext context, MultiStageOperator leftInput, DataSchema leftSchema, + MultiStageOperator rightInput, DataSchema rightSchema, JoinNode node, DataSchema nonEquiEvaluationSchema) { + this(context, leftInput, leftSchema, rightInput, rightSchema, node, nonEquiEvaluationSchema, true); + } + + private HashJoinOperator(OpChainExecutionContext context, MultiStageOperator leftInput, DataSchema leftSchema, + MultiStageOperator rightInput, @Nullable DataSchema rightSchema, JoinNode node, + DataSchema nonEquiEvaluationSchema, boolean rightSchemaRequired) { super(context, leftInput, leftSchema, rightInput, node, nonEquiEvaluationSchema); List<Integer> leftKeys = node.getLeftKeys(); Preconditions.checkState(!leftKeys.isEmpty(), "Hash join operator requires join keys"); + Preconditions.checkArgument(!rightSchemaRequired || rightSchema != null, "Right input schema must not be null"); + JoinKeyTypeValidator.validate(node, leftSchema, rightSchema); _leftKeySelector = KeySelectorFactory.getKeySelector(leftKeys); _rightKeySelector = KeySelectorFactory.getKeySelector(node.getRightKeys()); _rightTable = createLookupTable(leftKeys, leftSchema); _matchedRightRows = needUnmatchedRightRows() ? new HashMap<>() : null; } + @Nullable + private static DataSchema tryInferRightSchema(DataSchema leftSchema, JoinNode node) { Review Comment: tryInferRightSchema reconstructs the right schema by slicing the result schema, and returns null for SEMI/ANTI joins - where JoinKeyTypeValidator.validate then validates nothing. The result is a guard that looks present but silently does not apply on exactly the paths the javadoc admits it cannot cover. The javadoc is honest, which I appreciate, but the outcome is four overlapping constructors where two enforce the contract and two do not, and a raw VARIANT right-hand key in a SEMI or ANTI join reaches the lookup table with no check. Since the planner-level TypeCapabilityValidationVisitor should normally catch it first, this is defence-in-depth rather than the only line - but then the defence has a documented hole. Two options: deprecate the two legacy constructors and fix the in-repo call sites to pass rightSchema explicitly (they appear to be the only callers), or make the null-rightSchema case still validate the left keys and log rather than skip entirely. Either way, worth a line in the PR description since the javadoc is currently the only record of the limitation. ########## compatibility-verifier/compCheck.sh: ########## @@ -496,14 +498,24 @@ setupControllerVariables setupBrokerVariables setupServerVariables -export JAVA_OPTS="-DControllerPort=${CONTROLLER_PORT} -DBrokerQueryPort=${BROKER_QUERY_PORT} -DServerAdminPort=${SERVER_ADMIN_PORT}" - mkdir ${PID_DIR} mkdir ${LOG_DIR} oldTargetDir="$workingDir"/oldTargetDir newTargetDir="$workingDir"/newTargetDir +oldServerSupportsVariant=false +oldExpressionsProto="${oldTargetDir}/pinot-common/src/main/proto/expressions.proto" +if [ -f "${oldExpressionsProto}" ] \ + && grep -Eq '^[[:space:]]*VARIANT[[:space:]]*=[[:space:]]*24[[:space:]]*;' "${oldExpressionsProto}"; then Review Comment: This decides whether to run the Variant wire assertions by grepping the OLD checkout's expressions.proto source text for the literal pattern `VARIANT = 24;`. If that grep ever stops matching for a reason unrelated to capability, the suite quietly skips the Variant checks instead of failing. To spell out the mechanism: the compat suite builds two Pinot trees (old and new) and needs to know whether the OLD side understands the VARIANT wire type, because the expected outcome differs - an old server must reject the type deterministically, a new one must serve it. This script answers that question by reading the old tree's .proto file as text and setting -Dpinot.compat.oldServerSupportsVariant, which the new runIfSystemProperty gate on BaseOp then uses to enable or skip individual ops. The fragility is that the answer is derived from source formatting rather than from the built artifact. Reformat the enum, move it to another file, add a comment between the name and the number, renumber it, or wrap the line differently, and the grep returns false. False means 'old side does not support VARIANT', which means the ops gated on the true branch are skipped and the run still reports success - so the failure mode is a green build with the Variant compatibility coverage silently switched off, which is the worst possible outcome for a suite whose entire job is proving the mixed-version contract. It also means the gate silently rots the first time someone touches that proto file for an unrelated reason. Two better options: derive the capability from the built old artifact (reflect over the generated ColumnDataType enum in the old jar and check for a VARIANT constant - that is the actual thing you care about, and it cannot disagree with reality), or keep the grep but fail loudly if the file exists and the enum block cannot be parsed at all, so 'I could not tell' is distinguishable from 'the answer is no'. Small separate note on this file: the new old-broker-new-servers phase at line 621 passes "$genNum" without incrementing it, unlike every other phase. That is presumably deliberate so the old broker queries the generation that already exists - but it should be commented, because it will silently collide the day that yaml gains a segmentOp or tableOp. ########## 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: Correctness: strict/tolerant asymmetry on DATE -> TIMESTAMP. toTimestampMillis() uses Math.multiplyExact, but this tryConvert path uses a plain multiply - so try_variant_get(v, '$.d', 'TIMESTAMP') silently returns a wrapped garbage timestamp where the strict form correctly throws. The try_ contract is 'return SQL null on failure', not 'return a wrong answer quietly' - a silently wrapped value is worse than an error because nothing downstream can detect it. The fix is Math.multiplyExact wrapped so overflow returns false, which is exactly the shape tryConvertToInt already uses for LONG out of range a few lines above. An epoch-day value only has to exceed ~106 million to overflow, which a hostile or corrupt producer can trivially emit. ########## 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: Design question worth settling before merge: Cursor is a ~700-line hand-rolled reimplementation of the Parquet Variant binary decoder that Pinot already has on the classpath (org.apache.parquet.variant.Variant), duplicating every type tag, offset layout and metadata-dictionary rule. The performance motivation is real - zero-copy navigation, no per-row Variant/ByteBuffer allocation, and the cross-row dictionary/entry-index memo is genuinely clever - and asVariant() proves the two can coexist. But this is now a second implementation of a spec Pinot does not own. Concrete failure mode: when parquet-variant gains a primitive type tag or a v2 encoding, the Parquet converters accept the file at ingestion while this cursor throws UnsupportedOperationException from getType()/encodedValueLength() at query time. So the data lands and only breaks when someone queries it. Please state the sync policy explicitly - a test that asserts this constant table matches parquet-variant's VariantUtil constants, or a hard-pinned parquet version with a comment - and consider falling back to the library decoder for tags Cursor does not recognise rather than throwing. ########## 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: Worth calling out in the PR body even if it stays as-is: compare() now throws UnsupportedOperationException for STRUCT / MAP / OPEN_STRUCT / LIST / UNKNOWN, which previously reached the default branch and threw IllegalStateException. equals() and hashCode() gained a new throwing branch on the same public SPI enum. The fail-closed design is right for VARIANT; the incidental part is that five pre-existing types changed their exception type. It only bites a caller doing catch (IllegalStateException) - unlikely in-tree, possible in a plugin or a downstream fork. If you keep it, mention it in the PR description so it reaches the release notes; if you would rather have zero blast radius, throw only for VARIANT here and let the other five fall through to the old default branch as before. ########## 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: Nit: this VARIANT arm is byte-identical to the BYTES arm three lines above. Just fall through (case BYTES: case VARIANT:) as the file already does elsewhere for the BYTES/VARIANT pair. Same in convert() and convertInternal(), where the VARIANT arms differ from BYTES only by the added VariantEnvelope.decode() validation - those are worth keeping separate. This one is pure duplication and will drift the first time the BYTES hex encoding changes. ########## pinot-query-planner/src/main/java/org/apache/pinot/query/planner/validation/TypeCapabilityValidationVisitor.java: ########## @@ -0,0 +1,199 @@ +/** + * 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.query.planner.validation; + +import java.util.List; +import org.apache.calcite.rel.RelFieldCollation; +import org.apache.pinot.common.utils.DataSchema; +import org.apache.pinot.common.utils.VariantUtils; +import org.apache.pinot.query.planner.logical.RexExpression; +import org.apache.pinot.query.planner.plannode.AggregateNode; +import org.apache.pinot.query.planner.plannode.JoinNode; +import org.apache.pinot.query.planner.plannode.PlanNode; +import org.apache.pinot.query.planner.plannode.PlanNodeVisitor; +import org.apache.pinot.query.planner.plannode.SetOpNode; +import org.apache.pinot.query.planner.plannode.SortNode; +import org.apache.pinot.query.planner.plannode.WindowNode; +import org.apache.pinot.spi.exception.QueryErrorCode; +import org.apache.pinot.spi.exception.QueryException; + + +/// Validates that each logical input type supports the capabilities required by its operation, including equality, +/// hashing, ordering, aggregation, and lossless result projection. The visitor has no mutable state and is thread-safe, +/// so callers may share {@link #INSTANCE}. +public final class TypeCapabilityValidationVisitor extends PlanNodeVisitor.DepthFirstVisitor<Void, Void> { + public static final TypeCapabilityValidationVisitor INSTANCE = new TypeCapabilityValidationVisitor(); + + private TypeCapabilityValidationVisitor() { + } + + @Override + protected boolean traverseStageBoundary() { + return false; + } + + @Override + public Void visitAggregate(AggregateNode node, Void context) { + List<PlanNode> inputs = node.getInputs(); + if (inputs.size() == 1) { + validateAggregateInputs(node, inputs.get(0).getDataSchema()); + } + return super.visitAggregate(node, context); + } + + /// Validates aggregate operands against their logical input schema. + /// + /// <p>This method is also invoked by the runtime as a defensive check for plans that did not pass through the + /// current broker planner. + public static void validateAggregateInputs(AggregateNode node, DataSchema inputSchema) { + for (int key : node.getGroupKeys()) { + DataSchema.ColumnDataType dataType = inputSchema.getColumnDataType(key); + if (!dataType.supportsEquality() || !dataType.supportsHashing()) { + throw unsupported("GROUP BY", dataType); + } + } + validateAggregateInputs(node.getAggCalls(), inputSchema); + } + + /// Validates aggregate or window-function operands against their logical input schema. + public static void validateAggregateInputs(List<RexExpression.FunctionCall> aggCalls, DataSchema inputSchema) { + for (RexExpression.FunctionCall aggCall : aggCalls) { + if (isRawVariantIndependent(aggCall)) { + continue; + } + for (RexExpression operand : aggCall.getFunctionOperands()) { + DataSchema.ColumnDataType dataType = getLogicalType(operand, inputSchema); + if (!dataType.supportsDirectAggregation()) { + throw unsupported("Aggregate function " + aggCall.getFunctionName(), dataType); + } + } + } + } + + /// Rejects a raw VARIANT result when query null handling is disabled. Without the null bitmap, the reserved empty + /// byte placeholder cannot participate in normal disabled-null semantics while also remaining distinguishable from + /// an encoded Variant null. + public static void validateResultSchema(DataSchema resultSchema, boolean nullHandlingEnabled) { + if (VariantUtils.requiresNullHandlingForRawVariantResult(resultSchema, nullHandlingEnabled)) { + throw new QueryException(QueryErrorCode.QUERY_PLANNING, + VariantUtils.RAW_VARIANT_REQUIRES_NULL_HANDLING_ERROR); + } + } + + @Override + public Void visitSort(SortNode node, Void context) { + DataSchema dataSchema = node.getDataSchema(); + for (RelFieldCollation collation : node.getCollations()) { + int fieldIndex = collation.getFieldIndex(); + DataSchema.ColumnDataType dataType = dataSchema.getColumnDataType(fieldIndex); + if (!dataType.supportsOrdering()) { + throw unsupported("ORDER BY", dataType); + } + } + return super.visitSort(node, context); + } + + @Override + public Void visitSetOp(SetOpNode node, Void context) { Review Comment: Biggest scope concern in the PR. This visitor runs on EVERY multi-stage plan, and the capability predicates are false for arrays and OBJECT as well as VARIANT - so these guards reject non-VARIANT queries that plan successfully today. Concretely: INTERSECT / EXCEPT / UNION DISTINCT over any array or OBJECT column becomes a QUERY_PLANNING error, and SortOperator adds a matching runtime Preconditions check for ORDER BY. The unsupported() helper explicitly formats messages for 'non-VARIANT opaque types (OBJECT, arrays, MAP)', and OrderByComparatorFactoryTest adds a MAP rejection test, so this is clearly deliberate - but it is a user-visible behaviour change that the PR description does not mention. The backward-incompat label is already applied; what is missing is a sentence in the PR body so it reaches the release notes, plus the test coverage in my note on TypeCapabilityValidationVisitorTest. Note also that GROUP BY on an MV column still produces the friendlier 'Use ARRAY_TO_MV()' message only because runValidations() happens to run ArrayToMvValidationVisitor before this one - that ordering is now load-bearing and undocumented; a comment in PinotDispatchPlanner.runValidations would help. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/ColumnarValueNormalizer.java: ########## @@ -28,21 +28,21 @@ /// (`buildColumnar`) path. /// /// The row-major build runs each record through a `TransformPipeline` whose -/// `NullValueTransformer` substitutes [FieldSpec#getDefaultNullValue()] for `null` -/// and whose `DataTypeTransformer` coerces every value to the column's stored type (e.g. +/// `DataTypeTransformer` coerces every non-null value to the column's stored type (e.g. /// `Boolean` → `Integer` for a `BOOLEAN` column stored as `INT`, -/// `Timestamp` → `Long` for `TIMESTAMP`). The column-major driver deliberately runs -/// with no transform pipeline, so a non-segment source (e.g. Arrow) delivers values in the source's +/// `Timestamp` → `Long` for `TIMESTAMP`) and whose `NullValueTransformer` substitutes +/// [FieldSpec#getDefaultNullValue()] for values that remain `null`. The column-major driver deliberately runs with +/// no transform pipeline, so a non-segment source (e.g. Arrow) delivers values in the source's /// logical type with raw `null`s — which the typed collectors / index creators do not accept. /// /// This helper applies the equivalent of those two transformers to one value, in the same order: /// -/// 1. `NullValueTransformer`: a `null` value becomes the column default — the scalar -/// default for single-value columns, or a one-element `Object[]` of that scalar for -/// multi-value columns (matching `NullValueTransformerUtils.getDefaultNullValue`). -/// 2. `DataTypeTransformer`: [DataTypeTransformerUtils#transformValue] standardizes the +/// 1. `DataTypeTransformer`: [DataTypeTransformerUtils#transformValue] standardizes the Review Comment: This reorder is correct - I checked RecordTransformerUtils.getTransformers() and the real pipeline is DataTypeTransformer then NullValueTransformer, so the previous doc and code had it backwards - but it changes the columnar build path for every data type, not just VARIANT. Net effect: the FieldSpec default null value no longer passes through DataTypeTransformerUtils.transformValue. That happens to be safe as written (BOOLEAN default is Integer 0 and TIMESTAMP default is Long 0 - already in stored form), but the safety is incidental rather than enforced: a future type whose FieldSpec default is not already stored-form would now reach the collectors unconverted. Worth a line in the PR body, and see my test note on ColumnarValueNormalizerTest. ########## pinot-query-planner/src/test/java/org/apache/pinot/query/planner/validation/TypeCapabilityValidationVisitorTest.java: ########## @@ -0,0 +1,276 @@ +/** + * 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.query.planner.validation; + +import java.util.List; +import org.apache.calcite.rel.RelFieldCollation; +import org.apache.calcite.rel.core.JoinRelType; +import org.apache.pinot.common.utils.DataSchema; +import org.apache.pinot.query.planner.logical.RexExpression; +import org.apache.pinot.query.planner.plannode.AggregateNode; +import org.apache.pinot.query.planner.plannode.JoinNode; +import org.apache.pinot.query.planner.plannode.PlanNode; +import org.apache.pinot.query.planner.plannode.SetOpNode; +import org.apache.pinot.query.planner.plannode.SortNode; +import org.apache.pinot.query.planner.plannode.ValueNode; +import org.apache.pinot.query.planner.plannode.WindowNode; +import org.apache.pinot.spi.exception.QueryException; +import org.testng.Assert; +import org.testng.annotations.Test; + + +public class TypeCapabilityValidationVisitorTest { + private static final DataSchema VARIANT_SCHEMA = + new DataSchema(new String[]{"payload"}, new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.VARIANT}); + private static final DataSchema TYPED_EXTRACTION_SCHEMA = + new DataSchema(new String[]{"typedPayload"}, new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.STRING}); + + @Test + public void testRejectsVariantOrderBy() { + SortNode sortNode = new SortNode(0, VARIANT_SCHEMA, PlanNode.NodeHint.EMPTY, List.of(), + List.of(new RelFieldCollation(0)), 10, 0); + + QueryException exception = + Assert.expectThrows(QueryException.class, () -> sortNode.visit(TypeCapabilityValidationVisitor.INSTANCE, null)); + Assert.assertTrue(exception.getMessage().contains("ORDER BY")); + } + + @Test + public void testNamesUnsupportedNonVariantOrderByType() { Review Comment: Test gap: the guards now reject arrays, but this suite has zero array coverage. The only non-VARIANT case is this single OBJECT ORDER BY test. Since the widening to non-VARIANT types is intentional, it should be pinned. Please add cases for INT_ARRAY / STRING_ARRAY through visitSetOp (INTERSECT / EXCEPT / UNION DISTINCT), visitSort, visitAggregate group keys and visitWindow partition keys - asserting whatever the intended behaviour is. Right now a future change that relaxes or tightens array handling would not fail a single test, and the only signal that arrays are affected at all is a comment in the unsupported() helper. ########## pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/ColumnarValueNormalizerTest.java: ########## @@ -0,0 +1,43 @@ +/** + * 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.segment.local.segment.creator.impl; + +import org.apache.pinot.spi.data.DimensionFieldSpec; +import org.apache.pinot.spi.data.FieldSpec; +import org.apache.pinot.spi.data.FieldSpec.DataType; +import org.apache.pinot.spi.utils.PinotDataType; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertSame; + + +public class ColumnarValueNormalizerTest { + private static final String COLUMN = "payload"; + + @Test + public void testNullVariantReturnsDefaultSentinelWithoutDecoding() { Review Comment: Test gap: this new suite has exactly one test, and it only covers the VARIANT sentinel. The transformer reorder it accompanies affects every type on the columnar path, with nothing pinning the result. Please add cases asserting the normalized default for BOOLEAN (expect Integer 0, not Boolean), TIMESTAMP (expect Long, not java.sql.Timestamp) and a multi-value column (expect a one-element Object[] of the scalar default) - i.e. the invariants that make the reorder safe. Those three assertions are what turn 'happens to work' into 'guaranteed to work'. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java: ########## @@ -1970,7 +2060,7 @@ private static void validateStarTreeIndexConfigs(List<StarTreeIndexConfig> starT List<String> dimensionsSplitOrder = starTreeIndexConfig.getDimensionsSplitOrder(); assert CollectionUtils.isNotEmpty(dimensionsSplitOrder); for (String dimension : dimensionsSplitOrder) { - if (timestampIndexColumns.contains(dimension)) { + if (timestampIndexColumns.contains(dimension) && schema.getFieldSpecFor(dimension) == null) { Review Comment: This is the change in the PR most likely to break existing users, and it has nothing to do with VARIANT. Adding `&& schema.getFieldSpecFor(dimension) == null` means TIMESTAMP-index derived columns that ARE declared in the schema now go through full star-tree validation instead of being skipped - at three call sites in this method. To spell out the mechanism: previously any name in timestampIndexColumns was waved through all three loops unconditionally. Now it is only skipped when the schema has no field spec for it, so whenever the field spec IS present the column must satisfy every star-tree check - existence, not MAP, not VARIANT, single-value, and the dictionary-encoding/cardinality rules further down. Whether the field spec is present at validation time is not obvious, and that is the problem. TimestampIndexUtils.applyTimestampIndex() mutates the Schema object in place to add the derived columns, and it is invoked from SegmentGeneratorConfig, IndexLoadingConfig, SegmentProcessorConfig and HelixInstanceDataManager. So presence depends on whether the Schema instance reaching TableConfigUtils.validate() has already been through one of those paths - which is very plausibly why the original code skipped unconditionally rather than reasoning about it. Flipping that to a conditional makes validation outcome depend on Schema-instance provenance, and table-config validation failures surface on controller restart and on any subsequent config PUT, so if it does trigger it can block unrelated operational changes on a live table. I am not arguing against the intent - the updated javadoc's reasoning is sound, a user column colliding with a derived name should not bypass checks. But this is three one-line changes with no test, in a 15k-line diff about something else. Please add: (a) a test with a schema that declares $ts$DAY plus a star-tree config referencing it, asserting the intended outcome; (b) a clearly-labelled separate commit; (c) a note in the PR description, since this can change validation results for tables that contain no VARIANT column at all. ########## pinot-query-planner/src/main/java/org/apache/pinot/calcite/sql/fun/PinotOperatorTable.java: ########## @@ -470,6 +472,11 @@ public void lookupOperatorOverloads(SqlIdentifier opName, @Nullable SqlFunctionC if (!opName.isSimple()) { return; } + if (!_nullHandlingEnabled && TransformFunctionType.requiresNullHandling(opName.getSimple())) { Review Comment: Agreed on throwing a QueryException instead of IllegalStateException. Concrete optimization below: reuse the canonicalName this method already computes, and gate on the map hit so the check stops running for every operator in every query. Two separate problems, both fixable in this one method. (1) WRONG LAYER FOR THE ERROR - agreed this should be a QueryException. lookupOperatorOverloads is Calcite's name-resolution hook; an IllegalStateException thrown from there is not mapped by Pinot's error handling, so a user who forgot enableNullHandling=true gets an internal planner failure with a Calcite stack trace instead of a clean error, and the actionable 'set enableNullHandling=true' text gets buried. It is also an odd contract for a lookup method, which Calcite may call speculatively while resolving a name. While changing it, please also pick ONE error code for this condition. Right now the same user mistake reports differently depending on the path: TypeCapabilityValidationVisitor.validateResultSchema uses QUERY_PLANNING, while SelectionResultsBlock.getDataTable and BaseSingleStageBrokerRequestHandler.validateRawVariantResult both use QUERY_VALIDATION - for the identical 'raw VARIANT needs null handling' message. Clients keying on error codes will see two codes for one mistake. My preference is QUERY_VALIDATION everywhere (it is a query-option problem, not a planning failure), but consistency matters more than which one. (2) HOW TO OPTIMIZE - the method already canonicalizes the name two lines below, so the check is doing that work twice, and it does it for every operator name in every query rather than only for names that resolve to a Pinot operator. Reorder so the existing canonicalName is reused and the gate runs only on a map hit: String canonicalName = FunctionRegistry.canonicalize(opName.getSimple()); List<SqlOperator> operators = _operatorMap.get(canonicalName); if (operators == null) { return; } if (!_nullHandlingGatedNames.isEmpty() && _nullHandlingGatedNames.contains(canonicalName)) { throw new QueryException(QueryErrorCode.QUERY_VALIDATION, ...); } operatorList.addAll(operators); where _nullHandlingGatedNames is a field set in the constructor: nullHandlingEnabled ? Set.of() : TransformFunctionType.getNullHandlingRequiredCanonicalNames(). That gives you three things at once - zero extra String allocations (canonicalName is already computed and already canonical, so no second canonicalize call is needed), zero hash lookups on the common path (the isEmpty() check short-circuits whenever null handling is on), and the gate only fires for names that actually are Pinot operators. It also means TransformFunctionType can expose the pre-canonicalized set directly instead of canonicalizing per call; keep the existing requiresNullHandling(String) for any caller that has a raw name. ########## pinot-core/src/main/java/org/apache/pinot/core/plan/DistinctPlanNode.java: ########## @@ -54,6 +55,18 @@ public DistinctPlanNode(SegmentContext segmentContext, QueryContext queryContext @Override public Operator<DistinctResultsBlock> run() { List<ExpressionContext> expressions = _queryContext.getSelectExpressions(); + for (ExpressionContext expression : expressions) { + String column = expression.getIdentifier(); + if (column != null) { + DataType dataType = _indexSegment.getDataSource(column, _queryContext.getSchema()) Review Comment: Perf: this adds an _indexSegment.getDataSource(column, schema) call per select expression per segment on every DISTINCT query, ahead of the dictionary fast path - paid by every table whether or not a VARIANT column exists anywhere. AggregationFunctionUtils.validateRawVariantIdentifierInputs does the same on the aggregation path, and buildAggregationInfoWithStarTree now calls it before even attempting the star-tree route. Each call is cheap, but these are two hot per-segment planning paths and the cost is unconditional. A one-shot check on the QueryContext schema for 'any VARIANT column' (which is already at hand here) would let both loops be skipped entirely for the overwhelming majority of tables. Worth a JMH or at least a fan-out sanity check on a high-segment-count table before merge. ########## pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetNativeRecordReader.java: ########## @@ -122,9 +159,11 @@ public GenericRow next(GenericRow reuse) } catch (Exception e) { Review Comment: Good catch moving _currentPageIdx++ ahead of extraction so a continueOnError caller cannot re-read the same row or leave hasNext() stuck at EOF. Two smaller notes on this file and its siblings. (1) The init/rewind/close rewrites across all four readers - publish only the fully initialised replacement, suppress close exceptions into the primary, null out state on close - are a genuine robustness improvement, and independent of VARIANT. Worth their own clearly-labelled commit so they can be backported without the feature. (2) ParquetRecordReader.useAvroParquetRecordReader() now derives from the delegate instance, so it returns false before init() and after close(), where before it defaulted to true and retained its value post-close. In-repo callers are tests only, but it is a public method on a plugin class, so a downstream caller could notice. -- 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]
