raghavyadav01 commented on code in PR #19101: URL: https://github.com/apache/pinot/pull/19101#discussion_r4328340409
########## pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetVariantConverter.java: ########## @@ -0,0 +1,324 @@ +/** + * 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.plugin.inputformat.parquet; + +import java.nio.ByteBuffer; +import java.util.HashSet; +import java.util.Set; +import java.util.function.Consumer; +import java.util.function.Predicate; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.io.api.Converter; +import org.apache.parquet.io.api.GroupConverter; +import org.apache.parquet.io.api.PrimitiveConverter; +import org.apache.parquet.schema.GroupType; +import org.apache.parquet.schema.LogicalTypeAnnotation; +import org.apache.parquet.schema.PrimitiveType; +import org.apache.parquet.schema.Type; +import org.apache.parquet.variant.ImmutableMetadata; +import org.apache.parquet.variant.Variant; +import org.apache.parquet.variant.VariantBuilder; +import org.apache.parquet.variant.VariantConverters; +import org.apache.pinot.spi.utils.VariantEnvelope; + + +/// Converts a top-level Parquet `VARIANT(1)` group into Pinot's self-describing VARIANT byte envelope. +/// +/// Detection is based only on [LogicalTypeAnnotation.VariantLogicalTypeAnnotation], never on field names. +/// The parquet-java [VariantConverters] implementation performs the logical reconstruction so unshredded, +/// shredded, and partially shredded encodings all produce complete Variant metadata and value buffers. +/// +/// Instances retain parquet-java's converter tree and are intentionally scoped to one record reader because the +/// converter and builder holder are mutable and not thread-safe. +final class ParquetVariantConverter { + private static final byte SUPPORTED_SPEC_VERSION = 1; + + private final GroupType _variantType; + private final int _metadataIndex; + private final int _valueIndex; + private final int _typedValueIndex; + private final BuilderHolder _holder = new BuilderHolder(); + private final GroupConverter _converter; + + private ParquetVariantConverter(GroupType variantType) { + _variantType = variantType; + _metadataIndex = variantType.getFieldIndex("metadata"); + _valueIndex = variantType.containsField("value") ? variantType.getFieldIndex("value") : -1; + _typedValueIndex = variantType.containsField("typed_value") ? variantType.getFieldIndex("typed_value") : -1; + _converter = VariantConverters.newVariantConverter(variantType, _holder::setMetadata, _holder::build); + } + + /// Validates every VARIANT annotation in the file and returns the supported top-level field names. + /// + /// Pinot currently supports only non-repeated, top-level `VARIANT(1)` values. Failing during reader + /// initialization avoids silently surfacing an unsupported nested/repeated Variant as an ordinary struct. + static Set<String> validateAndGetTopLevelVariantFields(GroupType schema) { + return validateAndGetTopLevelVariantFields(schema, fieldName -> true); + } + + private static Set<String> validateAndGetTopLevelVariantFields(GroupType schema, Predicate<String> fieldSelector) { + Set<String> variantFields = new HashSet<>(); + for (Type field : schema.getFields()) { + if (!fieldSelector.test(field.getName())) { + continue; + } + if (isVariant(field)) { + validateTopLevelVariant(field); + variantFields.add(field.getName()); + } else if (!field.isPrimitive()) { + rejectNestedVariants(field.asGroupType(), field.getName()); + } + } + return Set.copyOf(variantFields); + } + + /// Validates the file schema and builds an index-aligned reusable converter tree for each top-level VARIANT column. + static ParquetVariantConverter[] createTopLevelVariantConverters(GroupType schema) { + Set<String> variantFields = validateAndGetTopLevelVariantFields(schema); + return buildTopLevelVariantConverters(schema, variantFields); + } + + /// Validates selected top-level fields and builds index-aligned converters for selected VARIANT columns. + static ParquetVariantConverter[] createTopLevelVariantConverters(GroupType schema, Predicate<String> fieldSelector) { + Set<String> variantFields = validateAndGetTopLevelVariantFields(schema, fieldSelector); + return buildTopLevelVariantConverters(schema, variantFields); + } + + private static ParquetVariantConverter[] buildTopLevelVariantConverters(GroupType schema, + Set<String> variantFields) { + ParquetVariantConverter[] variantConverters = new ParquetVariantConverter[schema.getFieldCount()]; + for (int fieldIndex = 0; fieldIndex < schema.getFieldCount(); fieldIndex++) { + Type field = schema.getType(fieldIndex); + if (variantFields.contains(field.getName())) { + variantConverters[fieldIndex] = new ParquetVariantConverter(field.asGroupType()); + } + } + return variantConverters; + } + + static boolean isVariant(Type type) { + return type.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.VariantLogicalTypeAnnotation; + } + + byte[] convert(Group group) { + if (_valueIndex >= 0 && group.getFieldRepetitionCount(_valueIndex) == 1 + && (_typedValueIndex < 0 || group.getFieldRepetitionCount(_typedValueIndex) == 0)) { + return pack(group.getBinary(_metadataIndex, 0), group.getBinary(_valueIndex, 0)); + } + _holder.reset(); + feedVariantGroup(group, _converter, _variantType, _metadataIndex); + Variant variant = _holder.finish(); + return VariantEnvelope.encode(variant.getMetadataBuffer(), variant.getValueBuffer()); + } + + static byte[] pack(ByteBuffer metadata, ByteBuffer value) { + return VariantEnvelope.encode(metadata, value); + } + + private byte[] pack(Binary metadata, Binary value) { + // Binary.writeTo(OutputStream) materializes the full payload for a direct or read-only ByteBuffer-backed Binary. + // Consume the independent views returned by toByteBuffer() so both payloads go straight into the final envelope, + // without intermediate byte arrays or another duplicate view. + int metadataLength = metadata.length(); + int valueLength = value.length(); + byte[] envelope = VariantEnvelope.allocateUnfilled(metadataLength, valueLength); + copyInto(metadata, envelope, VariantEnvelope.HEADER_SIZE, metadataLength, "metadata"); + copyInto(value, envelope, VariantEnvelope.HEADER_SIZE + metadataLength, valueLength, "value"); + return envelope; + } + + private static void copyInto(Binary source, byte[] target, int targetOffset, int expectedLength, String payload) { + ByteBuffer sourceView = source.toByteBuffer(); + if (sourceView.remaining() != expectedLength) { + throw new IllegalStateException( + "Parquet VARIANT " + payload + " length changed while copying: expected " + expectedLength + " but found " + + sourceView.remaining()); + } + if (sourceView.hasArray()) { + System.arraycopy(sourceView.array(), sourceView.arrayOffset() + sourceView.position(), target, targetOffset, + expectedLength); + } else { + sourceView.get(target, targetOffset, expectedLength); + } + } + + private static void validateTopLevelVariant(Type field) { + if (!isVariant(field) || field.isPrimitive()) { + throw new IllegalArgumentException("Parquet VARIANT must be a group: " + field); + } + LogicalTypeAnnotation.VariantLogicalTypeAnnotation annotation = + (LogicalTypeAnnotation.VariantLogicalTypeAnnotation) field.getLogicalTypeAnnotation(); + if (annotation.getSpecVersion() != SUPPORTED_SPEC_VERSION) { + throw new UnsupportedOperationException( + "Unsupported Parquet VARIANT spec version: " + annotation.getSpecVersion()); + } + if (field.isRepetition(Type.Repetition.REPEATED)) { + throw new UnsupportedOperationException("Repeated Parquet VARIANT is not supported: " + field.getName()); + } + + GroupType group = field.asGroupType(); + for (Type child : group.getFields()) { + String name = child.getName(); + if (!"metadata".equals(name) && !"value".equals(name) && !"typed_value".equals(name)) { + throw new IllegalArgumentException("Invalid Parquet VARIANT field: " + child); Review Comment: What happens on a spec-illegal file, e.g. a key present in both `typed_value` and the residual `value`? Does it fail the row, fail the file, or pick one silently? ########## docs/design/VARIANT.md: ########## @@ -0,0 +1,355 @@ +<!-- + + 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 + +Status: draft for community and component-owner review. + +## 1. Motivation and goals + +Apache Parquet `VARIANT(1)` provides a portable binary representation for +semi-structured values. Pinot should ingest that representation without converting +every row through a JSON tree, retain the complete value for late-bound queries, and +let users materialize frequently queried paths into ordinary Pinot columns. + +This design has five goals: + +1. Define a first-class Pinot `VARIANT` logical type with an explicit persisted and + query-wire contract. +2. Ingest top-level, non-repeated Parquet `VARIANT(1)` values, including unshredded, + fully shredded, and partially shredded layouts. +3. Provide Spark-style parse, extraction, existence, type, null, and JSON-rendering + functions in both Pinot query engines. +4. Fail explicitly when a raw Variant value reaches an operation whose semantics are + undefined, while allowing users to extract a typed scalar for that operation. +5. Ship an executable table-definition, ingestion, and query story with integration + coverage. + +### 1.1 Non-goals + +The initial contract does not include: + +- nested or repeated Parquet Variant columns; +- Pinot multi-value `VARIANT` columns; +- streaming-format Variant ingestion; +- quoted object keys in Variant paths; +- semantic equality, hashing, or ordering of raw Variant values; +- mixed-version execution for queries that contain a Variant type; or +- automatic rollback after a Variant table has been activated. + +## 2. User model + +A Variant column is declared as a single-value dimension: + +```json +{ + "name": "payload", + "dataType": "VARIANT" +} +``` + +The table stores `payload` in a raw forward index without a dictionary. Storage null +handling must be enabled. A common layout retains the source value while materializing +an indexed scalar during ingestion: + +```json +{ + "columnName": "eventType", + "transformFunction": "variant_get(payload, '$.eventType', 'STRING')" +} +``` + +Late-bound paths remain queryable: + +```sql +SET enableNullHandling=true; + +SELECT + eventId, + variant_get(payload, '$.user.id', 'STRING') AS userId, + variant_get(payload, '$.amount', 'DOUBLE') AS amount +FROM variantEvents +WHERE eventType = 'checkout'; +``` + +## 3. Data flow and ownership + +```text +Parquet VARIANT(1) + | + | pinot-parquet: validate shape and reconstruct shredded values + v +metadata ByteBuffer + value ByteBuffer + | + | VariantEnvelope: add stable Pinot PVAR framing + v +single-value raw BYTES forward index in a Pinot segment + | + | DataSchema/protobuf: carry logical VARIANT, not generic BYTES + v +single-stage or multi-stage Variant functions + | + +--> typed scalar for filtering/grouping/joining/aggregation + | + +--> canonical JSON text for response clients +``` + +Ownership is deliberately split by module: + +- `pinot-spi` owns the public logical type, capability contract, and stable `PVAR` + envelope. +- `pinot-segment-local` owns schema, table, index, null, and segment-metadata + validation. +- `pinot-parquet` owns Parquet logical-type validation and reconstruction. +- `pinot-common`, `pinot-core`, `pinot-query-planner`, and `pinot-query-runtime` own + function semantics and defensive query guards. +- client, response, and DDL modules carry the logical type without redefining its + semantics. + +`pinot-common` depends on `parquet-variant` and its required `parquet-common` and +`parquet-column` codec closure; the unused `parquet-jackson` artifact is excluded. +Hadoop and record-reader integrations remain isolated to the input-format plugin. + +## 4. Persisted `PVAR` envelope + +Pinot segments need one byte sequence that contains both buffers in a Parquet Variant +value. `VariantEnvelope` owns that framing. Version 1 is: + +```text +offset size field +0 4 ASCII magic "PVAR" +4 1 envelope version (1) +5 1 flags (0) +6 2 reserved (0) +8 4 metadata length, big-endian signed int restricted to >= 0 +12 4 value length, big-endian signed int restricted to >= 0 +16 M Parquet Variant metadata bytes +16 + M V Parquet Variant value bytes +``` + +The complete envelope must fit in a Java byte array. Decoders reject unknown versions, +non-zero flags or reserved bytes, negative lengths, bad magic, and length mismatches. +Metadata and value bytes are copied without interpretation by the envelope layer; the +Parquet Variant codec validates their internal representation. + +Version 1 is immutable and protected by a golden-byte test. A future incompatible +framing change must use a new envelope version and retain an explicit reader for every +supported old version. Unknown versions fail closed; they must never fall back to +generic `BYTES`. + +`decode(byte[])` returns zero-copy read-only views that alias the input array. The input +must therefore remain immutable for the lifetime of those views. Array, direct, and +read-only source buffers are supported without changing their positions. Parquet +ingestion copies payload bytes directly into the final envelope and must not allocate a +full intermediate payload array. + +## 5. Logical type and segment contract + +`FieldSpec.DataType.VARIANT` is logically distinct from `BYTES` but uses variable-width +bytes as its physical stored type. + +The initial schema and table constraints are: + +- dimension field only; +- single-value only; +- the reserved empty-byte default null value only; +- no truncating or default-substitution max-length policy; +- storage null handling enabled at schema or table level; +- enabled raw forward index; +- no dictionary or secondary index on the raw Variant column; +- not a primary key, partition column, sorted column, upsert comparison column, + metrics-aggregation key, or star-tree dimension; +- partial upsert may use only the `OVERWRITE` strategy; and +- minion merge tasks (`MergeRollupTask`, `RealtimeToOfflineSegmentsTask`) may use only the `CONCAT` merge type, + because rollup and dedup group rows by raw stored bytes. + +Segment creation validates every non-null value as a complete `PVAR` envelope. Raw +Variant columns do not publish logical minimum or maximum values and are never marked +sorted, including single-row segments. + +### 5.1 Capability policy + +The Parquet representation is not a canonical semantic encoding: equivalent logical +objects can use different metadata dictionaries or shredded layouts. Byte equality, +byte hashing, and lexicographic byte order would therefore expose storage-layout +accidents as SQL semantics. + +The logical type reports: + +| Capability on raw value | Supported | +|---|---| +| equality / `IN` / pattern predicates | no | +| hashing / `GROUP BY` / `DISTINCT` | no | +| ordering / min-max / sort / ASOF match key | no | +| join, lookup, primary, or partition key | no | +| direct value-consuming aggregation | no | +| `CAST` from a raw value | no | +| non-distinct `COUNT(raw_variant)` | yes | + +Planner validation provides the primary error. Runtime and single-stage guards remain +in place for mixed plans or future planner changes. Error messages direct users to +`variant_get` a typed path first. + +## 6. Parquet ingestion + +Reader initialization scans the file schema before publishing reader state. + +- A supported field is a top-level, non-repeated group annotated `VARIANT(1)`. +- `metadata` is required binary; at least one of `value` or `typed_value` is present. +- The parquet-java `VariantConverters` tree reconstructs fully or partially shredded + values. +- The unshredded path retains the exact metadata and value buffers and adds only the + `PVAR` envelope. +- Converter plans and field indexes are immutable and created once per reader. +- A failed reinitialization leaves the previous reader usable. A failure while closing + the old reader does not invalidate the successfully initialized replacement. + +Automatic reader selection preserves existing Avro-metadata precedence for backward +compatibility. A Parquet file with Avro metadata must explicitly select the native +reader to ingest Variant. + +Malformed rows follow the existing record-reader skip policy and do not corrupt reader +progress. Unsupported Variant spec versions and unsupported nested/repeated shapes fail +during initialization rather than appearing as ordinary structs. + +## 7. Query functions + +Both query engines expose: + +- `variant_get(value, path[, targetType])` +- `try_variant_get(value, path[, targetType])` +- `variant_exists(value, path)` +- `is_variant_null(value[, path])` +- `variant_typeof(value[, path])` +- `variant_to_json(value)` +- `parse_json(text)` / `parse_json_to_variant(text)` +- `try_parse_json(text)` / `try_parse_json_to_variant(text)` + +Function lookup remains case-insensitive under Pinot's existing registration rules. +The v1 path grammar supports root `$`, dot-separated object fields, and non-negative +array indexes. Supported extraction targets are `BOOLEAN`, `INT`, `LONG`, `FLOAT`, +`DOUBLE`, `BIG_DECIMAL`, `STRING`, `BYTES`, `UUID`, `TIMESTAMP`, `VARIANT`, and `JSON`. Review Comment: How do Variant primitives map to Pinot types in `variant_get` and materialization? Mainly decimals (BIG_DECIMAL vs DOUBLE), timestamp and timestamp_ntz (micros into Pinot's millis loses precision), time, uuid and binary. ########## docs/design/VARIANT.md: ########## @@ -0,0 +1,355 @@ +<!-- + + 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 + +Status: draft for community and component-owner review. + +## 1. Motivation and goals + +Apache Parquet `VARIANT(1)` provides a portable binary representation for +semi-structured values. Pinot should ingest that representation without converting +every row through a JSON tree, retain the complete value for late-bound queries, and +let users materialize frequently queried paths into ordinary Pinot columns. + +This design has five goals: + +1. Define a first-class Pinot `VARIANT` logical type with an explicit persisted and + query-wire contract. +2. Ingest top-level, non-repeated Parquet `VARIANT(1)` values, including unshredded, + fully shredded, and partially shredded layouts. +3. Provide Spark-style parse, extraction, existence, type, null, and JSON-rendering + functions in both Pinot query engines. +4. Fail explicitly when a raw Variant value reaches an operation whose semantics are + undefined, while allowing users to extract a typed scalar for that operation. +5. Ship an executable table-definition, ingestion, and query story with integration + coverage. + +### 1.1 Non-goals + +The initial contract does not include: + +- nested or repeated Parquet Variant columns; +- Pinot multi-value `VARIANT` columns; +- streaming-format Variant ingestion; +- quoted object keys in Variant paths; +- semantic equality, hashing, or ordering of raw Variant values; +- mixed-version execution for queries that contain a Variant type; or +- automatic rollback after a Variant table has been activated. + +## 2. User model Review Comment: External Tables already read Parquet VARIANT as OPEN_STRUCT, where each shredded key becomes its own typed, indexable child. This adds a second mapping as an opaque VARIANT value. Which should a Parquet VARIANT column map to by default, and can a table switch between them later? ########## pinot-plugins/pinot-input-format/pinot-parquet/src/main/java/org/apache/pinot/plugin/inputformat/parquet/ParquetVariantConverter.java: ########## @@ -0,0 +1,324 @@ +/** + * 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.plugin.inputformat.parquet; + +import java.nio.ByteBuffer; +import java.util.HashSet; +import java.util.Set; +import java.util.function.Consumer; +import java.util.function.Predicate; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.io.api.Converter; +import org.apache.parquet.io.api.GroupConverter; +import org.apache.parquet.io.api.PrimitiveConverter; +import org.apache.parquet.schema.GroupType; +import org.apache.parquet.schema.LogicalTypeAnnotation; +import org.apache.parquet.schema.PrimitiveType; +import org.apache.parquet.schema.Type; +import org.apache.parquet.variant.ImmutableMetadata; +import org.apache.parquet.variant.Variant; +import org.apache.parquet.variant.VariantBuilder; +import org.apache.parquet.variant.VariantConverters; +import org.apache.pinot.spi.utils.VariantEnvelope; + + +/// Converts a top-level Parquet `VARIANT(1)` group into Pinot's self-describing VARIANT byte envelope. +/// +/// Detection is based only on [LogicalTypeAnnotation.VariantLogicalTypeAnnotation], never on field names. +/// The parquet-java [VariantConverters] implementation performs the logical reconstruction so unshredded, Review Comment: Going through `VariantConverters` is a good call. Do the tests cover the shapes that bit us on External Tables: element fields shredded further, arrays inside arrays, element groups with `value` but no `typed_value`, and residual `value` at intermediate levels? Our random generator passed 4000 shapes and still missed 8 real bugs because it never produced those. ########## docs/design/VARIANT.md: ########## @@ -0,0 +1,355 @@ +<!-- + + 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 + +Status: draft for community and component-owner review. + +## 1. Motivation and goals + +Apache Parquet `VARIANT(1)` provides a portable binary representation for +semi-structured values. Pinot should ingest that representation without converting +every row through a JSON tree, retain the complete value for late-bound queries, and +let users materialize frequently queried paths into ordinary Pinot columns. + +This design has five goals: + +1. Define a first-class Pinot `VARIANT` logical type with an explicit persisted and + query-wire contract. +2. Ingest top-level, non-repeated Parquet `VARIANT(1)` values, including unshredded, + fully shredded, and partially shredded layouts. +3. Provide Spark-style parse, extraction, existence, type, null, and JSON-rendering + functions in both Pinot query engines. +4. Fail explicitly when a raw Variant value reaches an operation whose semantics are + undefined, while allowing users to extract a typed scalar for that operation. +5. Ship an executable table-definition, ingestion, and query story with integration + coverage. + +### 1.1 Non-goals + +The initial contract does not include: + +- nested or repeated Parquet Variant columns; +- Pinot multi-value `VARIANT` columns; +- streaming-format Variant ingestion; +- quoted object keys in Variant paths; +- semantic equality, hashing, or ordering of raw Variant values; +- mixed-version execution for queries that contain a Variant type; or +- automatic rollback after a Variant table has been activated. + +## 2. User model + +A Variant column is declared as a single-value dimension: + +```json +{ + "name": "payload", + "dataType": "VARIANT" +} +``` + +The table stores `payload` in a raw forward index without a dictionary. Storage null +handling must be enabled. A common layout retains the source value while materializing +an indexed scalar during ingestion: + +```json +{ + "columnName": "eventType", + "transformFunction": "variant_get(payload, '$.eventType', 'STRING')" +} +``` + +Late-bound paths remain queryable: + +```sql +SET enableNullHandling=true; + +SELECT + eventId, + variant_get(payload, '$.user.id', 'STRING') AS userId, + variant_get(payload, '$.amount', 'DOUBLE') AS amount +FROM variantEvents +WHERE eventType = 'checkout'; +``` + +## 3. Data flow and ownership + +```text +Parquet VARIANT(1) + | + | pinot-parquet: validate shape and reconstruct shredded values + v +metadata ByteBuffer + value ByteBuffer + | + | VariantEnvelope: add stable Pinot PVAR framing + v +single-value raw BYTES forward index in a Pinot segment + | + | DataSchema/protobuf: carry logical VARIANT, not generic BYTES + v +single-stage or multi-stage Variant functions + | + +--> typed scalar for filtering/grouping/joining/aggregation + | + +--> canonical JSON text for response clients +``` + +Ownership is deliberately split by module: + +- `pinot-spi` owns the public logical type, capability contract, and stable `PVAR` + envelope. +- `pinot-segment-local` owns schema, table, index, null, and segment-metadata + validation. +- `pinot-parquet` owns Parquet logical-type validation and reconstruction. +- `pinot-common`, `pinot-core`, `pinot-query-planner`, and `pinot-query-runtime` own + function semantics and defensive query guards. +- client, response, and DDL modules carry the logical type without redefining its + semantics. + +`pinot-common` depends on `parquet-variant` and its required `parquet-common` and +`parquet-column` codec closure; the unused `parquet-jackson` artifact is excluded. +Hadoop and record-reader integrations remain isolated to the input-format plugin. + +## 4. Persisted `PVAR` envelope + +Pinot segments need one byte sequence that contains both buffers in a Parquet Variant +value. `VariantEnvelope` owns that framing. Version 1 is: + +```text +offset size field +0 4 ASCII magic "PVAR" +4 1 envelope version (1) +5 1 flags (0) +6 2 reserved (0) +8 4 metadata length, big-endian signed int restricted to >= 0 +12 4 value length, big-endian signed int restricted to >= 0 +16 M Parquet Variant metadata bytes +16 + M V Parquet Variant value bytes +``` + +The complete envelope must fit in a Java byte array. Decoders reject unknown versions, +non-zero flags or reserved bytes, negative lengths, bad magic, and length mismatches. +Metadata and value bytes are copied without interpretation by the envelope layer; the +Parquet Variant codec validates their internal representation. + +Version 1 is immutable and protected by a golden-byte test. A future incompatible +framing change must use a new envelope version and retain an explicit reader for every +supported old version. Unknown versions fail closed; they must never fall back to +generic `BYTES`. + +`decode(byte[])` returns zero-copy read-only views that alias the input array. The input +must therefore remain immutable for the lifetime of those views. Array, direct, and +read-only source buffers are supported without changing their positions. Parquet +ingestion copies payload bytes directly into the final envelope and must not allocate a +full intermediate payload array. + +## 5. Logical type and segment contract + +`FieldSpec.DataType.VARIANT` is logically distinct from `BYTES` but uses variable-width +bytes as its physical stored type. + +The initial schema and table constraints are: + +- dimension field only; +- single-value only; +- the reserved empty-byte default null value only; +- no truncating or default-substitution max-length policy; +- storage null handling enabled at schema or table level; +- enabled raw forward index; +- no dictionary or secondary index on the raw Variant column; +- not a primary key, partition column, sorted column, upsert comparison column, + metrics-aggregation key, or star-tree dimension; +- partial upsert may use only the `OVERWRITE` strategy; and +- minion merge tasks (`MergeRollupTask`, `RealtimeToOfflineSegmentsTask`) may use only the `CONCAT` merge type, + because rollup and dedup group rows by raw stored bytes. + +Segment creation validates every non-null value as a complete `PVAR` envelope. Raw +Variant columns do not publish logical minimum or maximum values and are never marked +sorted, including single-row segments. + +### 5.1 Capability policy + +The Parquet representation is not a canonical semantic encoding: equivalent logical +objects can use different metadata dictionaries or shredded layouts. Byte equality, +byte hashing, and lexicographic byte order would therefore expose storage-layout +accidents as SQL semantics. + +The logical type reports: + +| Capability on raw value | Supported | +|---|---| +| equality / `IN` / pattern predicates | no | +| hashing / `GROUP BY` / `DISTINCT` | no | +| ordering / min-max / sort / ASOF match key | no | +| join, lookup, primary, or partition key | no | +| direct value-consuming aggregation | no | +| `CAST` from a raw value | no | +| non-distinct `COUNT(raw_variant)` | yes | + +Planner validation provides the primary error. Runtime and single-stage guards remain +in place for mixed plans or future planner changes. Error messages direct users to +`variant_get` a typed path first. + +## 6. Parquet ingestion + +Reader initialization scans the file schema before publishing reader state. + +- A supported field is a top-level, non-repeated group annotated `VARIANT(1)`. +- `metadata` is required binary; at least one of `value` or `typed_value` is present. +- The parquet-java `VariantConverters` tree reconstructs fully or partially shredded + values. +- The unshredded path retains the exact metadata and value buffers and adds only the + `PVAR` envelope. +- Converter plans and field indexes are immutable and created once per reader. +- A failed reinitialization leaves the previous reader usable. A failure while closing + the old reader does not invalidate the successfully initialized replacement. + +Automatic reader selection preserves existing Avro-metadata precedence for backward +compatibility. A Parquet file with Avro metadata must explicitly select the native +reader to ingest Variant. + +Malformed rows follow the existing record-reader skip policy and do not corrupt reader +progress. Unsupported Variant spec versions and unsupported nested/repeated shapes fail +during initialization rather than appearing as ordinary structs. + +## 7. Query functions + +Both query engines expose: + +- `variant_get(value, path[, targetType])` +- `try_variant_get(value, path[, targetType])` +- `variant_exists(value, path)` +- `is_variant_null(value[, path])` +- `variant_typeof(value[, path])` +- `variant_to_json(value)` +- `parse_json(text)` / `parse_json_to_variant(text)` +- `try_parse_json(text)` / `try_parse_json_to_variant(text)` + +Function lookup remains case-insensitive under Pinot's existing registration rules. +The v1 path grammar supports root `$`, dot-separated object fields, and non-negative +array indexes. Supported extraction targets are `BOOLEAN`, `INT`, `LONG`, `FLOAT`, +`DOUBLE`, `BIG_DECIMAL`, `STRING`, `BYTES`, `UUID`, `TIMESTAMP`, `VARIANT`, and `JSON`. + +Neither engine constant-folds the JSON-to-Variant functions: a folded result would +degrade to a plain `BYTES` literal and erase the logical type together with its guards. + +Strict functions reject malformed encodings, paths, incompatible conversions, numeric +overflow, and malformed JSON. `try_` functions return SQL null for those failures. +Literal paths, target types, and JSON inputs are compiled or parsed once per query. +Vectorized execution reuses thread-confined cursors and unboxed result holders. + +The query cursor directly decodes the Parquet Variant binary layout to avoid per-row +codec objects and buffer allocation. Its type tags, length rules, metadata version, +and wide-object threshold must stay synchronized with the pinned `parquet-variant` +dependency. Tests compare the cursor constants with parquet-java's codec constants, +exercise every `Variant.Type`, and reject unrecognized metadata versions and primitive +tags. A parquet-variant upgrade that adds an encoding must update the cursor and these +tests in the same change; unknown encodings fail closed rather than returning a +partially decoded or incorrectly typed result. + +JSON parsing is bounded to 100 nesting levels. Parquet Variant decimal encoding is +bounded to 38 digits of precision and a scale of at most 38 after exact normalization. +Exponent bounds are checked before expansion so hostile inputs cannot request +pathological intermediate allocations. + +### 7.1 Null states + +Storage and query null handling are required because four states must remain distinct: + +| State | Stored bytes / null vector | `variant_exists` | `is_variant_null` | `variant_typeof` | scalar `variant_get` | +|---|---|---:|---:|---|---| +| SQL null | empty bytes, null bit set | SQL null | false | SQL null | SQL null | +| Variant null | non-empty `PVAR`, null bit clear | true | true | `NULL` | SQL null | Review Comment: In parquet-java a Variant null and an unknown primitive type id both decode to null, which we couldn't tell apart. The PR says unknown tags fail explicitly. Does that also hold on ingestion, given reconstruction is parquet-java's? ########## docs/design/VARIANT.md: ########## @@ -0,0 +1,355 @@ +<!-- + + 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 + +Status: draft for community and component-owner review. + +## 1. Motivation and goals + +Apache Parquet `VARIANT(1)` provides a portable binary representation for +semi-structured values. Pinot should ingest that representation without converting +every row through a JSON tree, retain the complete value for late-bound queries, and +let users materialize frequently queried paths into ordinary Pinot columns. + +This design has five goals: + +1. Define a first-class Pinot `VARIANT` logical type with an explicit persisted and + query-wire contract. +2. Ingest top-level, non-repeated Parquet `VARIANT(1)` values, including unshredded, + fully shredded, and partially shredded layouts. +3. Provide Spark-style parse, extraction, existence, type, null, and JSON-rendering + functions in both Pinot query engines. +4. Fail explicitly when a raw Variant value reaches an operation whose semantics are + undefined, while allowing users to extract a typed scalar for that operation. +5. Ship an executable table-definition, ingestion, and query story with integration + coverage. + +### 1.1 Non-goals + +The initial contract does not include: + +- nested or repeated Parquet Variant columns; +- Pinot multi-value `VARIANT` columns; +- streaming-format Variant ingestion; +- quoted object keys in Variant paths; +- semantic equality, hashing, or ordering of raw Variant values; +- mixed-version execution for queries that contain a Variant type; or +- automatic rollback after a Variant table has been activated. + +## 2. User model + +A Variant column is declared as a single-value dimension: + +```json +{ + "name": "payload", + "dataType": "VARIANT" +} +``` + +The table stores `payload` in a raw forward index without a dictionary. Storage null +handling must be enabled. A common layout retains the source value while materializing +an indexed scalar during ingestion: + +```json +{ + "columnName": "eventType", + "transformFunction": "variant_get(payload, '$.eventType', 'STRING')" +} +``` + +Late-bound paths remain queryable: + +```sql +SET enableNullHandling=true; + +SELECT + eventId, + variant_get(payload, '$.user.id', 'STRING') AS userId, + variant_get(payload, '$.amount', 'DOUBLE') AS amount +FROM variantEvents +WHERE eventType = 'checkout'; +``` + +## 3. Data flow and ownership + +```text +Parquet VARIANT(1) + | + | pinot-parquet: validate shape and reconstruct shredded values + v +metadata ByteBuffer + value ByteBuffer + | + | VariantEnvelope: add stable Pinot PVAR framing + v +single-value raw BYTES forward index in a Pinot segment + | + | DataSchema/protobuf: carry logical VARIANT, not generic BYTES + v +single-stage or multi-stage Variant functions + | + +--> typed scalar for filtering/grouping/joining/aggregation + | + +--> canonical JSON text for response clients +``` + +Ownership is deliberately split by module: + +- `pinot-spi` owns the public logical type, capability contract, and stable `PVAR` + envelope. +- `pinot-segment-local` owns schema, table, index, null, and segment-metadata + validation. +- `pinot-parquet` owns Parquet logical-type validation and reconstruction. +- `pinot-common`, `pinot-core`, `pinot-query-planner`, and `pinot-query-runtime` own + function semantics and defensive query guards. +- client, response, and DDL modules carry the logical type without redefining its + semantics. + +`pinot-common` depends on `parquet-variant` and its required `parquet-common` and +`parquet-column` codec closure; the unused `parquet-jackson` artifact is excluded. +Hadoop and record-reader integrations remain isolated to the input-format plugin. + +## 4. Persisted `PVAR` envelope + +Pinot segments need one byte sequence that contains both buffers in a Parquet Variant +value. `VariantEnvelope` owns that framing. Version 1 is: + +```text +offset size field +0 4 ASCII magic "PVAR" +4 1 envelope version (1) +5 1 flags (0) +6 2 reserved (0) +8 4 metadata length, big-endian signed int restricted to >= 0 +12 4 value length, big-endian signed int restricted to >= 0 +16 M Parquet Variant metadata bytes +16 + M V Parquet Variant value bytes +``` + +The complete envelope must fit in a Java byte array. Decoders reject unknown versions, +non-zero flags or reserved bytes, negative lengths, bad magic, and length mismatches. +Metadata and value bytes are copied without interpretation by the envelope layer; the +Parquet Variant codec validates their internal representation. + +Version 1 is immutable and protected by a golden-byte test. A future incompatible +framing change must use a new envelope version and retain an explicit reader for every +supported old version. Unknown versions fail closed; they must never fall back to +generic `BYTES`. + +`decode(byte[])` returns zero-copy read-only views that alias the input array. The input +must therefore remain immutable for the lifetime of those views. Array, direct, and +read-only source buffers are supported without changing their positions. Parquet +ingestion copies payload bytes directly into the final envelope and must not allocate a +full intermediate payload array. + +## 5. Logical type and segment contract + +`FieldSpec.DataType.VARIANT` is logically distinct from `BYTES` but uses variable-width +bytes as its physical stored type. + +The initial schema and table constraints are: + +- dimension field only; +- single-value only; +- the reserved empty-byte default null value only; +- no truncating or default-substitution max-length policy; +- storage null handling enabled at schema or table level; +- enabled raw forward index; +- no dictionary or secondary index on the raw Variant column; Review Comment: With no path index and the JSON index rejected, every filter on a non-materialized path decodes every row. Is a per-path index planned, or is materialization the long-term answer? -- 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]
