xiangfu0 commented on code in PR #19101: URL: https://github.com/apache/pinot/pull/19101#discussion_r4366442436
########## 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: Documented in `docs/design/VARIANT.md` §2.1 in cb407e5149: this PR leaves the existing External Tables OPEN_STRUCT mapping unchanged; native ingestion uses an explicitly declared VARIANT destination, not a global default. For another destination, use `variantToJson(source)` for JSON or `jsonStringToMap(variantToJson(source))` for OPEN_STRUCT; the passing reader test creates and reads a segment with both mappings, and the docs explain the native type-tag loss through JSON. Changing an existing column's type is not an in-place toggle: use a new column/table and rebuild/reingest before switching queries. ########## 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: Added the conversion/precision table in `docs/design/VARIANT.md` §7.1 (cb407e5149): extraction without a target retains VARIANT; DECIMAL → BIG_DECIMAL is exact, FLOAT/DOUBLE round, and INT/LONG require an exact integral value in range. TIMESTAMP floors micros/nanos to milliseconds and loses sub-millisecond precision and TZ/NTZ distinction without a session-zone adjustment; temporal values cannot use LONG/STRING extraction as a workaround, and TIME has no typed target. UUID → UUID and binary → BYTES preserve bytes; keep VARIANT for native-type fidelity, or render JSON for inspection. Ingestion transforms use these same conversions and should match the destination type. ########## 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: Added `testNestedShreddedArraysAndIntermediateResidualObjects` in cb407e5149 with a real Parquet file covering all four shapes: further-shredded object elements, arrays within arrays, value-only elements, and intermediate residual objects. It includes empty arrays/null elements and actual object keys named `value` and `typed_value`; replay now distinguishes those keys from structural typed-value containers. All 23 `ParquetVariantRecordReaderTest` cases passed in the 143-test focused reactor. ########## 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: Fixed in cb407e5149: when a shredded typed-value object is present, reject residual keys declared in its shredded schema, including nested objects and absent child values, instead of inheriting parquet-java's silent preference; the raw path with an entirely absent typed parent remains unchanged. Corrected the old positive fixture containing illegal overlap and added `testDuplicateShreddedKeysFailRowAndReaderRecovers`, which passed for root/nested overlaps and recovery to subsequent valid rows. This is a row extraction error after physical advancement; the ingestion caller's continue-on-error policy chooses stop versus skip, not value substitution or null. -- 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]
