This is an automated email from the ASF dual-hosted git repository. RocMarshal pushed a commit to branch release-2.4 in repository https://gitbox.apache.org/repos/asf/flink.git
commit ace2e6cedcfdf5b9f9ce3de2c6ab9ec57adb5cc9 Author: Ramin Gharib <[email protected]> AuthorDate: Tue Nov 3 16:42:39 2026 +0100 [FLINK-41323][table-runtime] Add a binary encoding for a VARIANT column in the raw format The raw format reads and writes a VARIANT column as a JSON document. JSON has no type for dates, times, timestamps, bytes or UUIDs, so a round trip turns those values into strings. Every message is also parsed on read and rendered on write. The new option 'raw.variant-encoding' takes 'json', the default, or 'binary'. With 'binary', a message holds the Variant binary encoding, the metadata followed by the value. The metadata states its own length, so the reader splits the message without a separator and wraps the two parts without parsing. Every Variant type survives the round trip. The bytes are only checked as far as needed to split them, so a malformed value fails when it is read. 'binary' ignores 'raw.charset'. The docs call the JSON round trip lossy, and recommend 'binary' for correctness and performance and 'json' for convenience. Generated-by: Claude Code 2.1.316 (Claude Opus 5.5) (cherry picked from commit c59810c04f830f8d6d2d7a2fc77dcbd03b29e714) --- .../docs/connectors/table/formats/raw.md | 31 +++++++- docs/content/docs/connectors/table/formats/raw.md | 31 +++++++- .../flink/types/variant/BinaryVariantUtil.java | 27 +++++++ .../flink/types/variant/BinaryVariantTest.java | 26 +++++++ .../raw/RawFormatDeserializationSchema.java | 66 ++++++++++++---- .../apache/flink/formats/raw/RawFormatFactory.java | 38 ++++++++-- .../apache/flink/formats/raw/RawFormatOptions.java | 41 ++++++++++ .../formats/raw/RawFormatSerializationSchema.java | 46 ++++++++--- .../table/formats/raw/RawFormatFactoryTest.java | 73 ++++++++++++++++-- .../formats/raw/RawFormatLineDelimiterTest.java | 65 ++++++++++++---- .../formats/raw/RawFormatSerDeSchemaTest.java | 88 +++++++++++++++++++++- 11 files changed, 465 insertions(+), 67 deletions(-) diff --git a/docs/content.zh/docs/connectors/table/formats/raw.md b/docs/content.zh/docs/connectors/table/formats/raw.md index 1d728c3fadb..bcfccf3956d 100644 --- a/docs/content.zh/docs/connectors/table/formats/raw.md +++ b/docs/content.zh/docs/connectors/table/formats/raw.md @@ -119,6 +119,20 @@ Format 参数 <br><strong>注意：</strong>当序列化与反序列化使用相同的分隔符配置时，两者具有 round-trip 兼容性：序列化器追加的末尾分隔符会在反序列化时自动被去除。</td> </tr> + <tr> + <td><h5>raw.variant-encoding</h5></td> + <td>可选</td> + <td style="word-wrap: break-word;">json</td> + <td>Enum</td> + <td>Specify how a <code>VARIANT</code> column is encoded. Valid values are 'json' and 'binary'. + 'json' reads and writes a UTF-8 JSON document. For convenience, and also because JSON is a very + popular and human readable format, the stored VARIANT value is assumed to be JSON. However, JSON + values are lossy: JSON has no type for dates, times, timestamps, bytes or + UUIDs, and it does not keep the width of a number. 'binary' reads and writes the Variant binary + encoding, the metadata followed by the value. It keeps every Variant type and skips JSON parsing + and rendering, so use it for correctness and performance. 'raw.charset' does not apply to + 'binary'.</td> + </tr> </tbody> </table> @@ -183,13 +197,22 @@ Format 参数 </tr> <tr> <td><code>VARIANT</code></td> - <td>A UTF-8 encoded JSON document.<br> + <td>Depends on 'raw.variant-encoding'.<br> + <strong>json</strong> (default): a UTF-8 encoded JSON document. 'raw.charset' must be UTF-8. To read or write JSON in another charset, declare the column as <code>STRING</code> and convert it with <code>PARSE_JSON</code> when reading and <code>JSON_STRING</code> - when writing.<br> + when writing. On read, the bytes are parsed like <code>PARSE_JSON</code> with the same checks, so an invalid UTF-8 byte such as - <code>0xFF</code>, a duplicate object key or malformed JSON fails the job. On write, the value is rendered by <code>Variant#toJson</code>. The round trip is - value-lossless but not byte-lossless: insignificant whitespace is dropped and object keys are ordered.</td> + <code>0xFF</code>, a duplicate object key or malformed JSON fails the job. On write, the value is rendered by + <code>Variant#toJson</code>. + The round trip is lossy, because JSON has fewer types than VARIANT: a <code>DATE</code>, <code>TIME</code>, + <code>TIMESTAMP</code>, <code>BYTES</code> or <code>UUID</code> value comes back as a string, a <code>FLOAT</code> + or <code>DOUBLE</code> can come back as a <code>DECIMAL</code>, and an integer comes back as the smallest integer + type that holds it. + Insignificant whitespace is dropped and object keys are ordered.<br> + <strong>binary</strong>: the Variant binary encoding, the metadata followed by the value. The round trip keeps + every Variant type. The bytes are only checked as far as needed to split them, so a malformed value fails when + it is read.</td> </tr> </tbody> </table> diff --git a/docs/content/docs/connectors/table/formats/raw.md b/docs/content/docs/connectors/table/formats/raw.md index 2f183e53f44..42ef9b54c77 100644 --- a/docs/content/docs/connectors/table/formats/raw.md +++ b/docs/content/docs/connectors/table/formats/raw.md @@ -122,6 +122,20 @@ Format Options deserialization, the two are round-trip compatible: a trailing delimiter appended by the serializer is automatically stripped during deserialization.</td> </tr> + <tr> + <td><h5>raw.variant-encoding</h5></td> + <td>optional</td> + <td style="word-wrap: break-word;">json</td> + <td>Enum</td> + <td>Specify how a <code>VARIANT</code> column is encoded. Valid values are 'json' and 'binary'. + 'json' reads and writes a UTF-8 JSON document. For convenience, and also because JSON is a very + popular and human readable format, the stored VARIANT value is assumed to be JSON. However, JSON + values are lossy: JSON has no type for dates, times, timestamps, bytes or + UUIDs, and it does not keep the width of a number. 'binary' reads and writes the Variant binary + encoding, the metadata followed by the value. It keeps every Variant type and skips JSON parsing + and rendering, so use it for correctness and performance. 'raw.charset' does not apply to + 'binary'.</td> + </tr> </tbody> </table> @@ -186,13 +200,22 @@ The table below details the SQL types the format supports, including details of </tr> <tr> <td><code>VARIANT</code></td> - <td>A UTF-8 encoded JSON document.<br> + <td>Depends on 'raw.variant-encoding'.<br> + <strong>json</strong> (default): a UTF-8 encoded JSON document. 'raw.charset' must be UTF-8. To read or write JSON in another charset, declare the column as <code>STRING</code> and convert it with <code>PARSE_JSON</code> when reading and <code>JSON_STRING</code> - when writing.<br> + when writing. On read, the bytes are parsed like <code>PARSE_JSON</code> with the same checks, so an invalid UTF-8 byte such as - <code>0xFF</code>, a duplicate object key or malformed JSON fails the job. On write, the value is rendered by <code>Variant#toJson</code>. The round trip is - value-lossless but not byte-lossless: insignificant whitespace is dropped and object keys are ordered.</td> + <code>0xFF</code>, a duplicate object key or malformed JSON fails the job. On write, the value is rendered by + <code>Variant#toJson</code>. + The round trip is lossy, because JSON has fewer types than VARIANT: a <code>DATE</code>, <code>TIME</code>, + <code>TIMESTAMP</code>, <code>BYTES</code> or <code>UUID</code> value comes back as a string, a <code>FLOAT</code> + or <code>DOUBLE</code> can come back as a <code>DECIMAL</code>, and an integer comes back as the smallest integer + type that holds it. + Insignificant whitespace is dropped and object keys are ordered.<br> + <strong>binary</strong>: the Variant binary encoding, the metadata followed by the value. The round trip keeps + every Variant type. The bytes are only checked as far as needed to split them, so a malformed value fails when + it is read.</td> </tr> </tbody> </table> diff --git a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantUtil.java b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantUtil.java index 85d13a9f501..38f7eaebaa5 100644 --- a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantUtil.java +++ b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantUtil.java @@ -913,6 +913,33 @@ public class BinaryVariantUtil { return false; } + /** + * Returns the length of the metadata at the start of {@code bytes}. The header, dictionary size + * and last offset give the length, so metadata followed by a value can be split without a + * separator. + * + * @throws VariantTypeException if {@code bytes} does not start with complete metadata + */ + public static int metadataSize(byte[] bytes) { + checkIndex(0, bytes.length); + if ((bytes[0] & VERSION_MASK) != VERSION) { + throw malformedVariant(); + } + final int offsetSize = ((bytes[0] >> 6) & 0x3) + 1; + final int dictSize = readUnsigned(bytes, 1, offsetSize); + // The last of the `dictSize + 1` offsets is the length of the string data. + final long lastOffsetPos = 1 + (dictSize + 1L) * offsetSize; + if (lastOffsetPos + offsetSize > bytes.length) { + throw malformedVariant(); + } + final long size = + lastOffsetPos + offsetSize + readUnsigned(bytes, (int) lastOffsetPos, offsetSize); + if (size > bytes.length) { + throw malformedVariant(); + } + return (int) size; + } + /** * Locates the key at {@code id} in the variant metadata. The range is packed into a long so * that a lookup allocates nothing. diff --git a/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java b/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java index 961c2ff4a4d..c9976e067c2 100644 --- a/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java +++ b/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java @@ -19,6 +19,7 @@ package org.apache.flink.types.variant; import org.apache.flink.core.testutils.CommonTestUtils; +import org.apache.flink.util.StringUtils; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -37,6 +38,7 @@ import java.time.LocalTime; import java.time.ZoneOffset; import java.time.temporal.ChronoUnit; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.UUID; @@ -635,6 +637,30 @@ class BinaryVariantTest { .hasMessage("VARIANT_CONSTRUCTOR_SIZE_LIMIT"); } + @ParameterizedTest + @CsvSource({"1, 1", "300, 2", "70000, 3"}) + void testMetadataSize(final int keyLength, final int offsetSize) { + final BinaryVariant variant = + (BinaryVariant) + builder.object().add("k".repeat(keyLength), builder.of((byte) 1)).build(); + final byte[] metadata = variant.getMetadata(); + final byte[] value = variant.getValue(); + final byte[] bytes = Arrays.copyOf(metadata, metadata.length + value.length); + System.arraycopy(value, 0, bytes, metadata.length, value.length); + + assertThat(((metadata[0] >> 6) & 0x3) + 1).isEqualTo(offsetSize); + assertThat(BinaryVariantUtil.metadataSize(bytes)).isEqualTo(metadata.length); + } + + // Empty, version 2, a dictionary of one key without offsets, and a last offset past the end. + @ParameterizedTest + @ValueSource(strings = {"", "020000", "0101", "010005"}) + void testMetadataSizeRejectsIncompleteMetadata(final String hex) { + assertThatThrownBy(() -> BinaryVariantUtil.metadataSize(StringUtils.hexStringToByte(hex))) + .isInstanceOf(VariantTypeException.class) + .hasMessage("MALFORMED_VARIANT"); + } + @Test void testGetThrowException() { Variant variant = builder.of(10f); diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java index b89881eee49..c8945d27d82 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java @@ -23,13 +23,16 @@ import org.apache.flink.api.common.serialization.DeserializationSchema; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.core.memory.MemorySegment; import org.apache.flink.core.memory.MemorySegmentFactory; +import org.apache.flink.formats.raw.RawFormatOptions.VariantEncoding; import org.apache.flink.table.data.GenericRowData; import org.apache.flink.table.data.RawValueData; import org.apache.flink.table.data.RowData; import org.apache.flink.table.data.StringData; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.types.DeserializationException; +import org.apache.flink.types.variant.BinaryVariant; import org.apache.flink.types.variant.BinaryVariantInternalBuilder; +import org.apache.flink.types.variant.BinaryVariantUtil; import org.apache.flink.types.variant.Variant; import org.apache.flink.util.Collector; @@ -65,27 +68,25 @@ public class RawFormatDeserializationSchema implements DeserializationSchema<Row @Nullable private final String lineDelimiter; + private final VariantEncoding variantEncoding; + private final DeserializationRuntimeConverter converter; private final DataLengthValidator validator; - public RawFormatDeserializationSchema( - LogicalType deserializedType, - TypeInformation<RowData> producedTypeInfo, - String charsetName, - boolean isBigEndian) { - this(deserializedType, producedTypeInfo, charsetName, isBigEndian, null); - } - public RawFormatDeserializationSchema( LogicalType deserializedType, TypeInformation<RowData> producedTypeInfo, String charsetName, boolean isBigEndian, - @Nullable String lineDelimiter) { + @Nullable String lineDelimiter, + VariantEncoding variantEncoding) { this.deserializedType = checkNotNull(deserializedType); this.producedTypeInfo = checkNotNull(producedTypeInfo); - this.converter = createConverter(deserializedType, charsetName, isBigEndian, lineDelimiter); + this.variantEncoding = checkNotNull(variantEncoding); + this.converter = + createConverter( + deserializedType, charsetName, isBigEndian, lineDelimiter, variantEncoding); this.validator = createDataLengthValidator(deserializedType); this.charsetName = charsetName; this.isBigEndian = isBigEndian; @@ -159,13 +160,19 @@ public class RawFormatDeserializationSchema implements DeserializationSchema<Row && deserializedType.equals(that.deserializedType) && charsetName.equals(that.charsetName) && isBigEndian == that.isBigEndian - && Objects.equals(lineDelimiter, that.lineDelimiter); + && Objects.equals(lineDelimiter, that.lineDelimiter) + && variantEncoding == that.variantEncoding; } @Override public int hashCode() { return Objects.hash( - producedTypeInfo, deserializedType, charsetName, isBigEndian, lineDelimiter); + producedTypeInfo, + deserializedType, + charsetName, + isBigEndian, + lineDelimiter, + variantEncoding); } // ------------------------------------------------------------------------ @@ -192,7 +199,8 @@ public class RawFormatDeserializationSchema implements DeserializationSchema<Row LogicalType type, String charsetName, boolean isBigEndian, - @Nullable String lineDelimiter) { + @Nullable String lineDelimiter, + VariantEncoding variantEncoding) { switch (type.getTypeRoot()) { case CHAR: @@ -209,7 +217,9 @@ public class RawFormatDeserializationSchema implements DeserializationSchema<Row return RawValueData::fromBytes; case VARIANT: - return createVariantConverter(); + return variantEncoding == VariantEncoding.BINARY + ? createBinaryVariantConverter() + : createJsonVariantConverter(); case BOOLEAN: return data -> data[0] != 0; @@ -357,7 +367,7 @@ public class RawFormatDeserializationSchema implements DeserializationSchema<Row * Creates a converter that parses UTF-8 JSON bytes into a {@link Variant} like {@code * PARSE_JSON}, so invalid UTF-8 and duplicate object keys are rejected. */ - private static DeserializationRuntimeConverter createVariantConverter() { + private static DeserializationRuntimeConverter createJsonVariantConverter() { return new DeserializationRuntimeConverter() { private static final long serialVersionUID = 1L; @@ -375,6 +385,32 @@ public class RawFormatDeserializationSchema implements DeserializationSchema<Row }; } + /** + * Creates a converter that reads the Variant binary encoding, the metadata followed by the + * value. The bytes are only checked as far as needed to split them. A malformed value fails + * when it is read. + */ + private static DeserializationRuntimeConverter createBinaryVariantConverter() { + return new DeserializationRuntimeConverter() { + private static final long serialVersionUID = 1L; + + @Override + public Object convert(byte[] data) { + try { + final int metadataSize = BinaryVariantUtil.metadataSize(data); + return new BinaryVariant( + Arrays.copyOfRange(data, metadataSize, data.length), + Arrays.copyOf(data, metadataSize)); + } catch (Exception e) { + throw new DeserializationException( + "Failed to deserialize VARIANT type. " + + "The received data is not a valid VARIANT binary encoding.", + e); + } + } + }; + } + private static DeserializationRuntimeConverter createEndiannessAwareConverter( final boolean isBigEndian, final MemorySegmentConverter bigEndianConverter, diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatFactory.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatFactory.java index bf2899d64ec..48571c1de82 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatFactory.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatFactory.java @@ -24,6 +24,7 @@ import org.apache.flink.api.common.serialization.SerializationSchema; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.configuration.ConfigOption; import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.formats.raw.RawFormatOptions.VariantEncoding; import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.connector.ChangelogMode; import org.apache.flink.table.connector.format.DecodingFormat; @@ -76,6 +77,7 @@ public class RawFormatFactory implements DeserializationFormatFactory, Serializa options.add(RawFormatOptions.ENDIANNESS); options.add(RawFormatOptions.CHARSET); options.add(RawFormatOptions.LINE_DELIMITER); + options.add(RawFormatOptions.VARIANT_ENCODING); return options; } @@ -87,6 +89,8 @@ public class RawFormatFactory implements DeserializationFormatFactory, Serializa final boolean isBigEndian = isBigEndian(formatOptions); final Optional<String> lineDelimiter = formatOptions.getOptional(RawFormatOptions.LINE_DELIMITER); + final VariantEncoding variantEncoding = + formatOptions.get(RawFormatOptions.VARIANT_ENCODING); return new DecodingFormat<DeserializationSchema<RowData>>() { @Override @@ -95,7 +99,7 @@ public class RawFormatFactory implements DeserializationFormatFactory, Serializa final RowType physicalRowType = (RowType) producedDataType.getLogicalType(); final LogicalType fieldType = validateAndExtractSingleField(physicalRowType); checkLineDelimiter(fieldType, lineDelimiter); - checkVariantCharset(fieldType, charsetName); + checkVariantOptions(fieldType, charsetName, variantEncoding); final TypeInformation<RowData> producedTypeInfo = context.createTypeInformation(producedDataType); return new RawFormatDeserializationSchema( @@ -103,7 +107,8 @@ public class RawFormatFactory implements DeserializationFormatFactory, Serializa producedTypeInfo, charsetName, isBigEndian, - lineDelimiter.orElse(null)); + lineDelimiter.orElse(null), + variantEncoding); } @Override @@ -121,6 +126,8 @@ public class RawFormatFactory implements DeserializationFormatFactory, Serializa final boolean isBigEndian = isBigEndian(formatOptions); final Optional<String> lineDelimiter = formatOptions.getOptional(RawFormatOptions.LINE_DELIMITER); + final VariantEncoding variantEncoding = + formatOptions.get(RawFormatOptions.VARIANT_ENCODING); return new EncodingFormat<SerializationSchema<RowData>>() { @Override @@ -129,9 +136,13 @@ public class RawFormatFactory implements DeserializationFormatFactory, Serializa final RowType physicalRowType = (RowType) consumedDataType.getLogicalType(); final LogicalType fieldType = validateAndExtractSingleField(physicalRowType); checkLineDelimiter(fieldType, lineDelimiter); - checkVariantCharset(fieldType, charsetName); + checkVariantOptions(fieldType, charsetName, variantEncoding); return new RawFormatSerializationSchema( - fieldType, charsetName, isBigEndian, lineDelimiter.orElse(null)); + fieldType, + charsetName, + isBigEndian, + lineDelimiter.orElse(null), + variantEncoding); } @Override @@ -185,13 +196,26 @@ public class RawFormatFactory implements DeserializationFormatFactory, Serializa } } + /** + * Checks the options of a VARIANT column. The binary encoding is not text, so the charset does + * not apply to it. + */ + private static void checkVariantOptions( + LogicalType fieldType, String charsetName, VariantEncoding variantEncoding) { + if (!fieldType.is(LogicalTypeRoot.VARIANT)) { + return; + } + if (variantEncoding == VariantEncoding.JSON) { + checkVariantCharset(charsetName); + } + } + /** * Checks that a VARIANT column uses UTF-8, which RFC 8259 requires for JSON exchanged between * systems and which {@code PARSE_JSON} reads. */ - private static void checkVariantCharset(LogicalType fieldType, String charsetName) { - if (fieldType.is(LogicalTypeRoot.VARIANT) - && !StandardCharsets.UTF_8.equals(Charset.forName(charsetName))) { + private static void checkVariantCharset(String charsetName) { + if (!StandardCharsets.UTF_8.equals(Charset.forName(charsetName))) { throw new ValidationException( String.format( "The 'raw' format requires '%s.%s' to be UTF-8 for a VARIANT column, " diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatOptions.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatOptions.java index 7fa674b7f98..7ca5e46a9e1 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatOptions.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatOptions.java @@ -21,9 +21,13 @@ package org.apache.flink.formats.raw; import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.configuration.ConfigOption; import org.apache.flink.configuration.ConfigOptions; +import org.apache.flink.configuration.DescribedEnum; +import org.apache.flink.configuration.description.InlineElement; import java.nio.charset.StandardCharsets; +import static org.apache.flink.configuration.description.TextElement.text; + /** Options for the "raw" format. */ @PublicEvolving public class RawFormatOptions { @@ -52,5 +56,42 @@ public class RawFormatOptions { + "When set, deserialization splits each message by this delimiter and emits " + "one RowData per part. Serialization appends the delimiter after each row's value."); + public static final ConfigOption<VariantEncoding> VARIANT_ENCODING = + ConfigOptions.key("variant-encoding") + .enumType(VariantEncoding.class) + .defaultValue(VariantEncoding.JSON) + .withDescription( + "Defines how a VARIANT column is encoded. 'json' reads and writes a UTF-8 " + + "JSON document, which is convenient but lossy: JSON has no type " + + "for dates, times, timestamps, bytes or UUIDs. 'binary' reads and " + + "writes the Variant binary encoding, the metadata followed by the " + + "value, which keeps every Variant type and skips JSON parsing. " + + "'charset' does not apply to 'binary'."); + + /** Encoding of a VARIANT column, see {@link #VARIANT_ENCODING}. */ + @PublicEvolving + public enum VariantEncoding implements DescribedEnum { + JSON("json", text("A UTF-8 JSON document. Types JSON cannot represent are lost.")), + BINARY("binary", text("The Variant binary encoding, the metadata followed by the value.")); + + private final String value; + private final InlineElement description; + + VariantEncoding(String value, InlineElement description) { + this.value = value; + this.description = description; + } + + @Override + public String toString() { + return value; + } + + @Override + public InlineElement getDescription() { + return description; + } + } + private RawFormatOptions() {} } diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatSerializationSchema.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatSerializationSchema.java index 1661b154bbd..02d2f4bee79 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatSerializationSchema.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatSerializationSchema.java @@ -23,9 +23,11 @@ import org.apache.flink.api.common.serialization.SerializationSchema; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.core.memory.MemorySegment; import org.apache.flink.core.memory.MemorySegmentFactory; +import org.apache.flink.formats.raw.RawFormatOptions.VariantEncoding; import org.apache.flink.table.data.RowData; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.RawType; +import org.apache.flink.types.variant.BinaryVariant; import org.apache.flink.types.variant.Variant; import org.apache.flink.types.variant.VariantTypeException; @@ -38,6 +40,8 @@ import java.nio.charset.StandardCharsets; import java.util.Arrays; import java.util.Objects; +import static org.apache.flink.util.Preconditions.checkNotNull; + /** Serialization schema that serializes an {@link RowData} object into raw (byte based) value. */ @Internal public class RawFormatSerializationSchema implements SerializationSchema<RowData> { @@ -57,17 +61,16 @@ public class RawFormatSerializationSchema implements SerializationSchema<RowData /** Pre-computed delimiter bytes, or {@code null} if no delimiter is set. */ @Nullable private final byte[] delimiterBytes; - public RawFormatSerializationSchema( - LogicalType serializedType, String charsetName, boolean isBigEndian) { - this(serializedType, charsetName, isBigEndian, null); - } + private final VariantEncoding variantEncoding; public RawFormatSerializationSchema( LogicalType serializedType, String charsetName, boolean isBigEndian, - @Nullable String lineDelimiter) { + @Nullable String lineDelimiter, + VariantEncoding variantEncoding) { this.serializedType = serializedType; + this.variantEncoding = checkNotNull(variantEncoding); this.converter = createConverter(serializedType, charsetName, isBigEndian); this.charsetName = charsetName; this.isBigEndian = isBigEndian; @@ -108,12 +111,14 @@ public class RawFormatSerializationSchema implements SerializationSchema<RowData return serializedType.equals(that.serializedType) && charsetName.equals(that.charsetName) && isBigEndian == that.isBigEndian - && Objects.equals(lineDelimiter, that.lineDelimiter); + && Objects.equals(lineDelimiter, that.lineDelimiter) + && variantEncoding == that.variantEncoding; } @Override public int hashCode() { - return Objects.hash(serializedType, charsetName, isBigEndian, lineDelimiter); + return Objects.hash( + serializedType, charsetName, isBigEndian, lineDelimiter, variantEncoding); } // ------------------------------------------------------------------------ @@ -167,7 +172,9 @@ public class RawFormatSerializationSchema implements SerializationSchema<RowData return createRawValueConverter((RawType<?>) type); case VARIANT: - return createVariantConverter(charsetName); + return variantEncoding == VariantEncoding.BINARY + ? createBinaryVariantConverter() + : createJsonVariantConverter(charsetName); case BOOLEAN: return row -> { @@ -226,10 +233,12 @@ public class RawFormatSerializationSchema implements SerializationSchema<RowData } /** - * Creates a converter that renders the {@link Variant} as a JSON document. The result is - * value-lossless but not byte-lossless: whitespace and object key order are normalized. + * Creates a converter that renders the {@link Variant} as a JSON document. JSON has fewer types + * than a Variant, so a date, timestamp, byte string or UUID is written as a string. Whitespace + * and object key order are normalized. */ - private static SerializationRuntimeConverter createVariantConverter(final String charsetName) { + private static SerializationRuntimeConverter createJsonVariantConverter( + final String charsetName) { return new SerializationRuntimeConverter() { private static final long serialVersionUID = 1L; private transient Charset charset; @@ -246,6 +255,21 @@ public class RawFormatSerializationSchema implements SerializationSchema<RowData }; } + /** + * Creates a converter that writes the Variant binary encoding, the metadata followed by the + * value. It keeps every Variant type, unlike the JSON document. + */ + private static SerializationRuntimeConverter createBinaryVariantConverter() { + return row -> { + final BinaryVariant variant = (BinaryVariant) row.getVariant(0); + final byte[] metadata = variant.getMetadata(); + final byte[] value = variant.getValue(); + final byte[] result = Arrays.copyOf(metadata, metadata.length + value.length); + System.arraycopy(value, 0, result, metadata.length, value.length); + return result; + }; + } + @SuppressWarnings("unchecked") private static SerializationRuntimeConverter createRawValueConverter(RawType<?> rawType) { final TypeSerializer<Object> serializer = diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatFactoryTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatFactoryTest.java index 92e95efa688..a174770e9df 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatFactoryTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatFactoryTest.java @@ -22,6 +22,7 @@ import org.apache.flink.api.common.serialization.DeserializationSchema; import org.apache.flink.api.common.serialization.SerializationSchema; import org.apache.flink.formats.raw.RawFormatDeserializationSchema; import org.apache.flink.formats.raw.RawFormatFactory; +import org.apache.flink.formats.raw.RawFormatOptions.VariantEncoding; import org.apache.flink.formats.raw.RawFormatSerializationSchema; import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.catalog.Column; @@ -69,14 +70,20 @@ class RawFormatFactoryTest { // test deserialization final RawFormatDeserializationSchema expectedDeser = new RawFormatDeserializationSchema( - ROW_TYPE.getTypeAt(0), InternalTypeInfo.of(ROW_TYPE), UTF_8.name(), true); + ROW_TYPE.getTypeAt(0), + InternalTypeInfo.of(ROW_TYPE), + UTF_8.name(), + true, + null, + VariantEncoding.JSON); DeserializationSchema<RowData> actualDeser = createDeserializationSchema(SCHEMA, tableOptions); assertThat(actualDeser).isEqualTo(expectedDeser); // test serialization final RawFormatSerializationSchema expectedSer = - new RawFormatSerializationSchema(ROW_TYPE.getTypeAt(0), UTF_8.name(), true); + new RawFormatSerializationSchema( + ROW_TYPE.getTypeAt(0), UTF_8.name(), true, null, VariantEncoding.JSON); SerializationSchema<RowData> actualSer = createSerializationSchema(SCHEMA, tableOptions); assertThat(actualSer).isEqualTo(expectedSer); } @@ -93,14 +100,20 @@ class RawFormatFactoryTest { // test deserialization final RawFormatDeserializationSchema expectedDeser = new RawFormatDeserializationSchema( - ROW_TYPE.getTypeAt(0), InternalTypeInfo.of(ROW_TYPE), UTF_16.name(), false); + ROW_TYPE.getTypeAt(0), + InternalTypeInfo.of(ROW_TYPE), + UTF_16.name(), + false, + null, + VariantEncoding.JSON); DeserializationSchema<RowData> actualDeser = createDeserializationSchema(SCHEMA, tableOptions); assertThat(actualDeser).isEqualTo(expectedDeser); // test serialization final RawFormatSerializationSchema expectedSer = - new RawFormatSerializationSchema(ROW_TYPE.getTypeAt(0), UTF_16.name(), false); + new RawFormatSerializationSchema( + ROW_TYPE.getTypeAt(0), UTF_16.name(), false, null, VariantEncoding.JSON); SerializationSchema<RowData> actualSer = createSerializationSchema(SCHEMA, tableOptions); assertThat(actualSer).isEqualTo(expectedSer); } @@ -194,12 +207,19 @@ class RawFormatFactoryTest { variantRowType.getTypeAt(0), InternalTypeInfo.of(variantRowType), UTF_8.name(), - true); + true, + null, + VariantEncoding.JSON); assertThat(createDeserializationSchema(variantSchema, tableOptions)) .isEqualTo(expectedDeser); final RawFormatSerializationSchema expectedSer = - new RawFormatSerializationSchema(variantRowType.getTypeAt(0), UTF_8.name(), true); + new RawFormatSerializationSchema( + variantRowType.getTypeAt(0), + UTF_8.name(), + true, + null, + VariantEncoding.JSON); assertThat(createSerializationSchema(variantSchema, tableOptions)).isEqualTo(expectedSer); } @@ -243,6 +263,41 @@ class RawFormatFactoryTest { return Stream.of(DataTypes.VARIANT(), DataTypes.BYTES(), DataTypes.INT()); } + @Test + void testVariantBinaryEncoding() { + final ResolvedSchema variantSchema = + ResolvedSchema.of(Column.physical("field1", DataTypes.VARIANT())); + final RowType variantRowType = + (RowType) variantSchema.toPhysicalRowDataType().getLogicalType(); + // The charset does not apply to the binary encoding, so UTF-16 is accepted. + final Map<String, String> tableOptions = + getModifiedOptions( + options -> { + options.put("raw.variant-encoding", "binary"); + options.put("raw.charset", UTF_16.name()); + }); + + final RawFormatDeserializationSchema expectedDeser = + new RawFormatDeserializationSchema( + variantRowType.getTypeAt(0), + InternalTypeInfo.of(variantRowType), + UTF_16.name(), + true, + null, + VariantEncoding.BINARY); + assertThat(createDeserializationSchema(variantSchema, tableOptions)) + .isEqualTo(expectedDeser); + + final RawFormatSerializationSchema expectedSer = + new RawFormatSerializationSchema( + variantRowType.getTypeAt(0), + UTF_16.name(), + true, + null, + VariantEncoding.BINARY); + assertThat(createSerializationSchema(variantSchema, tableOptions)).isEqualTo(expectedSer); + } + @Test void testLineDelimiterOption() { final Map<String, String> tableOptions = @@ -258,14 +313,16 @@ class RawFormatFactoryTest { InternalTypeInfo.of(ROW_TYPE), UTF_8.name(), true, - "\n"); + "\n", + VariantEncoding.JSON); DeserializationSchema<RowData> actualDeser = createDeserializationSchema(SCHEMA, tableOptions); assertThat(actualDeser).isEqualTo(expectedDeser); // test serialization schema contains line delimiter final RawFormatSerializationSchema expectedSer = - new RawFormatSerializationSchema(ROW_TYPE.getTypeAt(0), UTF_8.name(), true, "\n"); + new RawFormatSerializationSchema( + ROW_TYPE.getTypeAt(0), UTF_8.name(), true, "\n", VariantEncoding.JSON); SerializationSchema<RowData> actualSer = createSerializationSchema(SCHEMA, tableOptions); assertThat(actualSer).isEqualTo(expectedSer); } diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatLineDelimiterTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatLineDelimiterTest.java index 0c138ab33e1..d08e9137c8b 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatLineDelimiterTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatLineDelimiterTest.java @@ -22,6 +22,7 @@ import org.apache.flink.api.common.serialization.DeserializationSchema; import org.apache.flink.api.common.serialization.SerializationSchema; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.formats.raw.RawFormatDeserializationSchema; +import org.apache.flink.formats.raw.RawFormatOptions.VariantEncoding; import org.apache.flink.formats.raw.RawFormatSerializationSchema; import org.apache.flink.metrics.MetricGroup; import org.apache.flink.table.data.GenericRowData; @@ -68,7 +69,8 @@ class RawFormatLineDelimiterTest { TypeInformation.of(RowData.class), StandardCharsets.UTF_8.name(), true, - null); + null, + VariantEncoding.JSON); openDeser(schema); List<RowData> rows = collectRows(schema, "hello".getBytes(StandardCharsets.UTF_8)); @@ -86,7 +88,8 @@ class RawFormatLineDelimiterTest { TypeInformation.of(RowData.class), StandardCharsets.UTF_8.name(), true, - "\n")); + "\n", + VariantEncoding.JSON)); openDeser(schema); byte[] message = "line1\nline2\nline3".getBytes(StandardCharsets.UTF_8); @@ -105,7 +108,8 @@ class RawFormatLineDelimiterTest { TypeInformation.of(RowData.class), StandardCharsets.UTF_8.name(), true, - "||"); + "||", + VariantEncoding.JSON); openDeser(schema); byte[] message = "record1||record2||record3".getBytes(StandardCharsets.UTF_8); @@ -124,7 +128,8 @@ class RawFormatLineDelimiterTest { TypeInformation.of(RowData.class), StandardCharsets.UTF_8.name(), true, - "\n"); + "\n", + VariantEncoding.JSON); openDeser(schema); List<RowData> rows = collectRows(schema, null); @@ -143,7 +148,12 @@ class RawFormatLineDelimiterTest { RawFormatDeserializationSchema schema = InstantiationUtil.clone( new RawFormatDeserializationSchema( - STRING_TYPE, TypeInformation.of(RowData.class), "GBK", true, "|")); + STRING_TYPE, + TypeInformation.of(RowData.class), + "GBK", + true, + "|", + VariantEncoding.JSON)); openDeser(schema); List<RowData> rows = collectRows(schema, message); @@ -160,7 +170,8 @@ class RawFormatLineDelimiterTest { TypeInformation.of(RowData.class), StandardCharsets.UTF_8.name(), true, - "\n"); + "\n", + VariantEncoding.JSON); openDeser(schema); byte[] message = {'a', (byte) 0xFF, '\n', (byte) 0x80}; @@ -177,7 +188,11 @@ class RawFormatLineDelimiterTest { void testSerializeWithoutDelimiter_noAppend() throws Exception { RawFormatSerializationSchema schema = new RawFormatSerializationSchema( - STRING_TYPE, StandardCharsets.UTF_8.name(), true, null); + STRING_TYPE, + StandardCharsets.UTF_8.name(), + true, + null, + VariantEncoding.JSON); openSer(schema); RowData row = buildStringRow("hello"); @@ -189,7 +204,11 @@ class RawFormatLineDelimiterTest { void testSerializeWithNewlineDelimiter_appendsDelimiter() throws Exception { RawFormatSerializationSchema schema = new RawFormatSerializationSchema( - STRING_TYPE, StandardCharsets.UTF_8.name(), true, "\n"); + STRING_TYPE, + StandardCharsets.UTF_8.name(), + true, + "\n", + VariantEncoding.JSON); openSer(schema); RowData row = buildStringRow("hello"); @@ -201,7 +220,11 @@ class RawFormatLineDelimiterTest { void testSerializeWithCustomDelimiter_appendsDelimiter() throws Exception { RawFormatSerializationSchema schema = new RawFormatSerializationSchema( - STRING_TYPE, StandardCharsets.UTF_8.name(), true, "||"); + STRING_TYPE, + StandardCharsets.UTF_8.name(), + true, + "||", + VariantEncoding.JSON); openSer(schema); RowData row = buildStringRow("record1"); @@ -213,7 +236,11 @@ class RawFormatLineDelimiterTest { void testSerializeNullRow_returnsNull() throws Exception { RawFormatSerializationSchema schema = new RawFormatSerializationSchema( - STRING_TYPE, StandardCharsets.UTF_8.name(), true, "\n"); + STRING_TYPE, + StandardCharsets.UTF_8.name(), + true, + "\n", + VariantEncoding.JSON); openSer(schema); GenericRowData nullRow = new GenericRowData(1); @@ -233,7 +260,8 @@ class RawFormatLineDelimiterTest { TypeInformation.of(RowData.class), StandardCharsets.UTF_8.name(), true, - "\n"); + "\n", + VariantEncoding.JSON); openDeser(schema); // Message already ends with the delimiter (as produced by the serializer) @@ -249,7 +277,11 @@ class RawFormatLineDelimiterTest { // deserializer when both share the same delimiter configuration. RawFormatSerializationSchema ser = new RawFormatSerializationSchema( - STRING_TYPE, StandardCharsets.UTF_8.name(), true, "\n"); + STRING_TYPE, + StandardCharsets.UTF_8.name(), + true, + "\n", + VariantEncoding.JSON); openSer(ser); RawFormatDeserializationSchema deser = @@ -258,7 +290,8 @@ class RawFormatLineDelimiterTest { TypeInformation.of(RowData.class), StandardCharsets.UTF_8.name(), true, - "\n"); + "\n", + VariantEncoding.JSON); openDeser(deser); // Serialize a single row -> "hello\n" @@ -278,7 +311,8 @@ class RawFormatLineDelimiterTest { TypeInformation.of(RowData.class), StandardCharsets.UTF_8.name(), true, - "é"); + "é", + VariantEncoding.JSON); openDeser(schema); // The delimiter é is c3 a9 and the data holds ü, which is c3 bc, so a match on the first @@ -297,7 +331,8 @@ class RawFormatLineDelimiterTest { TypeInformation.of(RowData.class), StandardCharsets.UTF_8.name(), true, - null); + null, + VariantEncoding.JSON); openDeser(deser); byte[] message = {'{', '"', 'a', '"', ':', '"', (byte) 0xFF, '"', '}'}; diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatSerDeSchemaTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatSerDeSchemaTest.java index 1f38c570af4..0ad9c5ba49b 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatSerDeSchemaTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatSerDeSchemaTest.java @@ -24,22 +24,28 @@ import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.common.typeutils.base.LocalDateTimeSerializer; import org.apache.flink.core.memory.DataOutputSerializer; import org.apache.flink.formats.raw.RawFormatDeserializationSchema; +import org.apache.flink.formats.raw.RawFormatOptions.VariantEncoding; import org.apache.flink.formats.raw.RawFormatSerializationSchema; import org.apache.flink.table.data.RowData; import org.apache.flink.table.data.conversion.DataStructureConverter; import org.apache.flink.table.data.conversion.DataStructureConverters; import org.apache.flink.table.types.DataType; +import org.apache.flink.types.DeserializationException; import org.apache.flink.types.Row; import org.apache.flink.types.variant.BinaryVariantInternalBuilder; import org.apache.flink.types.variant.Variant; +import org.apache.flink.types.variant.VariantBuilder; +import org.apache.flink.types.variant.VariantTypeException; import org.apache.flink.util.InstantiationUtil; import org.apache.flink.util.StringUtils; +import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.MethodSource; import java.io.IOException; import java.nio.charset.StandardCharsets; +import java.time.LocalDate; import java.time.LocalDateTime; import java.util.ArrayList; import java.util.Arrays; @@ -62,6 +68,7 @@ import static org.apache.flink.table.api.DataTypes.VARCHAR; import static org.apache.flink.table.api.DataTypes.VARIANT; import static org.apache.flink.util.StringUtils.hexStringToByte; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.Mockito.mock; /** Tests for {@link RawFormatDeserializationSchema} {@link RawFormatSerializationSchema}. */ @@ -70,6 +77,8 @@ class RawFormatSerDeSchemaTest { private static final String JSON_OBJECT = "{\"a\":1,\"b\":\"x\",\"c\":[1,2,3],\"d\":null,\"e\":true}"; + private static final VariantBuilder VARIANT_BUILDER = Variant.newBuilder(); + static List<TestSpec> testData() { return Arrays.asList( TestSpec.type(TINYINT()).values(Byte.MAX_VALUE).binary(new byte[] {Byte.MAX_VALUE}), @@ -164,7 +173,29 @@ class RawFormatSerDeSchemaTest { TestSpec.type(RAW(LocalDateTime.class, new LocalDateTimeSerializer())) .values((Object) null) .binary((byte[]) null), - TestSpec.type(VARIANT()).values((Object) null).binary((byte[]) null)); + TestSpec.type(VARIANT()).values((Object) null).binary((byte[]) null), + + // test variants in the binary encoding, the metadata followed by the value + TestSpec.type(VARIANT()) + .values(VARIANT_BUILDER.of((byte) 7)) + .withVariantEncoding(VariantEncoding.BINARY) + .binary(hexStringToByte("0100000c07")), + // {"d": DATE 2026-11-03, "b": BYTES x'04'}, two types JSON cannot represent + TestSpec.type(VARIANT()) + .values( + VARIANT_BUILDER + .object() + .add("d", VARIANT_BUILDER.of(LocalDate.parse("2026-11-03"))) + .add("b", VARIANT_BUILDER.of(new byte[] {4})) + .build()) + .withVariantEncoding(VariantEncoding.BINARY) + .binary( + hexStringToByte( + "01020001026462" + "0202010005000b2c185100003c0100000004")), + TestSpec.type(VARIANT()) + .values((Object) null) + .withVariantEncoding(VariantEncoding.BINARY) + .binary((byte[]) null)); } @ParameterizedTest @@ -178,13 +209,17 @@ class RawFormatSerDeSchemaTest { testSpec.type.getLogicalType(), TypeInformation.of(RowData.class), testSpec.charsetName, - testSpec.isBigEndian)); + testSpec.isBigEndian, + null, + testSpec.variantEncoding)); RawFormatSerializationSchema serializationSchema = InstantiationUtil.clone( new RawFormatSerializationSchema( testSpec.type.getLogicalType(), testSpec.charsetName, - testSpec.isBigEndian)); + testSpec.isBigEndian, + null, + testSpec.variantEncoding)); deserializationSchema.open(mock(DeserializationSchema.InitializationContext.class)); serializationSchema.open(mock(SerializationSchema.InitializationContext.class)); @@ -219,6 +254,45 @@ class RawFormatSerDeSchemaTest { } } + @Test + void testDeserializeMalformedVariantBinary() throws Exception { + final RawFormatDeserializationSchema schema = createBinaryVariantDeserializationSchema(); + + // A dictionary of one key without its offsets. + assertThatThrownBy(() -> schema.deserialize(hexStringToByte("0101"))) + .isInstanceOf(DeserializationException.class) + .hasMessage( + "Failed to deserialize VARIANT type. " + + "The received data is not a valid VARIANT binary encoding.") + .cause() + .hasMessage("MALFORMED_VARIANT"); + } + + @Test + void testDeserializeVariantBinaryChecksTheValueWhenItIsRead() throws Exception { + final RawFormatDeserializationSchema schema = createBinaryVariantDeserializationSchema(); + + // Complete metadata without a value is accepted, and fails once the value is read. + final Variant variant = schema.deserialize(hexStringToByte("010000")).getVariant(0); + assertThatThrownBy(variant::getType) + .isInstanceOf(VariantTypeException.class) + .hasMessage("MALFORMED_VARIANT"); + } + + private static RawFormatDeserializationSchema createBinaryVariantDeserializationSchema() + throws Exception { + final RawFormatDeserializationSchema schema = + new RawFormatDeserializationSchema( + VARIANT().getLogicalType(), + TypeInformation.of(RowData.class), + StandardCharsets.UTF_8.name(), + true, + null, + VariantEncoding.BINARY); + schema.open(mock(DeserializationSchema.InitializationContext.class)); + return schema; + } + private static Variant variant(String json) { try { return BinaryVariantInternalBuilder.parseJson(json, false); @@ -247,6 +321,7 @@ class RawFormatSerDeSchemaTest { private final DataType type; private String charsetName = "UTF-8"; private boolean isBigEndian = true; + private VariantEncoding variantEncoding = VariantEncoding.JSON; private TestSpec(DataType type) { this.type = type; @@ -271,6 +346,11 @@ class RawFormatSerDeSchemaTest { return this; } + public TestSpec withVariantEncoding(VariantEncoding variantEncoding) { + this.variantEncoding = variantEncoding; + return this; + } + public TestSpec withLittleEndian() { this.isBigEndian = false; return this; @@ -295,6 +375,8 @@ class RawFormatSerDeSchemaTest { + '\'' + ", isBigEndian=" + isBigEndian + + ", variantEncoding=" + + variantEncoding + '}'; } }
