You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "swuferhong (via GitHub)" <gi...@apache.org> on 2026/04/02 03:17:51 UTC
[PR] [WIP] Support variant type and shredding [fluss]
swuferhong opened a new pull request, #2981:
URL: https://github.com/apache/fluss/pull/2981
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
- **Generative AI disclosure:** Indicate whether generative AI tools were used in authoring this PR. If yes, specify the tool below.
- [ ] No generative AI tools used
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name and Version] following [the guidelines](https://github.com/apache/fluss/blob/main/AGENTS.md)
-->
**(The sections below can be removed for hotfixes or typos)**
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #xxx
<!-- What is the purpose of the change -->
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] Support variant type and shredding [fluss]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #2981:
URL: https://github.com/apache/fluss/pull/2981#discussion_r4260350981
##########
fluss-common/src/main/java/org/apache/fluss/types/variant/JsonToVariant.java:
##########
@@ -0,0 +1,288 @@
+/*
+ * 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.fluss.types.variant;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.shaded.arrow.org.apache.parquet.variant.VariantArrayBuilder;
+import org.apache.fluss.shaded.arrow.org.apache.parquet.variant.VariantBuilder;
+import org.apache.fluss.shaded.arrow.org.apache.parquet.variant.VariantObjectBuilder;
+
+import java.math.BigDecimal;
+
+/**
+ * Lightweight JSON parser that feeds parquet-variant's {@link VariantBuilder} to construct Variant
+ * binary encoding. This replaces the custom two-pass JSON parser previously in VariantBuilder.
+ *
+ * <p>The parser supports: null, boolean, numbers (integer/decimal/double), strings, objects, and
+ * arrays. Number handling: integers are stored as the smallest fitting type (byte/short/int/long);
+ * decimals without exponent are stored as DECIMAL; numbers with exponent are stored as DOUBLE.
+ */
+@Internal
+final class JsonToVariant {
+
+ private JsonToVariant() {}
+
+ /** Parses the JSON string and appends the result to the given builder. */
+ static void parse(String json, VariantBuilder builder) {
+ int[] pos = {0};
+ parseValue(json, pos, builder);
+ }
+
+ private static void parseValue(String json, int[] pos, VariantBuilder builder) {
Review Comment:
Why aren't we parsing the String with a Jackson? A lot of code would fall
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] Support variant type and shredding [fluss]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #2981:
URL: https://github.com/apache/fluss/pull/2981#discussion_r4260379173
##########
fluss-common/src/main/java/org/apache/fluss/types/variant/ShreddedVariant.java:
##########
@@ -0,0 +1,435 @@
+/*
+ * 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.fluss.types.variant;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.DataTypeRoot;
+
+import javax.annotation.Nullable;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * A lazily-merged Variant that carries shredded field data alongside residual binary data. Instead
+ * of immediately merging typed_value columns back into Variant binary encoding (which is then
+ * re-parsed by downstream consumers), this class defers the merge until {@link #metadata()} or
+ * {@link #value()} is actually called.
+ *
+ * <p>This enables downstream consumers (e.g., Flink's FlussVariant) to directly access shredded
+ * field values as Java objects (String, Long, Boolean, etc.) without any binary encoding/decoding
+ * overhead when querying individual fields.
+ *
+ * <h3>Type Mismatch Handling</h3>
+ *
+ * <p>When a shredded field's {@code typed_value} is null but the {@code fallbackValue} is non-null,
+ * this indicates the actual value's type doesn't match the shredded typed column (e.g., the field
+ * was shredded as INT but the row contains a STRING). In this case:
+ *
+ * <ul>
+ * <li>The fallback binary is used to reconstruct the field value
+ * <li>No null is returned and no exception is thrown
+ * <li>The complete Variant always preserves the original data
+ * </ul>
+ *
+ * <p>Data layout:
+ *
+ * <ul>
+ * <li><b>residualMetadata</b>: metadata dictionary from the residual Variant (may be null if no
+ * residual)
+ * <li><b>residualValue</b>: residual value bytes (may be null if all fields are shredded)
+ * <li><b>fieldNames</b>: parallel array of shredded field names
+ * <li><b>fieldTypes</b>: parallel array of shredded field data types
+ * <li><b>typedValues</b>: parallel array of Java objects (non-null if typed_value matched)
+ * <li><b>fallbackValues</b>: parallel array of Variant binary (non-null if typed_value didn't
+ * match)
+ * </ul>
+ *
+ * @since 0.7
+ */
+@Internal
+public class ShreddedVariant extends Variant {
Review Comment:
Should we break down this PR into multiple ones and keep the Shredding as a separate effort?
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] Support variant type and shredding [fluss]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #2981:
URL: https://github.com/apache/fluss/pull/2981#discussion_r4260273768
##########
fluss-common/src/main/java/org/apache/fluss/types/variant/Variant.java:
##########
@@ -0,0 +1,594 @@
+/*
+ * 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.fluss.types.variant;
+
+import org.apache.fluss.annotation.PublicEvolving;
+import org.apache.fluss.shaded.arrow.org.apache.parquet.variant.VariantBuilder;
+
+import java.io.Serializable;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant.Type;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/**
+ * Represents a semi-structured variant value, stored using the Parquet Variant Binary Encoding
+ * format. A Variant consists of two byte arrays:
+ *
+ * <ul>
+ * <li><b>metadata</b>: A dictionary of field names used by object values. Contains a header byte,
+ * dictionary size, offsets, and UTF-8 encoded string bytes.
+ * <li><b>value</b>: A self-describing binary-encoded value. The first byte encodes the basic type
+ * (Primitive, ShortString, Object, Array) and type-specific header information.
+ * </ul>
+ *
+ * <p>For storage in Arrow {@code VarBinaryVector}, the two arrays are concatenated as: {@code
+ * [4-byte metadata length (little-endian)][metadata bytes][value bytes]}.
+ *
+ * <h3>Read path: hybrid Arrow delegate + Fluss {@link VariantUtil}</h3>
+ *
+ * <p>This class deliberately uses two read paths:
+ *
+ * <ul>
+ * <li><b>High-level accessors</b> (typed getters such as {@link #getBoolean()}, {@link
+ * #getInt()}, {@link #getString()}, {@link #getFieldByName(String)}, {@link
+ * #getElementAt(int)}, {@link #toJson()}) are delegated to the Arrow shaded {@code
+ * parquet-variant} {@code Variant} class via {@link #getDelegate()}, so we share its
+ * battle-tested type dispatch.
+ * <li><b>Low-level byte/header inspection</b> (such as {@link #basicType()}, {@link #getDate()},
+ * {@link #getTimestamp()}, {@code encodeXxx} factories, and {@link #emptyMetadata()}) goes
+ * through Fluss {@link VariantUtil}. This is forced by the shaded Arrow library exposing its
+ * own {@code VariantUtil} as package-private (see the javadoc on {@link VariantUtil}). For
+ * these helpers we cannot call into Arrow from outside its package, so {@link VariantUtil} is
+ * a Fluss-side mirror of the Parquet Variant Binary Encoding spec.
+ * </ul>
+ *
+ * @since 0.7
+ */
+@PublicEvolving
+public class Variant implements Serializable {
Review Comment:
We should add Variant to `TableDescriptorValidation`. VARIANT cannot/should not be defined as PRIMARY KEY
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] Support variant type and shredding [fluss]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #2981:
URL: https://github.com/apache/fluss/pull/2981#discussion_r4260494767
##########
fluss-common/src/main/java/org/apache/fluss/types/variant/Variant.java:
##########
@@ -0,0 +1,594 @@
+/*
+ * 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.fluss.types.variant;
+
+import org.apache.fluss.annotation.PublicEvolving;
+import org.apache.fluss.shaded.arrow.org.apache.parquet.variant.VariantBuilder;
+
+import java.io.Serializable;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant.Type;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/**
+ * Represents a semi-structured variant value, stored using the Parquet Variant Binary Encoding
+ * format. A Variant consists of two byte arrays:
+ *
+ * <ul>
+ * <li><b>metadata</b>: A dictionary of field names used by object values. Contains a header byte,
+ * dictionary size, offsets, and UTF-8 encoded string bytes.
+ * <li><b>value</b>: A self-describing binary-encoded value. The first byte encodes the basic type
+ * (Primitive, ShortString, Object, Array) and type-specific header information.
+ * </ul>
+ *
+ * <p>For storage in Arrow {@code VarBinaryVector}, the two arrays are concatenated as: {@code
+ * [4-byte metadata length (little-endian)][metadata bytes][value bytes]}.
+ *
+ * <h3>Read path: hybrid Arrow delegate + Fluss {@link VariantUtil}</h3>
+ *
+ * <p>This class deliberately uses two read paths:
+ *
+ * <ul>
+ * <li><b>High-level accessors</b> (typed getters such as {@link #getBoolean()}, {@link
+ * #getInt()}, {@link #getString()}, {@link #getFieldByName(String)}, {@link
+ * #getElementAt(int)}, {@link #toJson()}) are delegated to the Arrow shaded {@code
+ * parquet-variant} {@code Variant} class via {@link #getDelegate()}, so we share its
+ * battle-tested type dispatch.
+ * <li><b>Low-level byte/header inspection</b> (such as {@link #basicType()}, {@link #getDate()},
+ * {@link #getTimestamp()}, {@code encodeXxx} factories, and {@link #emptyMetadata()}) goes
+ * through Fluss {@link VariantUtil}. This is forced by the shaded Arrow library exposing its
+ * own {@code VariantUtil} as package-private (see the javadoc on {@link VariantUtil}). For
+ * these helpers we cannot call into Arrow from outside its package, so {@link VariantUtil} is
+ * a Fluss-side mirror of the Parquet Variant Binary Encoding spec.
+ * </ul>
+ *
+ * @since 0.7
+ */
+@PublicEvolving
+public class Variant implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final byte[] metadata;
+ private final byte[] value;
+
+ public Variant(byte[] metadata, byte[] value) {
+ this.metadata = checkNotNull(metadata, "metadata must not be null");
+ this.value = checkNotNull(value, "value must not be null");
+ }
+
+ /**
+ * Protected constructor for subclasses that override {@link #metadata()} and {@link #value()}
+ * to provide lazy initialization. All public methods in this class access data through the
+ * accessor methods, so subclasses can safely override them.
+ */
+ protected Variant() {
+ this.metadata = null;
+ this.value = null;
+ }
+
+ /** Returns the metadata bytes (field name dictionary). */
+ public byte[] metadata() {
+ return metadata;
+ }
+
+ /** Returns the value bytes (self-describing encoded value). */
+ public byte[] value() {
+ return value;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Binary serialization (for Arrow VarBinaryVector storage)
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Serializes this variant to a single byte array for storage.
+ *
+ * <p>Format: {@code [4-byte metadata length (little-endian)][metadata bytes][value bytes]}
+ */
+ public byte[] toBytes() {
+ byte[] m = metadata();
+ byte[] v = value();
+ byte[] result = new byte[4 + m.length + v.length];
+ ByteBuffer buf = ByteBuffer.wrap(result).order(ByteOrder.LITTLE_ENDIAN);
+ buf.putInt(m.length);
+ buf.put(m);
+ buf.put(v);
+ return result;
+ }
+
+ /**
+ * Deserializes a variant from a byte array.
+ *
+ * @param bytes the serialized bytes
+ * @return the deserialized Variant
+ */
+ public static Variant fromBytes(byte[] bytes) {
+ return fromBytes(bytes, 0, bytes.length);
+ }
+
+ /**
+ * Deserializes a variant from a byte array with offset and length.
+ *
+ * @param bytes the byte array containing the serialized variant
+ * @param offset the starting offset
+ * @param length the total length
+ * @return the deserialized Variant
+ */
+ public static Variant fromBytes(byte[] bytes, int offset, int length) {
+ ByteBuffer buf = ByteBuffer.wrap(bytes, offset, length).order(ByteOrder.LITTLE_ENDIAN);
+ int metadataLen = buf.getInt();
+ byte[] meta = new byte[metadataLen];
+ buf.get(meta);
+ byte[] val = new byte[length - 4 - metadataLen];
+ buf.get(val);
+ return new Variant(meta, val);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Value type inspection
+ // --------------------------------------------------------------------------------------------
+
+ /** Returns the basic type of this variant value. */
+ public int basicType() {
+ return VariantUtil.basicType(value(), 0);
+ }
+
+ /** Returns true if this variant value is null. */
+ public boolean isNull() {
+ // Delegate to the Arrow shaded Variant for the typed null check (Type.NULL).
+ return getDelegate().getType() == Type.NULL;
+ }
+
+ /** Returns true if this variant is a primitive type. */
+ public boolean isPrimitive() {
+ return basicType() == VariantUtil.BASIC_TYPE_PRIMITIVE;
+ }
+
+ /** Returns true if this variant is a short string. */
+ public boolean isShortString() {
+ return basicType() == VariantUtil.BASIC_TYPE_SHORT_STRING;
+ }
+
+ /** Returns true if this variant is an object. */
+ public boolean isObject() {
+ return basicType() == VariantUtil.BASIC_TYPE_OBJECT;
+ }
+
+ /** Returns true if this variant is an array. */
+ public boolean isArray() {
+ return basicType() == VariantUtil.BASIC_TYPE_ARRAY;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Parquet-variant delegate (lazy)
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Lazily created delegate to parquet-variant's Variant for read operations. This avoids
+ * duplicating the Parquet Variant Binary Encoding read logic.
+ */
+ private transient org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant delegate;
+
+ /**
+ * Returns the parquet-variant delegate, creating it lazily from metadata()/value() bytes. This
+ * is used by accessor methods to delegate read operations.
+ */
+ protected org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant getDelegate() {
+ if (delegate == null) {
+ delegate =
+ new org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant(
+ value(), metadata());
+ }
+ return delegate;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Primitive value accessors (delegated to parquet-variant)
+ // --------------------------------------------------------------------------------------------
+
+ /** Returns the boolean value of this variant. */
+ public boolean getBoolean() {
+ return getDelegate().getBoolean();
+ }
+
+ /** Returns the byte value of this variant. */
+ public byte getByte() {
+ return getDelegate().getByte();
+ }
+
+ /** Returns the short value of this variant. */
+ public short getShort() {
+ return getDelegate().getShort();
+ }
+
+ /** Returns the int value of this variant. */
+ public int getInt() {
+ return getDelegate().getInt();
+ }
+
+ /** Returns the long value of this variant. */
+ public long getLong() {
+ return getDelegate().getLong();
+ }
+
+ /** Returns the float value of this variant. */
+ public float getFloat() {
+ return getDelegate().getFloat();
+ }
+
+ /** Returns the double value of this variant. */
+ public double getDouble() {
+ return getDelegate().getDouble();
+ }
+
+ /** Returns the string value of this variant. */
+ public String getString() {
+ return getDelegate().getString();
+ }
+
+ /** Returns the binary value of this variant. */
+ public byte[] getBinary() {
+ ByteBuffer buf = getDelegate().getBinary();
+ byte[] bytes = new byte[buf.remaining()];
+ buf.get(bytes);
+ return bytes;
+ }
+
+ /** Returns the decimal value of this variant. */
+ public BigDecimal getDecimal() {
+ return getDelegate().getDecimal();
+ }
+
+ /**
+ * Returns the date value (days since epoch) of this variant.
+ *
+ * <p>Note: the Arrow shaded {@code Variant} only exposes generic {@code getInt()} / {@code
+ * getLong()} accessors for DATE / TIMESTAMP types — they read raw bytes without verifying the
+ * variant type. Going through {@link VariantUtil#getDate(byte[], int)} keeps the type assertion
+ * explicit and avoids the indirect ByteBuffer hop.
+ */
+ public int getDate() {
+ return VariantUtil.getDate(value(), 0);
+ }
+
+ /**
+ * Returns the timestamp value (microseconds since epoch) of this variant.
+ *
+ * <p>See {@link #getDate()} for why this stays on {@link VariantUtil} rather than the Arrow
+ * delegate.
+ */
+ public long getTimestamp() {
+ return VariantUtil.getTimestamp(value(), 0);
+ }
+
+ /**
+ * Returns the timestamp without timezone value (microseconds since epoch) of this variant.
+ *
+ * <p>See {@link #getDate()} for why this stays on {@link VariantUtil} rather than the Arrow
+ * delegate.
+ */
+ public long getTimestampNtz() {
+ return VariantUtil.getTimestampNtz(value(), 0);
+ }
+
+ /**
+ * Returns the total size of this variant in bytes (metadata + value).
+ *
+ * @return total byte size
+ */
+ public int sizeInBytes() {
+ return metadata().length + value().length;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Object field access
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Returns the number of fields in this object variant.
+ *
+ * @throws IllegalStateException if this variant is not an object
+ */
+ public int objectSize() {
+ return getDelegate().numObjectElements();
+ }
+
+ /**
+ * Gets a field value from this object variant by field name.
+ *
+ * @param fieldName the field name to look up
+ * @return a new Variant for the field value, or null if the field does not exist
+ */
+ public Variant getFieldByName(String fieldName) {
+ org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant field =
+ getDelegate().getFieldByKey(fieldName);
+ if (field == null) {
+ return null;
+ }
+ return fromParquetVariant(field);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Array element access
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Returns the number of elements in this array variant.
+ *
+ * @throws IllegalStateException if this variant is not an array
+ */
+ public int arraySize() {
+ return getDelegate().numArrayElements();
+ }
+
+ /**
+ * Gets an element from this array variant by index.
+ *
+ * @param index the element index
+ * @return a new Variant for the element value
+ */
+ public Variant getElementAt(int index) {
+ return fromParquetVariant(getDelegate().getElementAtIndex(index));
+ }
+
+ /**
+ * Converts a parquet-variant Variant back to a Fluss Variant by extracting the metadata and
+ * value bytes.
+ */
+ private static Variant fromParquetVariant(
+ org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant pv) {
+ ByteBuffer metaBuf = pv.getMetadataBuffer().duplicate();
+ ByteBuffer valBuf = pv.getValueBuffer().duplicate();
+ byte[] m = new byte[metaBuf.remaining()];
+ metaBuf.get(m);
+ byte[] v = new byte[valBuf.remaining()];
+ valBuf.get(v);
+ return new Variant(m, v);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // JSON conversion
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Creates a Variant from a JSON string using parquet-variant's VariantBuilder.
+ *
+ * @param json the JSON string to parse
+ * @return a new Variant representing the parsed JSON
+ */
+ public static Variant fromJson(String json) {
+ VariantBuilder builder = new VariantBuilder();
+ JsonToVariant.parse(json, builder);
+ org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant built = builder.build();
+ ByteBuffer metaBuf = built.getMetadataBuffer();
+ ByteBuffer valBuf = built.getValueBuffer();
+ byte[] meta = new byte[metaBuf.remaining()];
+ metaBuf.get(meta);
+ byte[] val = new byte[valBuf.remaining()];
+ valBuf.get(val);
+ return new Variant(meta, val);
+ }
+
+ /**
+ * Converts this variant to its JSON string representation. Uses parquet-variant's type system
+ * and accessor methods to produce standard JSON output.
+ *
+ * @return a JSON string
+ */
+ public String toJson() {
+ StringBuilder sb = new StringBuilder();
+ appendJson(getDelegate(), sb);
+ return sb.toString();
+ }
+
+ private static void appendJson(
+ org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant v, StringBuilder sb) {
+ org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant.Type type = v.getType();
+ switch (type) {
+ case NULL:
+ sb.append("null");
+ break;
+ case BOOLEAN:
+ sb.append(v.getBoolean());
+ break;
+ case BYTE:
+ sb.append(v.getByte());
+ break;
+ case SHORT:
+ sb.append(v.getShort());
+ break;
+ case INT:
+ sb.append(v.getInt());
+ break;
+ case LONG:
+ sb.append(v.getLong());
+ break;
+ case FLOAT:
+ sb.append(v.getFloat());
+ break;
+ case DOUBLE:
+ sb.append(v.getDouble());
+ break;
+ case DATE:
+ sb.append(v.getInt());
+ break;
+ case TIMESTAMP_TZ:
+ case TIMESTAMP_NTZ:
+ sb.append(v.getLong());
+ break;
+ case DECIMAL4:
+ case DECIMAL8:
+ case DECIMAL16:
+ sb.append(v.getDecimal().toPlainString());
Review Comment:
We should check for NaN, Infinity values here. JSON does not support that but VARIANT does
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] Support variant type and shredding [fluss]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #2981:
URL: https://github.com/apache/fluss/pull/2981#discussion_r4260331602
##########
fluss-common/src/main/java/org/apache/fluss/types/variant/JsonToVariant.java:
##########
@@ -0,0 +1,288 @@
+/*
+ * 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.fluss.types.variant;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.shaded.arrow.org.apache.parquet.variant.VariantArrayBuilder;
+import org.apache.fluss.shaded.arrow.org.apache.parquet.variant.VariantBuilder;
+import org.apache.fluss.shaded.arrow.org.apache.parquet.variant.VariantObjectBuilder;
+
+import java.math.BigDecimal;
+
+/**
+ * Lightweight JSON parser that feeds parquet-variant's {@link VariantBuilder} to construct Variant
+ * binary encoding. This replaces the custom two-pass JSON parser previously in VariantBuilder.
+ *
+ * <p>The parser supports: null, boolean, numbers (integer/decimal/double), strings, objects, and
+ * arrays. Number handling: integers are stored as the smallest fitting type (byte/short/int/long);
+ * decimals without exponent are stored as DECIMAL; numbers with exponent are stored as DOUBLE.
+ */
+@Internal
+final class JsonToVariant {
+
+ private JsonToVariant() {}
+
+ /** Parses the JSON string and appends the result to the given builder. */
+ static void parse(String json, VariantBuilder builder) {
+ int[] pos = {0};
+ parseValue(json, pos, builder);
+ }
Review Comment:
How about?
```suggestion
static Variant parse(String json) {
VariantBuilder builder = new VariantBuilder();
int[] pos = {0};
parseValue(json, pos, builder);
return builder.build();
}
```
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] Support variant type and shredding [fluss]
Posted by "swuferhong (via GitHub) via issues" <is...@fluss.apache.org>.
swuferhong closed pull request #2981: [WIP] Support variant type and shredding
URL: https://github.com/apache/fluss/pull/2981
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] Support variant type and shredding [fluss]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #2981:
URL: https://github.com/apache/fluss/pull/2981#discussion_r4260200929
##########
fluss-flink/fluss-flink-2.2/src/main/java/org/apache/fluss/flink/adapter/VariantAdapter.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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.fluss.flink.adapter;
+
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.VariantType;
+import org.apache.flink.types.variant.BinaryVariant;
+
+/**
+ * Flink 2.2 implementation of the Variant adapter.
+ *
+ * <p>This class provides actual Variant type support using Flink 2.2's native Variant APIs ({@link
+ * BinaryVariant}, {@link VariantType}).
+ *
+ * @see org.apache.fluss.flink.adapter.SchemaAdapter for a similar adapter pattern
+ */
+public class VariantAdapter {
+ private VariantAdapter() {}
+
+ /** Whether the current Flink version supports the Variant type. */
+ public static boolean supportVariant() {
+ return true;
+ }
+
+ /** Extract metadata bytes from a Flink BinaryVariant object. */
+ public static byte[] getMetadata(Object flinkVariant) {
+ return ((BinaryVariant) flinkVariant).getMetadata();
+ }
+
+ /** Extract value bytes from a Flink BinaryVariant object. */
+ public static byte[] getValue(Object flinkVariant) {
+ return ((BinaryVariant) flinkVariant).getValue();
+ }
+
+ /** Get a Variant object from a Flink RowData at the given position. */
+ public static Object getVariantFromRow(Object rowData, int pos) {
+ return ((RowData) rowData).getVariant(pos);
+ }
+
+ /** Get a Variant object from a Flink ArrayData at the given position. */
+ public static Object getVariantFromArray(Object arrayData, int pos) {
+ return ((ArrayData) arrayData).getVariant(pos);
+ }
+
+ /**
+ * Create a Flink BinaryVariant instance from metadata and value byte arrays.
+ *
+ * @param metadata the Variant metadata bytes
+ * @param value the Variant value bytes
+ * @return a Flink BinaryVariant object
+ */
+ public static Object createBinaryVariant(byte[] metadata, byte[] value) {
+ return new BinaryVariant(metadata, value);
Review Comment:
Flink core defines [BinaryVariant](https://github.com/apache/flink/blob/73e8970f6f71ef2e639a6b436d6e11fed6cb3384/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariant.java#L74) like this:
```java
public BinaryVariant(byte[] value, byte[] metadata) {
this(value, metadata, 0);
}
```
So this will break the Variant type. We should add a round-trip test for this
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] Support variant type and shredding [fluss]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #2981:
URL: https://github.com/apache/fluss/pull/2981#discussion_r4260216842
##########
fluss-flink/fluss-flink-2.2/src/main/java/org/apache/fluss/flink/adapter/VariantAdapter.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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.fluss.flink.adapter;
+
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.VariantType;
+import org.apache.flink.types.variant.BinaryVariant;
+
+/**
+ * Flink 2.2 implementation of the Variant adapter.
+ *
+ * <p>This class provides actual Variant type support using Flink 2.2's native Variant APIs ({@link
+ * BinaryVariant}, {@link VariantType}).
+ *
+ * @see org.apache.fluss.flink.adapter.SchemaAdapter for a similar adapter pattern
+ */
+public class VariantAdapter {
+ private VariantAdapter() {}
+
+ /** Whether the current Flink version supports the Variant type. */
+ public static boolean supportVariant() {
+ return true;
+ }
+
+ /** Extract metadata bytes from a Flink BinaryVariant object. */
+ public static byte[] getMetadata(Object flinkVariant) {
+ return ((BinaryVariant) flinkVariant).getMetadata();
+ }
+
+ /** Extract value bytes from a Flink BinaryVariant object. */
+ public static byte[] getValue(Object flinkVariant) {
+ return ((BinaryVariant) flinkVariant).getValue();
+ }
+
+ /** Get a Variant object from a Flink RowData at the given position. */
+ public static Object getVariantFromRow(Object rowData, int pos) {
+ return ((RowData) rowData).getVariant(pos);
+ }
+
+ /** Get a Variant object from a Flink ArrayData at the given position. */
+ public static Object getVariantFromArray(Object arrayData, int pos) {
+ return ((ArrayData) arrayData).getVariant(pos);
+ }
+
+ /**
+ * Create a Flink BinaryVariant instance from metadata and value byte arrays.
+ *
+ * @param metadata the Variant metadata bytes
+ * @param value the Variant value bytes
+ * @return a Flink BinaryVariant object
+ */
+ public static Object createBinaryVariant(byte[] metadata, byte[] value) {
Review Comment:
Any reason returning Object here and not BinaryVariant?
```suggestion
public static BinaryVariant createBinaryVariant(byte[] metadata, byte[] value) {
```
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] Support variant type and shredding [fluss]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #2981:
URL: https://github.com/apache/fluss/pull/2981#discussion_r4260522745
##########
fluss-common/src/main/java/org/apache/fluss/types/variant/Variant.java:
##########
@@ -0,0 +1,594 @@
+/*
+ * 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.fluss.types.variant;
+
+import org.apache.fluss.annotation.PublicEvolving;
+import org.apache.fluss.shaded.arrow.org.apache.parquet.variant.VariantBuilder;
+
+import java.io.Serializable;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant.Type;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/**
+ * Represents a semi-structured variant value, stored using the Parquet Variant Binary Encoding
+ * format. A Variant consists of two byte arrays:
+ *
+ * <ul>
+ * <li><b>metadata</b>: A dictionary of field names used by object values. Contains a header byte,
+ * dictionary size, offsets, and UTF-8 encoded string bytes.
+ * <li><b>value</b>: A self-describing binary-encoded value. The first byte encodes the basic type
+ * (Primitive, ShortString, Object, Array) and type-specific header information.
+ * </ul>
+ *
+ * <p>For storage in Arrow {@code VarBinaryVector}, the two arrays are concatenated as: {@code
+ * [4-byte metadata length (little-endian)][metadata bytes][value bytes]}.
+ *
+ * <h3>Read path: hybrid Arrow delegate + Fluss {@link VariantUtil}</h3>
+ *
+ * <p>This class deliberately uses two read paths:
+ *
+ * <ul>
+ * <li><b>High-level accessors</b> (typed getters such as {@link #getBoolean()}, {@link
+ * #getInt()}, {@link #getString()}, {@link #getFieldByName(String)}, {@link
+ * #getElementAt(int)}, {@link #toJson()}) are delegated to the Arrow shaded {@code
+ * parquet-variant} {@code Variant} class via {@link #getDelegate()}, so we share its
+ * battle-tested type dispatch.
+ * <li><b>Low-level byte/header inspection</b> (such as {@link #basicType()}, {@link #getDate()},
+ * {@link #getTimestamp()}, {@code encodeXxx} factories, and {@link #emptyMetadata()}) goes
+ * through Fluss {@link VariantUtil}. This is forced by the shaded Arrow library exposing its
+ * own {@code VariantUtil} as package-private (see the javadoc on {@link VariantUtil}). For
+ * these helpers we cannot call into Arrow from outside its package, so {@link VariantUtil} is
+ * a Fluss-side mirror of the Parquet Variant Binary Encoding spec.
+ * </ul>
+ *
+ * @since 0.7
+ */
+@PublicEvolving
+public class Variant implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final byte[] metadata;
+ private final byte[] value;
+
+ public Variant(byte[] metadata, byte[] value) {
+ this.metadata = checkNotNull(metadata, "metadata must not be null");
+ this.value = checkNotNull(value, "value must not be null");
+ }
+
+ /**
+ * Protected constructor for subclasses that override {@link #metadata()} and {@link #value()}
+ * to provide lazy initialization. All public methods in this class access data through the
+ * accessor methods, so subclasses can safely override them.
+ */
+ protected Variant() {
+ this.metadata = null;
+ this.value = null;
+ }
+
+ /** Returns the metadata bytes (field name dictionary). */
+ public byte[] metadata() {
+ return metadata;
+ }
+
+ /** Returns the value bytes (self-describing encoded value). */
+ public byte[] value() {
+ return value;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Binary serialization (for Arrow VarBinaryVector storage)
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Serializes this variant to a single byte array for storage.
+ *
+ * <p>Format: {@code [4-byte metadata length (little-endian)][metadata bytes][value bytes]}
+ */
+ public byte[] toBytes() {
+ byte[] m = metadata();
+ byte[] v = value();
+ byte[] result = new byte[4 + m.length + v.length];
+ ByteBuffer buf = ByteBuffer.wrap(result).order(ByteOrder.LITTLE_ENDIAN);
+ buf.putInt(m.length);
+ buf.put(m);
+ buf.put(v);
+ return result;
+ }
+
+ /**
+ * Deserializes a variant from a byte array.
+ *
+ * @param bytes the serialized bytes
+ * @return the deserialized Variant
+ */
+ public static Variant fromBytes(byte[] bytes) {
+ return fromBytes(bytes, 0, bytes.length);
+ }
+
+ /**
+ * Deserializes a variant from a byte array with offset and length.
+ *
+ * @param bytes the byte array containing the serialized variant
+ * @param offset the starting offset
+ * @param length the total length
+ * @return the deserialized Variant
+ */
+ public static Variant fromBytes(byte[] bytes, int offset, int length) {
+ ByteBuffer buf = ByteBuffer.wrap(bytes, offset, length).order(ByteOrder.LITTLE_ENDIAN);
+ int metadataLen = buf.getInt();
+ byte[] meta = new byte[metadataLen];
+ buf.get(meta);
+ byte[] val = new byte[length - 4 - metadataLen];
+ buf.get(val);
+ return new Variant(meta, val);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Value type inspection
+ // --------------------------------------------------------------------------------------------
+
+ /** Returns the basic type of this variant value. */
+ public int basicType() {
+ return VariantUtil.basicType(value(), 0);
+ }
+
+ /** Returns true if this variant value is null. */
+ public boolean isNull() {
+ // Delegate to the Arrow shaded Variant for the typed null check (Type.NULL).
+ return getDelegate().getType() == Type.NULL;
+ }
+
+ /** Returns true if this variant is a primitive type. */
+ public boolean isPrimitive() {
+ return basicType() == VariantUtil.BASIC_TYPE_PRIMITIVE;
+ }
+
+ /** Returns true if this variant is a short string. */
+ public boolean isShortString() {
+ return basicType() == VariantUtil.BASIC_TYPE_SHORT_STRING;
+ }
+
+ /** Returns true if this variant is an object. */
+ public boolean isObject() {
+ return basicType() == VariantUtil.BASIC_TYPE_OBJECT;
+ }
+
+ /** Returns true if this variant is an array. */
+ public boolean isArray() {
+ return basicType() == VariantUtil.BASIC_TYPE_ARRAY;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Parquet-variant delegate (lazy)
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Lazily created delegate to parquet-variant's Variant for read operations. This avoids
+ * duplicating the Parquet Variant Binary Encoding read logic.
+ */
+ private transient org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant delegate;
+
+ /**
+ * Returns the parquet-variant delegate, creating it lazily from metadata()/value() bytes. This
+ * is used by accessor methods to delegate read operations.
+ */
+ protected org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant getDelegate() {
+ if (delegate == null) {
+ delegate =
+ new org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant(
+ value(), metadata());
+ }
+ return delegate;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Primitive value accessors (delegated to parquet-variant)
+ // --------------------------------------------------------------------------------------------
+
+ /** Returns the boolean value of this variant. */
+ public boolean getBoolean() {
+ return getDelegate().getBoolean();
+ }
+
+ /** Returns the byte value of this variant. */
+ public byte getByte() {
+ return getDelegate().getByte();
+ }
+
+ /** Returns the short value of this variant. */
+ public short getShort() {
+ return getDelegate().getShort();
+ }
+
+ /** Returns the int value of this variant. */
+ public int getInt() {
+ return getDelegate().getInt();
+ }
+
+ /** Returns the long value of this variant. */
+ public long getLong() {
+ return getDelegate().getLong();
+ }
+
+ /** Returns the float value of this variant. */
+ public float getFloat() {
+ return getDelegate().getFloat();
+ }
+
+ /** Returns the double value of this variant. */
+ public double getDouble() {
+ return getDelegate().getDouble();
+ }
+
+ /** Returns the string value of this variant. */
+ public String getString() {
+ return getDelegate().getString();
+ }
+
+ /** Returns the binary value of this variant. */
+ public byte[] getBinary() {
+ ByteBuffer buf = getDelegate().getBinary();
+ byte[] bytes = new byte[buf.remaining()];
+ buf.get(bytes);
+ return bytes;
+ }
+
+ /** Returns the decimal value of this variant. */
+ public BigDecimal getDecimal() {
+ return getDelegate().getDecimal();
+ }
+
+ /**
+ * Returns the date value (days since epoch) of this variant.
+ *
+ * <p>Note: the Arrow shaded {@code Variant} only exposes generic {@code getInt()} / {@code
+ * getLong()} accessors for DATE / TIMESTAMP types — they read raw bytes without verifying the
+ * variant type. Going through {@link VariantUtil#getDate(byte[], int)} keeps the type assertion
+ * explicit and avoids the indirect ByteBuffer hop.
+ */
+ public int getDate() {
+ return VariantUtil.getDate(value(), 0);
+ }
+
+ /**
+ * Returns the timestamp value (microseconds since epoch) of this variant.
+ *
+ * <p>See {@link #getDate()} for why this stays on {@link VariantUtil} rather than the Arrow
+ * delegate.
+ */
+ public long getTimestamp() {
+ return VariantUtil.getTimestamp(value(), 0);
+ }
+
+ /**
+ * Returns the timestamp without timezone value (microseconds since epoch) of this variant.
+ *
+ * <p>See {@link #getDate()} for why this stays on {@link VariantUtil} rather than the Arrow
+ * delegate.
+ */
+ public long getTimestampNtz() {
+ return VariantUtil.getTimestampNtz(value(), 0);
+ }
+
+ /**
+ * Returns the total size of this variant in bytes (metadata + value).
+ *
+ * @return total byte size
+ */
+ public int sizeInBytes() {
+ return metadata().length + value().length;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Object field access
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Returns the number of fields in this object variant.
+ *
+ * @throws IllegalStateException if this variant is not an object
+ */
+ public int objectSize() {
+ return getDelegate().numObjectElements();
+ }
+
+ /**
+ * Gets a field value from this object variant by field name.
+ *
+ * @param fieldName the field name to look up
+ * @return a new Variant for the field value, or null if the field does not exist
+ */
+ public Variant getFieldByName(String fieldName) {
+ org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant field =
+ getDelegate().getFieldByKey(fieldName);
+ if (field == null) {
+ return null;
+ }
+ return fromParquetVariant(field);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Array element access
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Returns the number of elements in this array variant.
+ *
+ * @throws IllegalStateException if this variant is not an array
+ */
+ public int arraySize() {
+ return getDelegate().numArrayElements();
+ }
+
+ /**
+ * Gets an element from this array variant by index.
+ *
+ * @param index the element index
+ * @return a new Variant for the element value
+ */
+ public Variant getElementAt(int index) {
+ return fromParquetVariant(getDelegate().getElementAtIndex(index));
+ }
+
+ /**
+ * Converts a parquet-variant Variant back to a Fluss Variant by extracting the metadata and
+ * value bytes.
+ */
+ private static Variant fromParquetVariant(
+ org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant pv) {
+ ByteBuffer metaBuf = pv.getMetadataBuffer().duplicate();
+ ByteBuffer valBuf = pv.getValueBuffer().duplicate();
+ byte[] m = new byte[metaBuf.remaining()];
+ metaBuf.get(m);
+ byte[] v = new byte[valBuf.remaining()];
+ valBuf.get(v);
+ return new Variant(m, v);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // JSON conversion
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Creates a Variant from a JSON string using parquet-variant's VariantBuilder.
+ *
+ * @param json the JSON string to parse
+ * @return a new Variant representing the parsed JSON
+ */
+ public static Variant fromJson(String json) {
+ VariantBuilder builder = new VariantBuilder();
+ JsonToVariant.parse(json, builder);
+ org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant built = builder.build();
+ ByteBuffer metaBuf = built.getMetadataBuffer();
+ ByteBuffer valBuf = built.getValueBuffer();
+ byte[] meta = new byte[metaBuf.remaining()];
+ metaBuf.get(meta);
+ byte[] val = new byte[valBuf.remaining()];
+ valBuf.get(val);
+ return new Variant(meta, val);
+ }
+
+ /**
+ * Converts this variant to its JSON string representation. Uses parquet-variant's type system
+ * and accessor methods to produce standard JSON output.
+ *
+ * @return a JSON string
+ */
+ public String toJson() {
+ StringBuilder sb = new StringBuilder();
+ appendJson(getDelegate(), sb);
+ return sb.toString();
+ }
+
+ private static void appendJson(
+ org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant v, StringBuilder sb) {
+ org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant.Type type = v.getType();
+ switch (type) {
+ case NULL:
+ sb.append("null");
+ break;
+ case BOOLEAN:
+ sb.append(v.getBoolean());
+ break;
+ case BYTE:
+ sb.append(v.getByte());
+ break;
+ case SHORT:
+ sb.append(v.getShort());
+ break;
+ case INT:
+ sb.append(v.getInt());
+ break;
+ case LONG:
+ sb.append(v.getLong());
+ break;
+ case FLOAT:
+ sb.append(v.getFloat());
+ break;
+ case DOUBLE:
+ sb.append(v.getDouble());
+ break;
+ case DATE:
+ sb.append(v.getInt());
+ break;
+ case TIMESTAMP_TZ:
+ case TIMESTAMP_NTZ:
+ sb.append(v.getLong());
+ break;
+ case DECIMAL4:
+ case DECIMAL8:
+ case DECIMAL16:
+ sb.append(v.getDecimal().toPlainString());
+ break;
+ case STRING:
+ sb.append('"');
+ escapeJsonString(v.getString(), sb);
+ sb.append('"');
+ break;
+ case BINARY:
+ sb.append('"');
+ ByteBuffer binBuf = v.getBinary();
+ byte[] binBytes = new byte[binBuf.remaining()];
+ binBuf.get(binBytes);
+ sb.append(java.util.Base64.getEncoder().encodeToString(binBytes));
+ sb.append('"');
+ break;
+ case OBJECT:
+ sb.append('{');
+ int numFields = v.numObjectElements();
+ for (int i = 0; i < numFields; i++) {
+ if (i > 0) {
+ sb.append(',');
+ }
+ org.apache.fluss.shaded.arrow.org.apache.parquet.variant.Variant.ObjectField
+ field = v.getFieldAtIndex(i);
+ sb.append('"');
+ escapeJsonString(field.key, sb);
+ sb.append("\":");
+ appendJson(field.value, sb);
+ }
+ sb.append('}');
+ break;
+ case ARRAY:
+ sb.append('[');
+ int numElements = v.numArrayElements();
+ for (int i = 0; i < numElements; i++) {
+ if (i > 0) {
+ sb.append(',');
+ }
+ appendJson(v.getElementAtIndex(i), sb);
+ }
+ sb.append(']');
+ break;
+ default:
+ sb.append("null");
+ }
+ }
+
+ private static void escapeJsonString(String s, StringBuilder sb) {
+ for (int i = 0; i < s.length(); i++) {
+ char c = s.charAt(i);
+ switch (c) {
+ case '"':
+ sb.append("\\\"");
+ break;
+ case '\\':
+ sb.append("\\\\");
+ break;
+ case '\b':
+ sb.append("\\b");
+ break;
+ case '\f':
+ sb.append("\\f");
+ break;
+ case '\n':
+ sb.append("\\n");
+ break;
+ case '\r':
+ sb.append("\\r");
+ break;
+ case '\t':
+ sb.append("\\t");
+ break;
+ default:
+ if (c < 0x20) {
+ sb.append(String.format("\\u%04x", (int) c));
+ } else {
+ sb.append(c);
+ }
+ }
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof Variant)) {
+ return false;
+ }
+ Variant variant = (Variant) o;
+ return Arrays.equals(metadata(), variant.metadata())
+ && Arrays.equals(value(), variant.value());
+ }
+
+ @Override
+ public int hashCode() {
+ int result = Arrays.hashCode(metadata());
+ result = 31 * result + Arrays.hashCode(value());
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ return toJson();
+ }
Review Comment:
IMO we should not only return JSON here. JSON is very limited. Imagine your VARIANT has a Infinity filed in it. The toString will return an exception, but your VARIANT is still valid. So IMO we should have a lenient string representation of variant here.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org