This is an automated email from the ASF dual-hosted git repository. twalthr pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 65d94e746ee5e6e485a686b26f9cf3756165866e Author: Ramin Gharib <[email protected]> AuthorDate: Mon Oct 5 15:12:48 2026 +0200 [FLINK-40826][table] Cast primitives to VARIANT through ToVariantConverter ToVariantConverter in flink-table-common is now the only place that maps a SQL type to a variant kind. It is built once per type as a tree of writers that write a whole value into one builder. A format can write a value into its own builder through writeTo, exactly like the cast does. PrimitiveToVariantCastRule becomes ToVariantCastRule, whose generated code calls the converter. The VariantCastUtils.from* and append* methods are removed, and their encoding moves into the converter, including the unscaled-long path for decimals. The UTF-8 handling of strings moves to StringUtf8Utils#toValidUtf8Bytes. CodeGeneratorCastRule.Context#declareReusableObject hands the converter to the generated code. The stored values are unchanged, with these fixes: - A DISTINCT type casts like its source type instead of failing in code generation. CastRuleUtils.box used to drop its result for a DISTINCT type. - A TIME outside one day fails with a clear message instead of a DateTimeException, and TRY_CAST returns NULL for it. - A TIMESTAMP(7..9) or TIMESTAMP_LTZ(7..9) in the first 0.78 ms of the nanosecond range no longer overflows while being converted. --- docs/content/docs/sql/reference/data-types.md | 4 +- .../flink/table/data/binary/StringUtf8Utils.java | 20 + .../flink/table/data/utils/ToVariantConverter.java | 403 +++++++++++++++ .../table/data/binary/StringUtf8UtilsTest.java | 20 + .../table/data/utils/ToVariantConverterTest.java | 557 +++++++++++++++++++++ .../casting/AbstractCodeGeneratorCastRule.java | 56 +-- .../AbstractExpressionCodeGeneratorCastRule.java | 6 + .../functions/casting/CastRuleProvider.java | 2 +- .../planner/functions/casting/CastRuleUtils.java | 2 +- .../functions/casting/CodeGeneratorCastRule.java | 9 + .../casting/PrimitiveToVariantCastRule.java | 179 ------- .../functions/casting/ToVariantCastRule.java | 123 +++++ .../planner/codegen/calls/ScalarOperatorGens.scala | 2 + .../planner/functions/CastFunctionITCase.java | 21 + .../functions/casting/CastRuleProviderTest.java | 18 +- .../planner/functions/casting/CastRulesTest.java | 15 + .../table/runtime/functions/VariantCastUtils.java | 198 +------- .../runtime/functions/VariantCastUtilsTest.java | 169 ------- 18 files changed, 1224 insertions(+), 580 deletions(-) diff --git a/docs/content/docs/sql/reference/data-types.md b/docs/content/docs/sql/reference/data-types.md index ca35e7dac3f..c2c94bb2c5c 100644 --- a/docs/content/docs/sql/reference/data-types.md +++ b/docs/content/docs/sql/reference/data-types.md @@ -1700,6 +1700,8 @@ A scalar value can also be cast to a `VARIANT` with `CAST` or `TRY_CAST`. Only a stored with microseconds, and above with nanoseconds, even when the value has no digits below a microsecond. Nanoseconds only cover 1677-09-21 to 2262-04-11, so for a precision above 6 a value outside that range fails the cast, and `TRY_CAST` returns `NULL`. +- A `TIME` holds the milliseconds of one day. Only a source or a function can produce a value + outside that range, but such a value fails the cast, and `TRY_CAST` returns `NULL`. - A `NaN` or infinite `FLOAT` or `DOUBLE` is stored as is, although `PARSE_JSON` rejects them. A `VARIANT` is not limited to JSON, but JSON has no literal for them. So `JSON_STRING` fails on such a variant, and so does writing it to a sink with the `json` or `raw` format, which fails the job. @@ -1936,7 +1938,7 @@ The matrix below describes the supported cast pairs, where "Y" means supported, | `FLOAT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | Y | N | | `DOUBLE` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | Y | N | | `DATE` | Y | N | N | N | N | N | N | N | N | N | Y | N | Y | Y | N | N | N | N | N | N | N | Y | N | -| `TIME` | Y | N | N | N | N | N | N | N | N | N | N | Y | Y | Y | N | N | N | N | N | N | N | Y | N | +| `TIME` | Y | N | N | N | N | N | N | N | N | N | N | Y | Y | Y | N | N | N | N | N | N | N | ! | N | | `TIMESTAMP` | Y | N | N | N | N | N | N | N | N | N | Y | Y | Y | Y | N | N | N | N | N | N | N | ! | N | | `TIMESTAMP_LTZ` | Y | N | N | N | N | N | N | N | N | N | Y | Y | Y | Y | N | N | N | N | N | N | N | ! | N | | `INTERVAL` | Y | N | N | N | N | N | Y⁵ | Y⁶ | N | N | N | N | N | N | Y | N | N | N | N | N | N | N | N | diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/StringUtf8Utils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/StringUtf8Utils.java index 4df4ebbe559..cfbeaa19e5b 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/StringUtf8Utils.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/StringUtf8Utils.java @@ -19,6 +19,7 @@ package org.apache.flink.table.data.binary; import org.apache.flink.annotation.Internal; import org.apache.flink.core.memory.MemorySegment; +import org.apache.flink.table.data.StringData; import java.lang.invoke.MethodHandles; import java.lang.invoke.VarHandle; @@ -300,6 +301,25 @@ public final class StringUtf8Utils { return -1; } + /** + * Returns the UTF-8 bytes of the given string. Every malformed sequence is replaced by the + * U+FFFD replacement character, the same as {@link StringData#toString()} decodes it. The + * returned array may be shared with the string, so it must not be modified. + */ + public static byte[] toValidUtf8Bytes(final StringData string) { + final BinaryStringData binaryString = (BinaryStringData) string; + if (binaryString.getBinarySection() == null) { + // A string that only exists as a Java object, such as a literal or a function result, + // is valid UTF-8 once encoded. Encoding it here skips the binary form and the check. + return encodeUTF8(binaryString.getJavaObject()); + } + final byte[] bytes = binaryString.toBytes(); + if (firstInvalidUtf8ByteIndex(bytes, 0, bytes.length) < 0) { + return bytes; + } + return encodeUTF8(defaultDecodeUTF8(bytes, 0, bytes.length)); + } + public static String decodeUTF8(MemorySegment input, int offset, int byteLen) { char[] chars = allocateReuseChars(byteLen); int len = decodeUTF8Strict(input, offset, byteLen, chars); diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/ToVariantConverter.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/ToVariantConverter.java new file mode 100644 index 00000000000..d6a36412953 --- /dev/null +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/ToVariantConverter.java @@ -0,0 +1,403 @@ +/* + * 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.flink.table.data.utils; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.TableRuntimeException; +import org.apache.flink.table.data.ArrayData; +import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.MapData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.data.binary.StringUtf8Utils; +import org.apache.flink.table.types.logical.ArrayType; +import org.apache.flink.table.types.logical.DistinctType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.LogicalTypeFamily; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.types.logical.MapType; +import org.apache.flink.table.types.logical.utils.LogicalTypeChecks; +import org.apache.flink.table.types.logical.utils.UuidUtils; +import org.apache.flink.types.variant.BinaryVariant; +import org.apache.flink.types.variant.BinaryVariantInternalBuilder; +import org.apache.flink.types.variant.BinaryVariantInternalBuilder.FieldEntry; +import org.apache.flink.types.variant.BinaryVariantUtil; +import org.apache.flink.types.variant.Variant; +import org.apache.flink.types.variant.VariantTypeException; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.List; + +import static org.apache.flink.table.utils.DateTimeUtils.MILLIS_PER_DAY; + +/** + * Converts internal data of a SQL type into a {@link Variant}. It implements {@code CAST(x AS + * VARIANT)}, and a format can use it to store a value in a {@code VARIANT} the same way. + * + * <p>A leaf keeps the kind of its SQL type. For example, a {@code BIGINT} is stored as a {@code + * BIGINT}, even when it would fit a smaller kind. A {@code TIMESTAMP(p)} or {@code + * TIMESTAMP_LTZ(p)} is stored with microseconds up to {@code p = 6} and with nanoseconds above, + * whatever the digits of the value. A {@code NaN} or infinite {@code FLOAT} or {@code DOUBLE} is + * stored as is, although {@code PARSE_JSON} rejects it. + * + * <p>An {@code ARRAY} becomes a variant array. A {@code ROW} or {@code STRUCTURED} value becomes a + * variant object keyed by its field names, and a {@code MAP} one keyed by its character string + * keys. A {@code NULL} becomes a variant null, and a nested {@code VARIANT} is copied in as is. + * + * <p>The converter is created once per type and writes a whole value into one builder. A nested + * {@code ARRAY}, {@code MAP} or {@code ROW} is written in place rather than built as a separate + * variant first. The converter keeps no state between calls. + */ +@Internal +public final class ToVariantConverter implements Serializable { + + private static final long serialVersionUID = 1L; + + /** The highest timestamp precision a {@code VARIANT} stores with microseconds. */ + public static final int TIMESTAMP_PRECISION = 6; + + /** + * The largest string or binary value a {@code VARIANT} holds, in bytes. A {@code VARIANT} is + * limited to 16 MiB, and a long string or binary value spends 5 bytes of that on its header. + */ + public static final int MAX_PAYLOAD_BYTES = + BinaryVariantUtil.SIZE_LIMIT - 1 - BinaryVariantUtil.U32_SIZE; + + private final String sourceType; + private final ValueWriter writer; + + private ToVariantConverter(String sourceType, ValueWriter writer) { + this.sourceType = sourceType; + this.writer = writer; + } + + /** + * Creates a converter for a type that {@code LogicalTypeCasts} allows to cast to VARIANT. + * + * @throws IllegalArgumentException if the type cannot be cast to VARIANT + */ + public static ToVariantConverter create(LogicalType type) { + return new ToVariantConverter(type.asSummaryString(), createNullableWriter(type)); + } + + /** + * Converts a value into a new {@code VARIANT}. A {@code null} value becomes a variant null. + * + * @throws TableRuntimeException if the value has a {@code NULL} map key, a time or timestamp + * outside the range of its variant kind, does not fit into the 16 MiB of a {@code VARIANT}, + * or is nested too deeply + * @throws VariantTypeException if a nested {@code VARIANT} is malformed + */ + public Variant convert(Object value) { + // A MAP can hold the same key twice at runtime. The builder keeps the last one, like the + // MAP constructor does. + final BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(true); + try { + writeTo(builder, value); + return builder.build(); + } catch (VariantTypeException e) { + if (e != BinaryVariantInternalBuilder.VARIANT_SIZE_LIMIT_EXCEPTION) { + throw e; + } + throw new TableRuntimeException( + String.format( + "Cannot cast a value of type %s to VARIANT. A VARIANT is limited to " + + "16 MiB.", + sourceType)); + } catch (StackOverflowError e) { + // The builder copies an embedded VARIANT with one call per nesting level, so a deeply + // nested one overflows the stack. A bare StackOverflowError would not say which cast + // failed or why. + throw new TableRuntimeException( + String.format( + "Cannot cast a value of type %s to VARIANT because it is nested too " + + "deeply.", + sourceType)); + } + } + + /** + * Writes a value into a builder that the caller owns, for example as a field of a larger {@code + * VARIANT}. A {@code null} value becomes a variant null. + * + * <p>It fails like {@link #convert} for a {@code NULL} map key, a time or timestamp out of + * range, and a string or binary value that can never fit. The caller builds the result, so it + * handles {@link BinaryVariantInternalBuilder#VARIANT_SIZE_LIMIT_EXCEPTION} and a {@link + * StackOverflowError} from a deeply nested {@code VARIANT} itself. A repeated {@code MAP} key + * keeps the last value if the caller's builder allows duplicate keys, and fails otherwise. The + * builder holds partial data after a failure, so the caller must not use it any further. + */ + public void writeTo(BinaryVariantInternalBuilder builder, Object value) { + writer.write(builder, value); + } + + @FunctionalInterface + private interface ValueWriter extends Serializable { + void write(BinaryVariantInternalBuilder builder, Object value); + } + + private static ValueWriter createNullableWriter(LogicalType type) { + final ValueWriter writer = createWriter(type); + return (builder, value) -> { + if (value == null) { + builder.appendNull(); + } else { + writer.write(builder, value); + } + }; + } + + private static ValueWriter createWriter(LogicalType type) { + switch (type.getTypeRoot()) { + case NULL: + return (builder, value) -> builder.appendNull(); + case BOOLEAN: + return (builder, value) -> builder.appendBoolean((Boolean) value); + case TINYINT: + return (builder, value) -> builder.appendByte((Byte) value); + case SMALLINT: + return (builder, value) -> builder.appendShort((Short) value); + case INTEGER: + return (builder, value) -> builder.appendInt((Integer) value); + case BIGINT: + return (builder, value) -> builder.appendLong((Long) value); + case FLOAT: + return (builder, value) -> builder.appendFloat((Float) value); + case DOUBLE: + return (builder, value) -> builder.appendDouble((Double) value); + case DECIMAL: + return (builder, value) -> appendDecimal(builder, (DecimalData) value); + case CHAR: + case VARCHAR: + return (builder, value) -> appendString(builder, (StringData) value); + case BINARY: + case VARBINARY: + return (builder, value) -> appendBinary(builder, (byte[]) value); + case DATE: + return (builder, value) -> builder.appendDate((Integer) value); + case TIME_WITHOUT_TIME_ZONE: + return (builder, value) -> builder.appendTime(timeMicros((Integer) value)); + case TIMESTAMP_WITHOUT_TIME_ZONE: + { + final int precision = LogicalTypeChecks.getPrecision(type); + if (precision <= TIMESTAMP_PRECISION) { + return (builder, value) -> + builder.appendTimestamp(timestampMicros((TimestampData) value)); + } + final String timestampType = "TIMESTAMP(" + precision + ")"; + return (builder, value) -> + builder.appendTimestampNanos( + timestampNanos((TimestampData) value, timestampType)); + } + case TIMESTAMP_WITH_LOCAL_TIME_ZONE: + { + final int precision = LogicalTypeChecks.getPrecision(type); + if (precision <= TIMESTAMP_PRECISION) { + return (builder, value) -> + builder.appendTimestampLtz(timestampMicros((TimestampData) value)); + } + final String timestampType = "TIMESTAMP_LTZ(" + precision + ")"; + return (builder, value) -> + builder.appendTimestampLtzNanos( + timestampNanos((TimestampData) value, timestampType)); + } + case UUID: + return (builder, value) -> builder.appendUuid(UuidUtils.fromBytes((byte[]) value)); + case VARIANT: + return (builder, value) -> builder.appendVariant((BinaryVariant) value); + case ARRAY: + return createArrayWriter((ArrayType) type); + case MAP: + return createMapWriter((MapType) type); + case ROW: + case STRUCTURED_TYPE: + return createRowWriter(type); + case DISTINCT_TYPE: + return createWriter(((DistinctType) type).getSourceType()); + default: + throw new IllegalArgumentException("Cannot cast " + type + " to VARIANT."); + } + } + + // A compact decimal is written from its unscaled long, which saves building a BigDecimal. + private static void appendDecimal(BinaryVariantInternalBuilder builder, DecimalData value) { + if (value.isCompact()) { + builder.appendDecimal(value.toUnscaledLong(), value.scale()); + } else { + builder.appendDecimal(value.toBigDecimal()); + } + } + + private static void appendString(BinaryVariantInternalBuilder builder, StringData value) { + // The variant spec only allows valid UTF-8, but a StringData may hold invalid bytes. + final byte[] utf8 = StringUtf8Utils.toValidUtf8Bytes(value); + try { + builder.appendString(utf8); + } catch (VariantTypeException e) { + throw payloadTooLarge(e, "string", utf8.length); + } + } + + private static void appendBinary(BinaryVariantInternalBuilder builder, byte[] value) { + try { + builder.appendBinary(value); + } catch (VariantTypeException e) { + throw payloadTooLarge(e, "binary", value.length); + } + } + + // A string or binary value too large on its own gets a message with its size. When only the + // value as a whole is too large, convert reports that instead. + private static RuntimeException payloadTooLarge( + VariantTypeException e, String kind, int bytes) { + if (e != BinaryVariantInternalBuilder.VARIANT_SIZE_LIMIT_EXCEPTION + || bytes <= MAX_PAYLOAD_BYTES) { + return e; + } + return new TableRuntimeException( + String.format( + "Cannot cast a %s value of %d bytes to VARIANT. A VARIANT is limited to " + + "16 MiB, so a string or binary value can have at most %d bytes.", + kind, bytes, MAX_PAYLOAD_BYTES)); + } + + /** Microseconds of the day. TIME is millisecond-of-day at runtime. */ + private static long timeMicros(int millisOfDay) { + if (millisOfDay < 0 || millisOfDay >= MILLIS_PER_DAY) { + throw new TableRuntimeException( + String.format( + "Cannot cast the TIME value %d to VARIANT. A TIME holds the " + + "milliseconds of one day, from 0 to %d.", + millisOfDay, MILLIS_PER_DAY - 1)); + } + return millisOfDay * 1_000L; + } + + /** Microseconds since the epoch, which cover every year a {@link TimestampData} can hold. */ + private static long timestampMicros(TimestampData value) { + return value.getMillisecond() * 1_000L + value.getNanoOfMillisecond() / 1_000; + } + + /** Nanoseconds since the epoch, which only cover 1677-09-21 to 2262-04-11. */ + private static long timestampNanos(TimestampData value, String timestampType) { + final long millis = value.getMillisecond(); + final int nanos = value.getNanoOfMillisecond(); + try { + // Before the epoch, the milliseconds alone can underflow although the sum fits, so + // one millisecond moves into the nanoseconds first. + return millis >= 0 + ? Math.addExact(Math.multiplyExact(millis, 1_000_000L), nanos) + : Math.addExact(Math.multiplyExact(millis + 1, 1_000_000L), nanos - 1_000_000L); + } catch (ArithmeticException e) { + throw new TableRuntimeException( + String.format( + "Cannot cast the %s value %s to VARIANT. A VARIANT timestamp with " + + "nanosecond precision only covers 1677-09-21 to 2262-04-11. " + + "Cast the value to a precision of 6 or less first.", + timestampType, value)); + } + } + + private static ValueWriter createArrayWriter(ArrayType type) { + final ArrayData.ElementGetter elementGetter = createElementGetter(type.getElementType()); + final ValueWriter elementWriter = createNullableWriter(type.getElementType()); + return (builder, value) -> { + final ArrayData array = (ArrayData) value; + final int size = array.size(); + final int start = builder.getWritePos(); + final ArrayList<Integer> offsets = new ArrayList<>(size); + for (int i = 0; i < size; i++) { + offsets.add(builder.getWritePos() - start); + elementWriter.write(builder, elementGetter.getElementOrNull(array, i)); + } + builder.finishWritingArray(start, offsets); + }; + } + + private static ValueWriter createMapWriter(MapType type) { + if (!type.getKeyType().is(LogicalTypeFamily.CHARACTER_STRING)) { + throw new IllegalArgumentException( + "Cannot cast " + type + " to VARIANT. A MAP key must be a character string."); + } + final ArrayData.ElementGetter keyGetter = createElementGetter(type.getKeyType()); + final ArrayData.ElementGetter valueGetter = createElementGetter(type.getValueType()); + final ValueWriter valueWriter = createNullableWriter(type.getValueType()); + final String mapType = type.asSummaryString(); + return (builder, value) -> { + final MapData map = (MapData) value; + final ArrayData keys = map.keyArray(); + final ArrayData values = map.valueArray(); + final int size = map.size(); + final int start = builder.getWritePos(); + final ArrayList<FieldEntry> fields = new ArrayList<>(size); + for (int i = 0; i < size; i++) { + final Object key = keyGetter.getElementOrNull(keys, i); + if (key == null) { + throw new TableRuntimeException( + String.format( + "Cannot cast a value of type %s with a NULL key to VARIANT. " + + "A VARIANT object key cannot be NULL.", + mapType)); + } + final String name = key.toString(); + fields.add( + new FieldEntry(name, builder.addKey(name), builder.getWritePos() - start)); + valueWriter.write(builder, valueGetter.getElementOrNull(values, i)); + } + builder.finishWritingObject(start, fields); + }; + } + + private static ValueWriter createRowWriter(LogicalType type) { + final List<LogicalType> fieldTypes = LogicalTypeChecks.getFieldTypes(type); + final String[] fieldNames = LogicalTypeChecks.getFieldNames(type).toArray(new String[0]); + final int fieldCount = fieldTypes.size(); + final RowData.FieldGetter[] fieldGetters = new RowData.FieldGetter[fieldCount]; + final ValueWriter[] fieldWriters = new ValueWriter[fieldCount]; + for (int i = 0; i < fieldCount; i++) { + fieldGetters[i] = createFieldGetter(fieldTypes.get(i), i); + fieldWriters[i] = createNullableWriter(fieldTypes.get(i)); + } + return (builder, value) -> { + final RowData row = (RowData) value; + final int start = builder.getWritePos(); + final ArrayList<FieldEntry> fields = new ArrayList<>(fieldCount); + for (int i = 0; i < fieldCount; i++) { + final String name = fieldNames[i]; + fields.add( + new FieldEntry(name, builder.addKey(name), builder.getWritePos() - start)); + fieldWriters[i].write(builder, fieldGetters[i].getFieldOrNull(row)); + } + builder.finishWritingObject(start, fields); + }; + } + + // A NULL type has no accessor, but its values are always NULL, which becomes a variant null. + private static ArrayData.ElementGetter createElementGetter(LogicalType type) { + return type.is(LogicalTypeRoot.NULL) + ? (array, pos) -> null + : ArrayData.createElementGetter(type); + } + + private static RowData.FieldGetter createFieldGetter(LogicalType type, int pos) { + return type.is(LogicalTypeRoot.NULL) ? row -> null : RowData.createFieldGetter(type, pos); + } +} diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/binary/StringUtf8UtilsTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/binary/StringUtf8UtilsTest.java index f2fd53ba132..c6dd60c5448 100644 --- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/binary/StringUtf8UtilsTest.java +++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/binary/StringUtf8UtilsTest.java @@ -87,6 +87,26 @@ class StringUtf8UtilsTest { .hasMessageContaining("Invalid UTF-8 byte at index 2 of 3"); } + @Test + void testToValidUtf8Bytes() { + final byte[] valid = "Grüße, 世界 🚀".getBytes(StandardCharsets.UTF_8); + + // Valid bytes are returned as they are. + assertThat(StringUtf8Utils.toValidUtf8Bytes(StringData.fromBytes(valid))).isSameAs(valid); + + // Every malformed sequence becomes U+FFFD, as when the string is decoded. + assertThat(StringUtf8Utils.toValidUtf8Bytes(StringData.fromBytes(bytes('a', 0xFF, 'b')))) + .isEqualTo("a\uFFFDb".getBytes(StandardCharsets.UTF_8)); + assertThat(StringUtf8Utils.toValidUtf8Bytes(StringData.fromBytes(bytes('a', 0xC3)))) + .isEqualTo("a\uFFFD".getBytes(StandardCharsets.UTF_8)); + + // A Java string is encoded without building its binary form. An unpaired surrogate has no + // UTF-8 form, so encoding stores '?' in its place. + final BinaryStringData javaString = BinaryStringData.fromString("a\uD800b"); + assertThat(StringUtf8Utils.toValidUtf8Bytes(javaString)).isEqualTo(bytes('a', '?', 'b')); + assertThat(javaString.getBinarySection()).isNull(); + } + @ParameterizedTest(name = "{0}") @ValueSource(strings = {"Hello, world", "Café au lait", "é", "a😀b"}) void testDecodeUtf8RoundTrip(String input) { diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/utils/ToVariantConverterTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/utils/ToVariantConverterTest.java new file mode 100644 index 00000000000..068757f8f42 --- /dev/null +++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/utils/ToVariantConverterTest.java @@ -0,0 +1,557 @@ +/* + * 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.flink.table.data.utils; + +import org.apache.flink.core.memory.MemorySegment; +import org.apache.flink.core.memory.MemorySegmentFactory; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.TableRuntimeException; +import org.apache.flink.table.catalog.ObjectIdentifier; +import org.apache.flink.table.data.ArrayData; +import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.GenericArrayData; +import org.apache.flink.table.data.GenericMapData; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.MapData; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.data.binary.BinaryStringData; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.DistinctType; +import org.apache.flink.table.types.logical.IntType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.types.logical.StructuredType; +import org.apache.flink.table.types.logical.StructuredType.StructuredAttribute; +import org.apache.flink.table.types.logical.utils.UuidUtils; +import org.apache.flink.types.variant.BinaryVariantInternalBuilder; +import org.apache.flink.types.variant.BinaryVariantInternalBuilder.FieldEntry; +import org.apache.flink.types.variant.Variant; +import org.apache.flink.types.variant.VariantBuilder; +import org.apache.flink.util.InstantiationUtil; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.EnumSource; +import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.ValueSource; + +import java.math.BigDecimal; +import java.nio.ByteBuffer; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Function; +import java.util.stream.Stream; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.apache.flink.table.api.DataTypes.FIELD; +import static org.apache.flink.table.data.utils.ToVariantConverter.MAX_PAYLOAD_BYTES; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class ToVariantConverterTest { + + private static final VariantBuilder BUILDER = Variant.newBuilder(); + + private static final String REPLACEMENT_CHARACTER = "\uFFFD"; + + private static final LocalDateTime DATE_TIME = + LocalDateTime.of(2026, 9, 25, 10, 15, 30, 123_000_000); + + private static final UUID UUID_VALUE = UUID.fromString("0e1f2a3b-4c5d-6e7f-8091-a2b3c4d5e6f7"); + + private static final LogicalType LEAVES = + DataTypes.ROW( + FIELD("bool", DataTypes.BOOLEAN()), + FIELD("tiny", DataTypes.TINYINT()), + FIELD("small", DataTypes.SMALLINT()), + FIELD("int", DataTypes.INT()), + FIELD("big", DataTypes.BIGINT()), + FIELD("float", DataTypes.FLOAT()), + FIELD("double", DataTypes.DOUBLE()), + FIELD("decimal", DataTypes.DECIMAL(10, 2)), + FIELD("char", DataTypes.CHAR(4)), + FIELD("bytes", DataTypes.BYTES()), + FIELD("date", DataTypes.DATE()), + FIELD("time", DataTypes.TIME()), + FIELD("ts", DataTypes.TIMESTAMP(3)), + FIELD("tsNanos", DataTypes.TIMESTAMP(9)), + FIELD("ltz", DataTypes.TIMESTAMP_LTZ(6)), + FIELD("ltzNanos", DataTypes.TIMESTAMP_LTZ(9)), + FIELD("uuid", DataTypes.UUID())) + .getLogicalType(); + + private static final LogicalType MONEY = + DistinctType.newBuilder(ObjectIdentifier.of("cat", "db", "Money"), new IntType()) + .build(); + + private static final LogicalType ORDER = + StructuredType.newBuilder(ObjectIdentifier.of("cat", "db", "Order")) + .attributes( + List.of( + new StructuredAttribute("id", new IntType()), + new StructuredAttribute("price", MONEY))) + .build(); + + @Test + void testLeavesKeepTheKindOfTheirType() { + final Variant variant = convert(LEAVES, leaves()); + + assertThat(variant.getField("bool").getBoolean()).isTrue(); + assertThat(variant.getField("tiny").getType()).isEqualTo(Variant.Type.TINYINT); + assertThat(variant.getField("small").getType()).isEqualTo(Variant.Type.SMALLINT); + assertThat(variant.getField("int").getType()).isEqualTo(Variant.Type.INT); + assertThat(variant.getField("big").getType()).isEqualTo(Variant.Type.BIGINT); + assertThat(variant.getField("float").getFloat()).isEqualTo(1.5f); + assertThat(variant.getField("double").getDouble()).isEqualTo(2.5d); + assertThat(variant.getField("decimal").getDecimal()).isEqualByComparingTo("12.50"); + assertThat(variant.getField("char").getString()).isEqualTo("ab "); + assertThat(variant.getField("bytes").getBytes()).containsExactly((byte) 0xCA, (byte) 0xFE); + assertThat(variant.getField("date").getDate()).isEqualTo(LocalDate.of(2026, 9, 25)); + assertThat(variant.getField("time").getTime()).isEqualTo(LocalTime.of(10, 15, 30)); + assertThat(variant.getField("ts").getType()).isEqualTo(Variant.Type.TIMESTAMP); + assertThat(variant.getField("ts").getDateTime()).isEqualTo(DATE_TIME); + assertThat(variant.getField("tsNanos").getType()).isEqualTo(Variant.Type.TIMESTAMP_NS); + assertThat(variant.getField("ltz").getType()).isEqualTo(Variant.Type.TIMESTAMP_LTZ); + assertThat(variant.getField("ltzNanos").getType()).isEqualTo(Variant.Type.TIMESTAMP_LTZ_NS); + assertThat(variant.getField("uuid").getUuid()).isEqualTo(UUID_VALUE); + } + + @Test + void testNullBecomesVariantNull() { + final LogicalType type = + DataTypes.ROW( + FIELD("a", DataTypes.INT()), + FIELD("list", DataTypes.ARRAY(DataTypes.INT())), + FIELD("map", DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()))) + .getLogicalType(); + final GenericRowData row = + GenericRowData.of( + null, + new GenericArrayData(new Object[] {1, null}), + mapOf(new Object[] {StringData.fromString("k")}, new Object[] {null})); + + assertThat(convert(type, row).toJson()) + .isEqualTo("{\"a\":null,\"list\":[1,null],\"map\":{\"k\":null}}"); + final GenericArrayData nullElement = new GenericArrayData(new Object[] {null}); + assertThat(toJson(DataTypes.ARRAY(DataTypes.NULL()), nullElement)).isEqualTo("[null]"); + final GenericRowData nullField = GenericRowData.of((Object) null); + assertThat(toJson(DataTypes.ROW(FIELD("n", DataTypes.NULL())), nullField)) + .isEqualTo("{\"n\":null}"); + } + + @Test + void testEmptyValues() { + assertThat(toJson(DataTypes.ARRAY(DataTypes.INT()), new GenericArrayData(new Object[0]))) + .isEqualTo("[]"); + final GenericMapData emptyMap = new GenericMapData(Map.of()); + assertThat(toJson(DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()), emptyMap)) + .isEqualTo("{}"); + assertThat(convert(new RowType(List.of()), new GenericRowData(0)).toJson()).isEqualTo("{}"); + } + + @Test + void testDuplicateMapKeyKeepsTheLastValue() { + final MapData map = + mapOf( + new Object[] {StringData.fromString("a"), StringData.fromString("a")}, + new Object[] {1, 2}); + + final DataType type = DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()); + + assertThat(convert(type, map).toJson()).isEqualTo("{\"a\":2}"); + // a builder that rejects duplicate keys fails instead + final ToVariantConverter converter = ToVariantConverter.create(type.getLogicalType()); + final BinaryVariantInternalBuilder strict = new BinaryVariantInternalBuilder(false); + assertThatThrownBy(() -> converter.writeTo(strict, map)) + .isSameAs(BinaryVariantInternalBuilder.VARIANT_DUPLICATE_KEY_EXCEPTION); + } + + @Test + void testNullMapKeyFails() { + final MapData map = mapOf(new Object[] {null}, new Object[] {1}); + + assertThatThrownBy(() -> convert(DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()), map)) + .isInstanceOf(TableRuntimeException.class) + .hasMessage( + "Cannot cast a value of type MAP<STRING, INT> with a NULL key to VARIANT. " + + "A VARIANT object key cannot be NULL."); + } + + @Test + void testStructuredAndDistinctTypes() { + final Variant variant = convert(ORDER, GenericRowData.of(1, 42)); + + assertThat(variant.toJson()).isEqualTo("{\"id\":1,\"price\":42}"); + assertThat(variant.getField("price").getType()).isEqualTo(Variant.Type.INT); + } + + @Test + void testLeafHoldsUpToTheSizeLimit() { + final byte[] bytes = new byte[MAX_PAYLOAD_BYTES]; + assertThat(convert(DataTypes.BYTES(), bytes).getBytes()).hasSize(MAX_PAYLOAD_BYTES); + final StringData string = StringData.fromString("x".repeat(MAX_PAYLOAD_BYTES)); + assertThat(convert(DataTypes.STRING(), string).getString()).hasSize(MAX_PAYLOAD_BYTES); + } + + @Test + void testLeafOverTheSizeLimitFails() { + final byte[] bytes = new byte[MAX_PAYLOAD_BYTES + 1]; + assertThatThrownBy(() -> convert(DataTypes.BYTES(), bytes)) + .isInstanceOf(TableRuntimeException.class) + .hasMessage( + "Cannot cast a binary value of 16777212 bytes to VARIANT. A VARIANT is " + + "limited to 16 MiB, so a string or binary value can have at " + + "most 16777211 bytes."); + final StringData string = StringData.fromString("x".repeat(MAX_PAYLOAD_BYTES + 1)); + assertThatThrownBy(() -> convert(DataTypes.STRING(), string)) + .isInstanceOf(TableRuntimeException.class) + .hasMessageStartingWith("Cannot cast a string value of 16777212 bytes to VARIANT."); + final byte[] tooLong = new byte[MAX_PAYLOAD_BYTES + 1]; + Arrays.fill(tooLong, (byte) 'x'); + assertThatThrownBy(() -> convert(DataTypes.STRING(), binaryString(tooLong))) + .isInstanceOf(TableRuntimeException.class) + .hasMessageStartingWith("Cannot cast a string value of 16777212 bytes to VARIANT."); + } + + @ParameterizedTest + @ValueSource(strings = {"", "hello", "Grüße, 世界 🚀"}) + void testStringKeepsItsUtf8Bytes(final String str) { + assertThat(convert(DataTypes.STRING(), binaryString(str.getBytes(UTF_8)))) + .isEqualTo(BUILDER.of(str)); + } + + @ParameterizedTest + @MethodSource("invalidUtf8") + void testInvalidUtf8BecomesReplacementCharacters(final byte[] invalid) { + final Variant variant = convert(DataTypes.STRING(), binaryString(invalid)); + + assertThat(variant.getString()).contains(REPLACEMENT_CHARACTER); + assertThat(variant).isEqualTo(BUILDER.of(binaryString(invalid).toString())); + } + + private static Stream<byte[]> invalidUtf8() { + return Stream.of( + new byte[] {'a', (byte) 0xFF, 'b'}, + new byte[] {'a', (byte) 0xC3}, + new byte[] {(byte) 0xC0, (byte) 0xAF}, + new byte[] {(byte) 0xED, (byte) 0xA0, (byte) 0x80}); + } + + @Test + void testJavaStringIsEncodedWithoutItsBinaryForm() { + // An unpaired surrogate has no UTF-8 form, so encoding stores '?' in its place. + final StringData value = StringData.fromString("a\uD800b"); + + final Variant variant = convert(DataTypes.STRING(), value); + + assertThat(variant).isEqualTo(BUILDER.of(value.toString())); + assertThat(variant.getString()).isEqualTo("a?b"); + assertThat(((BinaryStringData) value).getBinarySection()).isNull(); + } + + @ParameterizedTest + @EnumSource(Layout.class) + void testStringIsReadFromEverySegmentLayout(final Layout layout) { + final String valid = "Grüße, 世界 🚀"; + final byte[] invalid = {'a', (byte) 0xFF, 'b'}; + + assertThat(convert(DataTypes.STRING(), layout.place.apply(valid.getBytes(UTF_8)))) + .isEqualTo(BUILDER.of(layout.place.apply(valid.getBytes(UTF_8)).toString())); + assertThat(convert(DataTypes.STRING(), layout.place.apply(invalid))) + .isEqualTo(BUILDER.of(layout.place.apply(invalid).toString())); + } + + /** Where the bytes of a {@link BinaryStringData} live. */ + private enum Layout { + HEAP_SEGMENT(ToVariantConverterTest::binaryString), + OFF_HEAP_SEGMENT(ToVariantConverterTest::offHeapString), + FIRST_OF_TWO_SEGMENTS(ToVariantConverterTest::firstOfTwoSegments), + ACROSS_TWO_SEGMENTS(ToVariantConverterTest::splitString), + JAVA_OBJECT(utf8 -> StringData.fromString(new String(utf8, UTF_8))); + + private final Function<byte[], StringData> place; + + Layout(final Function<byte[], StringData> place) { + this.place = place; + } + } + + @ParameterizedTest(name = "{0} as DECIMAL({1}, {2})") + @CsvSource({ + "0, 1, 0", + "1.50, 10, 2", + "999999999, 9, 0", + "-0.999999999, 9, 9", + "1000000000, 10, 0", + "0.0000000001, 10, 10", + "-999999999999999999, 18, 0", + "0.999999999999999999, 18, 18", + "1000000000000000000, 19, 0", + "-99999999999999999999999999999999999999, 38, 0", + "0.99999999999999999999999999999999999999, 38, 38" + }) + void testDecimalKeepsUnscaledValueAndScale( + final BigDecimal decimal, final int precision, final int scale) { + assertThat( + convert( + DataTypes.DECIMAL(precision, scale), + DecimalData.fromBigDecimal(decimal, precision, scale))) + .isEqualTo(BUILDER.of(decimal)); + } + + @Test + void testValueOverTheSizeLimitFails() { + final StringData half = StringData.fromString("x".repeat(9 * 1024 * 1024)); + final ArrayData array = new GenericArrayData(new Object[] {half, half}); + + assertThatThrownBy(() -> convert(DataTypes.ARRAY(DataTypes.STRING()), array)) + .isInstanceOf(TableRuntimeException.class) + .hasMessage( + "Cannot cast a value of type ARRAY<STRING> to VARIANT. A VARIANT is " + + "limited to 16 MiB."); + } + + @Test + void testDeeplyNestedVariantFails() throws InterruptedException { + // [[[...[1]...]]], built without recursion + final BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false); + builder.appendInt(1); + for (int i = 0; i < 10_000; i++) { + builder.finishWritingArray(0, new ArrayList<>(List.of(0))); + } + final ArrayData array = new GenericArrayData(new Object[] {builder.build()}); + + // A small stack makes the overflow independent of the JVM's default stack size. + final AtomicReference<Throwable> error = new AtomicReference<>(); + final Thread thread = + new Thread( + null, + () -> { + try { + convert(DataTypes.ARRAY(DataTypes.VARIANT()), array); + } catch (Throwable t) { + error.set(t); + } + }, + "deeply-nested-variant", + 256 * 1024); + thread.start(); + thread.join(); + + assertThat(error.get()) + .isInstanceOf(TableRuntimeException.class) + .hasMessage( + "Cannot cast a value of type ARRAY<VARIANT> to VARIANT because it is " + + "nested too deeply."); + } + + @Test + void testWriteIntoASharedBuilder() { + final ToVariantConverter converter = + ToVariantConverter.create(DataTypes.ARRAY(DataTypes.INT()).getLogicalType()); + final BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false); + final int start = builder.getWritePos(); + final ArrayList<FieldEntry> fields = new ArrayList<>(); + fields.add(new FieldEntry("a", builder.addKey("a"), builder.getWritePos() - start)); + converter.writeTo(builder, new GenericArrayData(new Object[] {1, null})); + fields.add(new FieldEntry("b", builder.addKey("b"), builder.getWritePos() - start)); + converter.writeTo(builder, null); + builder.finishWritingObject(start, fields); + + assertThat(builder.build().toJson()).isEqualTo("{\"a\":[1,null],\"b\":null}"); + } + + @Test + void testTimeOutsideOneDayFails() { + assertThat(convert(DataTypes.TIME(), 86_399_999).getTime()) + .isEqualTo(LocalTime.of(23, 59, 59, 999_000_000)); + assertThatThrownBy(() -> convert(DataTypes.TIME(), -1)) + .isInstanceOf(TableRuntimeException.class) + .hasMessage( + "Cannot cast the TIME value -1 to VARIANT. A TIME holds the milliseconds " + + "of one day, from 0 to 86399999."); + assertThatThrownBy(() -> convert(DataTypes.TIME(), 86_400_000)) + .isInstanceOf(TableRuntimeException.class); + } + + @Test + void testNanosecondTimestampCoversTheWholeLongRange() { + final DataType type = DataTypes.TIMESTAMP(9); + final TimestampData min = nanosSinceEpoch(Long.MIN_VALUE); + final TimestampData max = nanosSinceEpoch(Long.MAX_VALUE); + + assertThat(convert(type, min).getDateTime()).isEqualTo(min.toLocalDateTime()); + assertThat(convert(type, max).getDateTime()).isEqualTo(max.toLocalDateTime()); + final TimestampData belowMin = + TimestampData.fromEpochMillis(min.getMillisecond(), min.getNanoOfMillisecond() - 1); + assertThatThrownBy(() -> convert(type, belowMin)) + .isInstanceOf(TableRuntimeException.class) + .hasMessageContaining("only covers 1677-09-21 to 2262-04-11"); + final TimestampData aboveMax = + TimestampData.fromEpochMillis(max.getMillisecond(), max.getNanoOfMillisecond() + 1); + assertThatThrownBy(() -> convert(type, aboveMax)) + .isInstanceOf(TableRuntimeException.class) + .hasMessageContaining("only covers 1677-09-21 to 2262-04-11"); + } + + @Test + void testCreateRejectsTypesWithoutVariantKind() { + assertThatThrownBy( + () -> + ToVariantConverter.create( + DataTypes.MULTISET(DataTypes.INT()).getLogicalType())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Cannot cast MULTISET<INT> to VARIANT."); + assertThatThrownBy( + () -> + ToVariantConverter.create( + DataTypes.MAP(DataTypes.INT(), DataTypes.STRING()) + .getLogicalType())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageEndingWith("A MAP key must be a character string."); + } + + @Test + void testConverterSurvivesJavaSerialization() throws Exception { + final LogicalType type = + DataTypes.ROW( + FIELD("leaves", DataTypes.of(LEAVES)), + FIELD("list", DataTypes.ARRAY(DataTypes.INT())), + FIELD("map", DataTypes.MAP(DataTypes.STRING(), DataTypes.INT())), + FIELD("nothing", DataTypes.NULL()), + FIELD("variant", DataTypes.VARIANT()), + FIELD("order", DataTypes.of(ORDER))) + .getLogicalType(); + final GenericRowData row = + GenericRowData.of( + leaves(), + new GenericArrayData(new Object[] {1, null}), + mapOf(new Object[] {StringData.fromString("k")}, new Object[] {2}), + null, + Variant.newBuilder().of("embedded"), + GenericRowData.of(1, 42)); + final ToVariantConverter converter = ToVariantConverter.create(type); + + assertThat(InstantiationUtil.clone(converter).convert(row)) + .isEqualTo(converter.convert(row)); + } + + private static GenericRowData leaves() { + return GenericRowData.of( + true, + (byte) 1, + (short) 2, + 3, + 4L, + 1.5f, + 2.5d, + DecimalData.fromBigDecimal(new BigDecimal("12.50"), 10, 2), + StringData.fromString("ab "), + new byte[] {(byte) 0xCA, (byte) 0xFE}, + (int) LocalDate.of(2026, 9, 25).toEpochDay(), + LocalTime.of(10, 15, 30).toSecondOfDay() * 1000, + TimestampData.fromLocalDateTime(DATE_TIME), + TimestampData.fromLocalDateTime(DATE_TIME), + TimestampData.fromEpochMillis(0), + TimestampData.fromEpochMillis(0, 1), + UuidUtils.toBytes(UUID_VALUE)); + } + + private static TimestampData nanosSinceEpoch(long nanos) { + return TimestampData.fromEpochMillis( + Math.floorDiv(nanos, 1_000_000L), (int) Math.floorMod(nanos, 1_000_000L)); + } + + /** Places the bytes inside a larger segment, the way a row field points into its row. */ + private static StringData binaryString(final byte[] utf8) { + final byte[] row = new byte[utf8.length + 2]; + System.arraycopy(utf8, 0, row, 1, utf8.length); + return BinaryStringData.fromAddress( + new MemorySegment[] {MemorySegmentFactory.wrap(row)}, 1, utf8.length); + } + + private static StringData offHeapString(final byte[] utf8) { + final MemorySegment segment = + MemorySegmentFactory.wrapOffHeapMemory(ByteBuffer.allocateDirect(utf8.length + 2)); + segment.put(1, utf8); + return BinaryStringData.fromAddress(new MemorySegment[] {segment}, 1, utf8.length); + } + + /** Places the bytes in the first of two segments, the way a short field sits in a large row. */ + private static StringData firstOfTwoSegments(final byte[] utf8) { + final byte[] first = new byte[utf8.length + 2]; + System.arraycopy(utf8, 0, first, 1, utf8.length); + final MemorySegment[] segments = { + MemorySegmentFactory.wrap(first), MemorySegmentFactory.wrap(new byte[first.length]) + }; + return BinaryStringData.fromAddress(segments, 1, utf8.length); + } + + /** Splits the bytes over two segments of the same size, the way a large row spans pages. */ + private static StringData splitString(final byte[] utf8) { + final int segmentSize = utf8.length / 2 + 1; + final byte[] row = new byte[2 * segmentSize]; + System.arraycopy(utf8, 0, row, 1, utf8.length); + final MemorySegment[] segments = { + MemorySegmentFactory.wrap(Arrays.copyOfRange(row, 0, segmentSize)), + MemorySegmentFactory.wrap(Arrays.copyOfRange(row, segmentSize, row.length)) + }; + return BinaryStringData.fromAddress(segments, 1, utf8.length); + } + + private static Variant convert(DataType dataType, Object value) { + return convert(dataType.getLogicalType(), value); + } + + private static String toJson(DataType dataType, Object value) { + return convert(dataType, value).toJson(); + } + + private static Variant convert(LogicalType type, Object value) { + return ToVariantConverter.create(type).convert(value); + } + + private static MapData mapOf(Object[] keys, Object[] values) { + return new MapData() { + @Override + public int size() { + return keys.length; + } + + @Override + public ArrayData keyArray() { + return new GenericArrayData(keys); + } + + @Override + public ArrayData valueArray() { + return new GenericArrayData(values); + } + }; + } +} diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractCodeGeneratorCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractCodeGeneratorCastRule.java index 71cdaa33563..4390717de40 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractCodeGeneratorCastRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractCodeGeneratorCastRule.java @@ -18,7 +18,6 @@ package org.apache.flink.table.planner.functions.casting; -import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.table.api.TableRuntimeException; import org.apache.flink.table.data.utils.CastExecutor; import org.apache.flink.table.planner.codegen.CodeGenUtils; @@ -28,8 +27,8 @@ import org.apache.flink.table.runtime.typeutils.InternalSerializers; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.util.FlinkRuntimeException; -import java.util.AbstractMap.SimpleImmutableEntry; import java.util.ArrayList; +import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -76,10 +75,11 @@ abstract class AbstractCodeGeneratorCastRule<IN, OUT> extends AbstractCastRule<I generateCodeBlock( ctx, inputTerm, inputIsNullTerm, inputLogicalType, targetLogicalType); - // Class fields can contain type serializers + // Type serializers and reusable objects are passed to the constructor + final Map<String, Object> constructorFields = ctx.getConstructorFields(); final String classFieldDecls = Stream.concat( - ctx.typeSerializers.values().stream() + constructorFields.entrySet().stream() .map( entry -> "private final " @@ -95,7 +95,7 @@ abstract class AbstractCodeGeneratorCastRule<IN, OUT> extends AbstractCastRule<I "public " + castExecutorClassName + "(" - + ctx.typeSerializers.values().stream() + + constructorFields.entrySet().stream() .map( entry -> className(entry.getValue().getClass()) @@ -104,7 +104,7 @@ abstract class AbstractCodeGeneratorCastRule<IN, OUT> extends AbstractCastRule<I .collect(Collectors.joining(", ")) + ")"; final String constructorBody = - ctx.getDeclaredTypeSerializers().stream() + constructorFields.keySet().stream() .map(name -> "this." + name + " = " + name + ";\n") .collect(Collectors.joining()); @@ -158,8 +158,7 @@ abstract class AbstractCodeGeneratorCastRule<IN, OUT> extends AbstractCastRule<I + "}\n}"; try { - Object[] constructorArgs = - ctx.getTypeSerializersInstances().toArray(new TypeSerializer[0]); + final Object[] constructorArgs = constructorFields.values().toArray(); return (CastExecutor<IN, OUT>) CompileUtils.compile( castRuleContext.getClassLoader(), @@ -180,8 +179,10 @@ abstract class AbstractCodeGeneratorCastRule<IN, OUT> extends AbstractCastRule<I private final CastRule.Context castRuleCtx; - private final Map<LogicalType, Map.Entry<String, TypeSerializer<?>>> typeSerializers = - new LinkedHashMap<>(); + // The objects the generated constructor receives, by field name, in the order of its + // parameters. + private final Map<String, Object> constructorFields = new LinkedHashMap<>(); + private final Map<LogicalType, String> typeSerializerFields = new HashMap<>(); private final List<String> variableDeclarationStatements = new ArrayList<>(); private final List<String> classFields = new ArrayList<>(); private int variableIndex = 0; @@ -217,18 +218,17 @@ abstract class AbstractCodeGeneratorCastRule<IN, OUT> extends AbstractCastRule<I @Override public String declareTypeSerializer(LogicalType type) { - return typeSerializers - .computeIfAbsent( - type, - t -> { - Map.Entry<String, TypeSerializer<?>> e = - new SimpleImmutableEntry<>( - "typeSerializer$" + variableIndex, - InternalSerializers.create(t)); - variableIndex++; - return e; - }) - .getKey(); + return typeSerializerFields.computeIfAbsent( + type, + t -> declareReusableObject(InternalSerializers.create(t), "typeSerializer")); + } + + @Override + public String declareReusableObject(Object object, String fieldPrefix) { + final String fieldName = fieldPrefix + "$" + variableIndex; + variableIndex++; + constructorFields.put(fieldName, object); + return fieldName; } @Override @@ -242,16 +242,8 @@ abstract class AbstractCodeGeneratorCastRule<IN, OUT> extends AbstractCastRule<I return castRuleCtx.getCodeGeneratorContext(); } - public List<String> getDeclaredTypeSerializers() { - return this.typeSerializers.values().stream() - .map(Map.Entry::getKey) - .collect(Collectors.toList()); - } - - public List<TypeSerializer<?>> getTypeSerializersInstances() { - return this.typeSerializers.values().stream() - .map(Map.Entry::getValue) - .collect(Collectors.toList()); + public Map<String, Object> getConstructorFields() { + return constructorFields; } public List<String> getClassFields() { diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractExpressionCodeGeneratorCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractExpressionCodeGeneratorCastRule.java index d51e3de2d40..feda119a1d6 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractExpressionCodeGeneratorCastRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractExpressionCodeGeneratorCastRule.java @@ -131,6 +131,12 @@ abstract class AbstractExpressionCodeGeneratorCastRule<IN, OUT> "No type serializer can be declared when using AbstractExpressionCodeGeneratorCastRule. You should use AbstractCodeGeneratorCastRule instead."); } + @Override + public String declareReusableObject(Object object, String fieldPrefix) { + throw new UnsupportedOperationException( + "No reusable object can be declared when using AbstractExpressionCodeGeneratorCastRule. You should use AbstractCodeGeneratorCastRule instead."); + } + @Override public String declareClassField(String type, String field, String initialization) { throw new UnsupportedOperationException( diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java index e3e07fd0941..54f3af8679a 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java @@ -101,7 +101,7 @@ public class CastRuleProvider { .addRule(VariantToArrayCastRule.INSTANCE) .addRule(VariantToRowCastRule.INSTANCE) .addRule(VariantToMapCastRule.INSTANCE) - .addRule(PrimitiveToVariantCastRule.INSTANCE) + .addRule(ToVariantCastRule.INSTANCE) // Bitmap rules .addRule(BitmapToStringCastRule.INSTANCE) .addRule(BitmapToBinaryCastRule.INSTANCE) diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleUtils.java index cda0bd3ddf3..d302bcd1f01 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleUtils.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleUtils.java @@ -158,7 +158,7 @@ final class CastRuleUtils { case DOUBLE: return staticCall(Double.class, "valueOf", term); case DISTINCT_TYPE: - box(term, ((DistinctType) type).getSourceType()); + return box(term, ((DistinctType) type).getSourceType()); } return term; } diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CodeGeneratorCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CodeGeneratorCastRule.java index 574b8bf4ccd..17a56dcc623 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CodeGeneratorCastRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CodeGeneratorCastRule.java @@ -75,6 +75,15 @@ public interface CodeGeneratorCastRule<IN, OUT> extends CastRule<IN, OUT> { */ String declareTypeSerializer(LogicalType type); + /** + * Declare a field that holds the given object. The class of the object must be public, + * because the generated code refers to it by name. The object must be serializable, because + * the planner copies it into the generated class and ships it with that class. + * + * @return the field term + */ + String declareReusableObject(Object object, String fieldPrefix); + /** * @return field term. The field is going to be declared as final. */ diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/PrimitiveToVariantCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/PrimitiveToVariantCastRule.java deleted file mode 100644 index 12c32e7324b..00000000000 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/PrimitiveToVariantCastRule.java +++ /dev/null @@ -1,179 +0,0 @@ -/* - * 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.flink.table.planner.functions.casting; - -import org.apache.flink.table.runtime.functions.VariantCastUtils; -import org.apache.flink.table.types.logical.LogicalType; -import org.apache.flink.table.types.logical.LogicalTypeRoot; -import org.apache.flink.table.types.logical.utils.LogicalTypeChecks; -import org.apache.flink.types.variant.Variant; - -import static org.apache.flink.table.planner.functions.casting.CastRuleUtils.staticCall; - -/** - * Cast rule from a primitive type to {@link LogicalTypeRoot#VARIANT}. - * - * <p>The value keeps the kind of its SQL type. For example, a {@code BIGINT} is stored as a {@code - * BIGINT}, even when it would fit a smaller kind. A {@code NaN} or infinite {@code FLOAT} or {@code - * DOUBLE} is stored as is, although {@code PARSE_JSON} rejects it. Some types can fail for some - * values, see {@link #canFail}. - */ -class PrimitiveToVariantCastRule extends AbstractExpressionCodeGeneratorCastRule<Object, Variant> { - - static final PrimitiveToVariantCastRule INSTANCE = new PrimitiveToVariantCastRule(); - - /** A character takes up to 4 bytes in UTF-8, which a declared length counts as one. */ - private static final int MAX_UTF8_BYTES_PER_CHAR = 4; - - private PrimitiveToVariantCastRule() { - super( - CastRulePredicate.builder() - .predicate( - (input, target) -> - target.is(LogicalTypeRoot.VARIANT) - && isSupportedSource(input)) - .build()); - } - - private static boolean isSupportedSource(LogicalType inputType) { - switch (inputType.getTypeRoot()) { - case BOOLEAN: - case TINYINT: - case SMALLINT: - case INTEGER: - case BIGINT: - case FLOAT: - case DOUBLE: - case DECIMAL: - case CHAR: - case VARCHAR: - case BINARY: - case VARBINARY: - case DATE: - case TIME_WITHOUT_TIME_ZONE: - case TIMESTAMP_WITHOUT_TIME_ZONE: - case TIMESTAMP_WITH_LOCAL_TIME_ZONE: - case UUID: - return true; - default: - return false; - } - } - - /** - * Returns whether a value of the input type can fail the cast, so that {@code TRY_CAST} returns - * {@code NULL} for it instead of failing. Two kinds of types can fail: - * - * <ul> - * <li>A {@code TIMESTAMP(p)} or {@code TIMESTAMP_LTZ(p)} with {@code p > 6} is stored with - * nanoseconds, which only cover 1677-09-21 to 2262-04-11. - * <li>A string or binary value over {@link VariantCastUtils#MAX_PAYLOAD_BYTES} does not fit - * into a {@code VARIANT}. Only a type whose declared length allows such a value can fail. - * </ul> - * - * <p>Every other type never fails. - * - * <p>The length check trusts the declared length. Flink does not enforce that length on values - * from a source, so a longer value can still reach the cast. {@code TRY_CAST} then fails for it - * instead of returning {@code NULL}. - */ - @Override - public boolean canFail(LogicalType inputLogicalType, LogicalType targetLogicalType) { - switch (inputLogicalType.getTypeRoot()) { - case TIMESTAMP_WITHOUT_TIME_ZONE: - case TIMESTAMP_WITH_LOCAL_TIME_ZONE: - return LogicalTypeChecks.getPrecision(inputLogicalType) - > VariantCastUtils.TIMESTAMP_PRECISION; - case CHAR: - case VARCHAR: - return (long) MAX_UTF8_BYTES_PER_CHAR - * LogicalTypeChecks.getLength(inputLogicalType) - > VariantCastUtils.MAX_PAYLOAD_BYTES; - case BINARY: - case VARBINARY: - return LogicalTypeChecks.getLength(inputLogicalType) - > VariantCastUtils.MAX_PAYLOAD_BYTES; - default: - return false; - } - } - - /* Example generated code for INT: - - result$0 = org.apache.flink.table.runtime.functions.VariantCastUtils.fromIntegral(int$0); - - */ - @Override - public String generateExpression( - CodeGeneratorCastRule.Context context, - String inputTerm, - LogicalType inputLogicalType, - LogicalType targetLogicalType) { - if (isTimestamp(inputLogicalType)) { - return staticCall( - VariantCastUtils.class, - helperName(inputLogicalType), - inputTerm, - LogicalTypeChecks.getPrecision(inputLogicalType)); - } - return staticCall(VariantCastUtils.class, helperName(inputLogicalType), inputTerm); - } - - private static boolean isTimestamp(LogicalType inputLogicalType) { - return inputLogicalType.is(LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE) - || inputLogicalType.is(LogicalTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE); - } - - private static String helperName(LogicalType inputLogicalType) { - switch (inputLogicalType.getTypeRoot()) { - case BOOLEAN: - return "fromBoolean"; - case TINYINT: - case SMALLINT: - case INTEGER: - case BIGINT: - return "fromIntegral"; - case FLOAT: - return "fromFloat"; - case DOUBLE: - return "fromDouble"; - case DECIMAL: - return "fromDecimal"; - case CHAR: - case VARCHAR: - return "fromString"; - case BINARY: - case VARBINARY: - return "fromBytes"; - case DATE: - return "fromDate"; - case TIME_WITHOUT_TIME_ZONE: - return "fromTime"; - case TIMESTAMP_WITHOUT_TIME_ZONE: - return "fromTimestamp"; - case TIMESTAMP_WITH_LOCAL_TIME_ZONE: - return "fromTimestampLtz"; - case UUID: - return "fromUuid"; - default: - throw new IllegalArgumentException( - "Unsupported source type for casting to VARIANT: " + inputLogicalType); - } - } -} diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/ToVariantCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/ToVariantCastRule.java new file mode 100644 index 00000000000..1a75db9dc7b --- /dev/null +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/ToVariantCastRule.java @@ -0,0 +1,123 @@ +/* + * 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.flink.table.planner.functions.casting; + +import org.apache.flink.table.data.utils.ToVariantConverter; +import org.apache.flink.table.types.logical.DistinctType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.types.logical.utils.LogicalTypeCasts; +import org.apache.flink.table.types.logical.utils.LogicalTypeChecks; +import org.apache.flink.types.variant.Variant; + +import static org.apache.flink.table.planner.functions.casting.CastRuleUtils.box; +import static org.apache.flink.table.planner.functions.casting.CastRuleUtils.methodCall; + +/** + * Cast rule to {@link LogicalTypeRoot#VARIANT} from every type that {@link LogicalTypeCasts} + * allows. + * + * <p>The generated code calls a {@link ToVariantConverter} created once for the input type, which + * describes how a value is stored. See {@link #canFail} for when a cast can fail. + */ +class ToVariantCastRule extends AbstractNullAwareCodeGeneratorCastRule<Object, Variant> { + + static final ToVariantCastRule INSTANCE = new ToVariantCastRule(); + + /** A character takes up to 4 bytes in UTF-8, which a declared length counts as one. */ + private static final int MAX_UTF8_BYTES_PER_CHAR = 4; + + private ToVariantCastRule() { + super(CastRulePredicate.builder().predicate(ToVariantCastRule::matches).build()); + } + + private static boolean matches(LogicalType input, LogicalType target) { + // No cast rule takes the NULL type, although LogicalTypeCasts allows it to cast to any + // type. + return target.is(LogicalTypeRoot.VARIANT) + && !input.is(LogicalTypeRoot.NULL) + && LogicalTypeCasts.supportsExplicitCast(input, target); + } + + /** + * Returns whether a value of the input type can fail the cast, so that {@code TRY_CAST} returns + * {@code NULL} for it instead of failing. Three kinds of types can fail: + * + * <ul> + * <li>A {@code TIMESTAMP(p)} or {@code TIMESTAMP_LTZ(p)} with {@code p > 6} is stored with + * nanoseconds, which only cover 1677-09-21 to 2262-04-11. + * <li>A {@code TIME} holds the milliseconds of one day. Only a source or a function can + * produce a value outside that range, but such a value fails. + * <li>A string or binary value over {@link ToVariantConverter#MAX_PAYLOAD_BYTES} does not fit + * into a {@code VARIANT}. Only a type whose declared length allows such a value can fail. + * </ul> + * + * <p>Every other type never fails. + * + * <p>The length check trusts the declared length. Flink does not enforce that length on values + * from a source, so a longer value can still reach the cast. {@code TRY_CAST} then fails for it + * instead of returning {@code NULL}. + */ + @Override + public boolean canFail(LogicalType inputLogicalType, LogicalType targetLogicalType) { + LogicalType type = inputLogicalType; + while (type.is(LogicalTypeRoot.DISTINCT_TYPE)) { + type = ((DistinctType) type).getSourceType(); + } + switch (type.getTypeRoot()) { + case TIME_WITHOUT_TIME_ZONE: + return true; + case TIMESTAMP_WITHOUT_TIME_ZONE: + case TIMESTAMP_WITH_LOCAL_TIME_ZONE: + return LogicalTypeChecks.getPrecision(type) > ToVariantConverter.TIMESTAMP_PRECISION; + case CHAR: + case VARCHAR: + return (long) MAX_UTF8_BYTES_PER_CHAR * LogicalTypeChecks.getLength(type) + > ToVariantConverter.MAX_PAYLOAD_BYTES; + case BINARY: + case VARBINARY: + return LogicalTypeChecks.getLength(type) > ToVariantConverter.MAX_PAYLOAD_BYTES; + default: + return false; + } + } + + /* Example generated code for INT, inside the null check of the base class. The converter is a + field, created once when the code is generated: + + result$1 = toVariantConverter$2.convert(java.lang.Integer.valueOf(int$0)); + + */ + @Override + protected String generateCodeBlockInternal( + CodeGeneratorCastRule.Context context, + String inputTerm, + String returnVariable, + LogicalType inputLogicalType, + LogicalType targetLogicalType) { + final String converterTerm = + context.declareReusableObject( + ToVariantConverter.create(inputLogicalType), "toVariantConverter"); + return new CastRuleUtils.CodeWriter() + .assignStmt( + returnVariable, + methodCall(converterTerm, "convert", box(inputTerm, inputLogicalType))) + .toString(); + } +} diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala index 8b08be95ecb..643e1f9b9dd 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala @@ -2098,6 +2098,8 @@ object ScalarOperatorGens { ctx.addReusableLocalVariable(ty, variablePrefix) override def declareTypeSerializer(ty: LogicalType): String = ctx.addReusableTypeSerializer(ty) + override def declareReusableObject(obj: AnyRef, fieldPrefix: String): String = + ctx.addReusableObject(obj, fieldPrefix) override def declareClassField(ty: String, field: String, init: String): String = { ctx.addReusableMember(s"private $ty $field = $init;") field diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java index d7f4dc4a2a0..c3d2d4c2353 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java @@ -19,10 +19,12 @@ package org.apache.flink.table.planner.functions; import org.apache.flink.configuration.Configuration; +import org.apache.flink.table.annotation.DataTypeHint; import org.apache.flink.table.api.TableException; import org.apache.flink.table.api.TableRuntimeException; import org.apache.flink.table.api.config.TableConfigOptions; import org.apache.flink.table.functions.BuiltInFunctionDefinitions; +import org.apache.flink.table.functions.ScalarFunction; import org.apache.flink.table.types.AbstractDataType; import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.LogicalType; @@ -551,6 +553,17 @@ public class CastFunctionITCase extends BuiltInFunctionTestBase { .testSqlResult( "TRY_CAST(REPEAT(f0, 17000000) AS VARIANT)", null, VARIANT()), // nanoseconds only cover 1677-09-21 to 2262-04-11, while microseconds hold any year + // only a source or a function can produce a TIME outside one day + TestSetSpec.forExpression("Cast a TIME outside one day to VARIANT") + .onFieldsWithData(-1000) + .andDataTypes(INT()) + .withFunction(TimeOfMillisFunction.class) + .testSqlRuntimeError( + "CAST(TimeOfMillisFunction(f0) AS VARIANT)", + TableRuntimeException.class, + "A TIME holds the milliseconds of one day") + .testSqlResult( + "TRY_CAST(TimeOfMillisFunction(f0) AS VARIANT)", null, VARIANT()), TestSetSpec.forExpression("Cast a late TIMESTAMP to VARIANT") .onFieldsWithData( LocalDateTime.parse("3000-01-01T00:00"), @@ -2482,4 +2495,12 @@ public class CastFunctionITCase extends BuiltInFunctionTestBase { private static boolean isTimestampToNumeric(LogicalType srcType, LogicalType trgType) { return srcType.is(LogicalTypeFamily.TIMESTAMP) && trgType.is(LogicalTypeFamily.NUMERIC); } + + /** Passes its argument through as the milliseconds of the day of a {@code TIME}. */ + public static class TimeOfMillisFunction extends ScalarFunction { + public @DataTypeHint(value = "TIME(0)", bridgedTo = Integer.class) Integer eval( + Integer millisOfDay) { + return millisOfDay; + } + } } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRuleProviderTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRuleProviderTest.java index 6849d79e6ad..788d849c50a 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRuleProviderTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRuleProviderTest.java @@ -154,10 +154,9 @@ class CastRuleProviderTest { @Test void testResolvePrimitiveToVariant() { - assertThat(CastRuleProvider.resolve(INT, VARIANT)) - .isSameAs(PrimitiveToVariantCastRule.INSTANCE); + assertThat(CastRuleProvider.resolve(INT, VARIANT)).isSameAs(ToVariantCastRule.INSTANCE); assertThat(CastRuleProvider.resolve(STRING_TYPE, VARIANT)) - .isSameAs(PrimitiveToVariantCastRule.INSTANCE); + .isSameAs(ToVariantCastRule.INSTANCE); assertThat(CastRuleProvider.exists(DECIMAL(10, 2).getLogicalType(), VARIANT)).isTrue(); assertThat(CastRuleProvider.exists(TIMESTAMP(9).getLogicalType(), VARIANT)).isTrue(); assertThat(CastRuleProvider.exists(TIMESTAMP_LTZ().getLogicalType(), VARIANT)).isTrue(); @@ -165,7 +164,7 @@ class CastRuleProviderTest { assertThat(CastRuleProvider.exists(BYTES().getLogicalType(), VARIANT)).isTrue(); assertThat(CastRuleProvider.exists(UUID().getLogicalType(), VARIANT)).isTrue(); - // only a nanosecond timestamp can fail, since that kind covers a limited range of years + // a nanosecond timestamp covers a limited range of years, and a TIME only one day assertThat(CastRuleProvider.canFail(INT, VARIANT)).isFalse(); assertThat(CastRuleProvider.canFail(DOUBLE().getLogicalType(), VARIANT)).isFalse(); assertThat(CastRuleProvider.canFail(FLOAT().getLogicalType(), VARIANT)).isFalse(); @@ -173,6 +172,17 @@ class CastRuleProviderTest { assertThat(CastRuleProvider.canFail(TIMESTAMP(9).getLogicalType(), VARIANT)).isTrue(); assertThat(CastRuleProvider.canFail(TIMESTAMP_LTZ(3).getLogicalType(), VARIANT)).isFalse(); assertThat(CastRuleProvider.canFail(TIMESTAMP_LTZ(7).getLogicalType(), VARIANT)).isTrue(); + assertThat(CastRuleProvider.canFail(TIME().getLogicalType(), VARIANT)).isTrue(); + // a DISTINCT type fails like its source type + assertThat(CastRuleProvider.canFail(DISTINCT_INT, VARIANT)).isFalse(); + assertThat( + CastRuleProvider.canFail( + DistinctType.newBuilder( + ObjectIdentifier.of("a", "b", "c"), + TIMESTAMP(9).getLogicalType()) + .build(), + VARIANT)) + .isTrue(); // a string or binary type fails only when its declared length allows a value over 16 MiB, // and a character takes up to 4 bytes in UTF-8 diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java index 33015cef9e1..abc7ec972e2 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java @@ -38,6 +38,7 @@ import org.apache.flink.table.data.utils.CastExecutor; import org.apache.flink.table.planner.codegen.CodeGeneratorContext; import org.apache.flink.table.planner.functions.CastFunctionITCase; import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.DistinctType; import org.apache.flink.table.types.logical.StructuredType; import org.apache.flink.table.utils.DateTimeUtils; import org.apache.flink.types.bitmap.Bitmap; @@ -371,6 +372,12 @@ class CastRulesTest { "d", ARRAY(STRING()).getLogicalType()))) .build()); + private static final DataType MY_DISTINCT_INT = + DataTypes.of( + DistinctType.newBuilder( + ObjectIdentifier.of("a", "b", "money"), INT().getLogicalType()) + .build()); + // Rebuilds a variant object field at position 0, matching the form a ROW cast produces when it // serializes each field with BinaryRowWriter.writeVariant and reads it back. private static Variant rowFieldVariant(Variant fieldView) { @@ -2078,7 +2085,15 @@ class CastRulesTest { TimestampData.fromInstant(Instant.parse("1500-01-01T00:00:00Z")), TableRuntimeException.class, "1677-09-21 to 2262-04-11") + // a TIME holds the milliseconds of one day + .fail( + TIME(), + -1, + TableRuntimeException.class, + "A TIME holds the milliseconds of one day") .fromCase(UUID(), UUID_BYTES, VARIANT_BUILDER.of(UUID_VALUE)) + // a DISTINCT type is stored like its source type + .fromCase(MY_DISTINCT_INT, 42, VARIANT_BUILDER.of(42)) // a SQL NULL stays a SQL NULL rather than becoming a variant null .fromCase(INT(), null, null) .fromCase(STRING(), null, null), diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java index 12a4fdd1fda..4ff53cc0b93 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java @@ -19,33 +19,30 @@ package org.apache.flink.table.runtime.functions; import org.apache.flink.annotation.Internal; -import org.apache.flink.core.memory.MemorySegment; import org.apache.flink.table.api.TableRuntimeException; import org.apache.flink.table.data.DecimalData; -import org.apache.flink.table.data.StringData; import org.apache.flink.table.data.TimestampData; import org.apache.flink.table.data.binary.BinaryStringData; import org.apache.flink.table.data.binary.BinaryStringDataUtil; import org.apache.flink.table.data.binary.StringUtf8Utils; +import org.apache.flink.table.data.utils.ToVariantConverter; import org.apache.flink.table.types.logical.utils.UuidUtils; import org.apache.flink.table.utils.DateTimeUtils; import org.apache.flink.table.utils.EncodingUtils; -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.types.variant.VariantBuilder; import org.apache.flink.types.variant.VariantTypeException; import java.math.BigDecimal; import java.math.RoundingMode; import java.nio.charset.StandardCharsets; -import java.time.LocalDate; -import java.time.LocalTime; import java.util.Arrays; import java.util.TimeZone; +import static org.apache.flink.table.data.utils.ToVariantConverter.TIMESTAMP_PRECISION; + /** - * Runtime helpers for casting between a {@code VARIANT} value and a SQL type. + * Runtime helpers for casting a {@code VARIANT} value to a SQL type. * * <p>A cast never reinterprets the stored value: one kind is not read as another, and a numeric * value is never wrapped or rounded to make it fit. Any numeric kind therefore reaches an integer @@ -58,8 +55,7 @@ import java.util.TimeZone; * beyond the target precision are truncated, and the fixed width targets {@code CHAR(n)} and {@code * BINARY(n)} pad a shorter value. * - * <p>A cast into {@code VARIANT} keeps the kind of the SQL type, so an integer keeps its width and - * a timestamp keeps the precision it was declared with. + * <p>{@link ToVariantConverter} implements the cast into {@code VARIANT}. */ @Internal public final class VariantCastUtils { @@ -71,16 +67,6 @@ public final class VariantCastUtils { */ private static final double LONG_MAGNITUDE_LIMIT = -(double) Long.MIN_VALUE; - /** The highest timestamp precision a {@code VARIANT} stores with microseconds. */ - public static final int TIMESTAMP_PRECISION = 6; - - /** - * The largest string or binary value a {@code VARIANT} holds, in bytes. A {@code VARIANT} is - * limited to 16 MiB, and a long string or binary value spends 5 bytes of that on its header. - */ - public static final int MAX_PAYLOAD_BYTES = - BinaryVariantUtil.SIZE_LIMIT - 1 - BinaryVariantUtil.U32_SIZE; - private static final int TIMESTAMP_NANOS_PRECISION = 9; // TIME is millisecond-of-day at runtime, so it renders with three fractional-second digits. @@ -88,8 +74,6 @@ public final class VariantCastUtils { private static final String NULL_NODE = "NULL"; - private static final VariantBuilder BUILDER = Variant.newBuilder(); - private VariantCastUtils() {} /** @@ -560,178 +544,6 @@ public final class VariantCastUtils { } } - public static Variant fromBoolean(boolean value) { - return BUILDER.of(value); - } - - /** - * Stores an integer in the integer kind of its SQL type, so a {@code BIGINT} stays a {@code - * BIGINT} whatever its value. The generated code passes the primitive of the source type, which - * selects the overload. - */ - public static Variant fromIntegral(byte value) { - return BUILDER.of(value); - } - - public static Variant fromIntegral(short value) { - return BUILDER.of(value); - } - - public static Variant fromIntegral(int value) { - return BUILDER.of(value); - } - - public static Variant fromIntegral(long value) { - return BUILDER.of(value); - } - - public static Variant fromFloat(float value) { - return BUILDER.of(value); - } - - public static Variant fromDouble(double value) { - return BUILDER.of(value); - } - - public static Variant fromDecimal(DecimalData value) { - final BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false); - appendDecimal(builder, value); - return builder.build(); - } - - /** Like {@link #fromDecimal(DecimalData)}, but writes into a shared builder. */ - static void appendDecimal(BinaryVariantInternalBuilder builder, DecimalData value) { - if (value.isCompact()) { - builder.appendDecimal(value.toUnscaledLong(), value.scale()); - } else { - builder.appendDecimal(value.toBigDecimal()); - } - } - - /** - * Stores the UTF-8 bytes of the string as they are. A {@link StringData} may hold invalid - * UTF-8, which the variant spec does not allow, so such a value is decoded first and every - * malformed sequence is stored as the U+FFFD replacement character. - */ - public static Variant fromString(StringData value) { - final BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false); - try { - appendString(builder, value); - return builder.build(); - } catch (VariantTypeException e) { - throw sizeLimitExceeded(e, "string", ((BinaryStringData) value).getSizeInBytes()); - } - } - - /** Like {@link #fromString(StringData)}, but writes into a shared builder. */ - static void appendString(BinaryVariantInternalBuilder builder, StringData value) { - final BinaryStringData string = (BinaryStringData) value; - if (string.getBinarySection() == null) { - // A string that only exists as a Java object, such as a literal or a function result, - // is valid UTF-8 once encoded. Encoding it here skips the binary form and the check. - builder.appendString(string.getJavaObject()); - return; - } - final MemorySegment[] segments = string.getSegments(); - final int length = string.getSizeInBytes(); - final byte[] utf8; - final int offset; - // Read a string that lies in the first heap segment in place, which saves copying it out. - if (!segments[0].isOffHeap() && string.getOffset() + length <= segments[0].size()) { - utf8 = segments[0].getHeapMemory(); - offset = string.getOffset(); - } else { - utf8 = string.toBytes(); - offset = 0; - } - if (StringUtf8Utils.firstInvalidUtf8ByteIndex(utf8, offset, length) >= 0) { - // The same decoding StringData#toString falls back to for invalid UTF-8. - builder.appendString(new String(utf8, offset, length, StandardCharsets.UTF_8)); - } else { - builder.appendString(utf8, offset, length); - } - } - - public static Variant fromBytes(byte[] value) { - try { - return BUILDER.of(value); - } catch (VariantTypeException e) { - throw sizeLimitExceeded(e, "binary", value.length); - } - } - - private static RuntimeException sizeLimitExceeded( - VariantTypeException e, String kind, int bytes) { - if (e != BinaryVariantInternalBuilder.VARIANT_SIZE_LIMIT_EXCEPTION) { - return e; - } - return new TableRuntimeException( - String.format( - "Cannot cast a %s value of %d bytes to VARIANT. A VARIANT is limited to " - + "16 MiB, so a string or binary value can have at most %d bytes.", - kind, bytes, MAX_PAYLOAD_BYTES)); - } - - public static Variant fromDate(int epochDay) { - return BUILDER.of(LocalDate.ofEpochDay(epochDay)); - } - - public static Variant fromTime(int millisOfDay) { - return BUILDER.of(LocalTime.ofNanoOfDay(millisOfDay * 1_000_000L)); - } - - /** - * Stores a timestamp in the kind its declared precision needs: microseconds up to {@code - * TIMESTAMP(6)} and nanoseconds above, whatever the digits of the value. A nanosecond kind only - * covers 1677-09-21 to 2262-04-11, so a {@code TIMESTAMP(7)} to {@code TIMESTAMP(9)} value - * outside that range fails. - */ - public static Variant fromTimestamp(TimestampData value, int precision) { - final BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false); - if (precision <= TIMESTAMP_PRECISION) { - builder.appendTimestamp(timestampMicros(value)); - } else { - builder.appendTimestampNanos(timestampNanos(value, "TIMESTAMP(" + precision + ")")); - } - return builder.build(); - } - - /** Like {@link #fromTimestamp(TimestampData, int)}, for {@code TIMESTAMP_LTZ}. */ - public static Variant fromTimestampLtz(TimestampData value, int precision) { - final BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false); - if (precision <= TIMESTAMP_PRECISION) { - builder.appendTimestampLtz(timestampMicros(value)); - } else { - builder.appendTimestampLtzNanos( - timestampNanos(value, "TIMESTAMP_LTZ(" + precision + ")")); - } - return builder.build(); - } - - /** Microseconds since the epoch, which cover every year a {@link TimestampData} can hold. */ - private static long timestampMicros(TimestampData value) { - return value.getMillisecond() * 1_000L + value.getNanoOfMillisecond() / 1_000; - } - - private static long timestampNanos(TimestampData value, String sourceType) { - try { - return Math.addExact( - Math.multiplyExact(value.getMillisecond(), 1_000_000L), - value.getNanoOfMillisecond()); - } catch (ArithmeticException e) { - throw new TableRuntimeException( - String.format( - "Cannot cast the %s value %s to VARIANT. A VARIANT timestamp with " - + "nanosecond precision only covers 1677-09-21 to 2262-04-11. " - + "Cast the value to a precision of 6 or less first.", - sourceType, value)); - } - } - - public static Variant fromUuid(byte[] value) { - return BUILDER.of(UuidUtils.fromBytes(value)); - } - private static TableRuntimeException unsupportedKind(Variant variant, String targetType) { return new TableRuntimeException( String.format( diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/VariantCastUtilsTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/VariantCastUtilsTest.java index c4419de2365..f07452165d3 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/VariantCastUtilsTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/VariantCastUtilsTest.java @@ -18,41 +18,19 @@ package org.apache.flink.table.runtime.functions; -import org.apache.flink.core.memory.MemorySegment; -import org.apache.flink.core.memory.MemorySegmentFactory; -import org.apache.flink.table.api.TableRuntimeException; -import org.apache.flink.table.data.DecimalData; -import org.apache.flink.table.data.StringData; -import org.apache.flink.table.data.binary.BinaryStringData; import org.apache.flink.types.variant.BinaryVariant; import org.apache.flink.types.variant.Variant; import org.apache.flink.types.variant.VariantBuilder; import org.junit.jupiter.api.Test; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.CsvSource; -import org.junit.jupiter.params.provider.EnumSource; -import org.junit.jupiter.params.provider.MethodSource; -import org.junit.jupiter.params.provider.ValueSource; -import java.math.BigDecimal; -import java.nio.ByteBuffer; import java.time.Instant; -import java.util.Arrays; import java.util.TimeZone; -import java.util.function.Function; -import java.util.stream.Stream; -import static java.nio.charset.StandardCharsets.UTF_8; -import static org.apache.flink.table.runtime.functions.VariantCastUtils.MAX_PAYLOAD_BYTES; -import static org.apache.flink.table.runtime.functions.VariantCastUtils.fromBytes; -import static org.apache.flink.table.runtime.functions.VariantCastUtils.fromDecimal; -import static org.apache.flink.table.runtime.functions.VariantCastUtils.fromString; import static org.apache.flink.table.runtime.functions.VariantCastUtils.toPrintString; import static org.apache.flink.types.variant.BinaryVariantUtil.primitiveHeader; import static org.apache.flink.types.variant.BinaryVariantUtil.shortStrHeader; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; class VariantCastUtilsTest { @@ -60,116 +38,6 @@ class VariantCastUtilsTest { private static final VariantBuilder BUILDER = Variant.newBuilder(); - private static final String REPLACEMENT_CHARACTER = "\uFFFD"; - - @Test - void testCastToVariantHoldsUpToTheSizeLimit() { - assertThat(fromBytes(new byte[MAX_PAYLOAD_BYTES]).getBytes()).hasSize(MAX_PAYLOAD_BYTES); - assertThat(fromString(StringData.fromString("x".repeat(MAX_PAYLOAD_BYTES))).getString()) - .hasSize(MAX_PAYLOAD_BYTES); - } - - @Test - void testCastToVariantRejectsValuesOverTheSizeLimit() { - assertThatThrownBy(() -> fromBytes(new byte[MAX_PAYLOAD_BYTES + 1])) - .isInstanceOf(TableRuntimeException.class) - .hasMessage( - "Cannot cast a binary value of 16777212 bytes to VARIANT. A VARIANT is " - + "limited to 16 MiB, so a string or binary value can have at " - + "most 16777211 bytes."); - assertThatThrownBy( - () -> fromString(StringData.fromString("x".repeat(MAX_PAYLOAD_BYTES + 1)))) - .isInstanceOf(TableRuntimeException.class) - .hasMessageStartingWith("Cannot cast a string value of 16777212 bytes to VARIANT."); - final byte[] tooLong = new byte[MAX_PAYLOAD_BYTES + 1]; - Arrays.fill(tooLong, (byte) 'x'); - assertThatThrownBy(() -> fromString(binaryString(tooLong))) - .isInstanceOf(TableRuntimeException.class) - .hasMessageStartingWith("Cannot cast a string value of 16777212 bytes to VARIANT."); - } - - @ParameterizedTest - @ValueSource(strings = {"", "hello", "Grüße, 世界 🚀"}) - void testCastStringToVariantStoresItsUtf8Bytes(final String str) { - assertThat(fromString(binaryString(str.getBytes(UTF_8)))).isEqualTo(BUILDER.of(str)); - } - - @ParameterizedTest - @MethodSource("invalidUtf8") - void testCastStringToVariantReplacesInvalidUtf8(final byte[] invalid) { - final Variant variant = fromString(binaryString(invalid)); - - assertThat(variant.getString()).contains(REPLACEMENT_CHARACTER); - assertThat(variant).isEqualTo(BUILDER.of(binaryString(invalid).toString())); - } - - @Test - void testCastStringToVariantEncodesAJavaStringWithoutItsBinaryForm() { - // An unpaired surrogate has no UTF-8 form, so encoding stores '?' in its place. - final StringData value = StringData.fromString("a\uD800b"); - - final Variant variant = fromString(value); - - assertThat(variant).isEqualTo(BUILDER.of(value.toString())); - assertThat(variant.getString()).isEqualTo("a?b"); - assertThat(((BinaryStringData) value).getBinarySection()).isNull(); - } - - private static Stream<byte[]> invalidUtf8() { - return Stream.of( - new byte[] {'a', (byte) 0xFF, 'b'}, - new byte[] {'a', (byte) 0xC3}, - new byte[] {(byte) 0xC0, (byte) 0xAF}, - new byte[] {(byte) 0xED, (byte) 0xA0, (byte) 0x80}); - } - - @ParameterizedTest - @EnumSource(Layout.class) - void testCastStringToVariantReadsEverySegmentLayout(final Layout layout) { - final String valid = "Grüße, 世界 🚀"; - final byte[] invalid = {'a', (byte) 0xFF, 'b'}; - - assertThat(fromString(layout.place.apply(valid.getBytes(UTF_8)))) - .isEqualTo(BUILDER.of(layout.place.apply(valid.getBytes(UTF_8)).toString())); - assertThat(fromString(layout.place.apply(invalid))) - .isEqualTo(BUILDER.of(layout.place.apply(invalid).toString())); - } - - /** Where the bytes of a {@link BinaryStringData} live. */ - private enum Layout { - HEAP_SEGMENT(VariantCastUtilsTest::binaryString), - OFF_HEAP_SEGMENT(VariantCastUtilsTest::offHeapString), - FIRST_OF_TWO_SEGMENTS(VariantCastUtilsTest::firstOfTwoSegments), - ACROSS_TWO_SEGMENTS(VariantCastUtilsTest::splitString), - JAVA_OBJECT(utf8 -> StringData.fromString(new String(utf8, UTF_8))); - - private final Function<byte[], StringData> place; - - Layout(final Function<byte[], StringData> place) { - this.place = place; - } - } - - @ParameterizedTest(name = "{0} as DECIMAL({1}, {2})") - @CsvSource({ - "0, 1, 0", - "1.50, 10, 2", - "999999999, 9, 0", - "-0.999999999, 9, 9", - "1000000000, 10, 0", - "0.0000000001, 10, 10", - "-999999999999999999, 18, 0", - "0.999999999999999999, 18, 18", - "1000000000000000000, 19, 0", - "-99999999999999999999999999999999999999, 38, 0", - "0.99999999999999999999999999999999999999, 38, 38" - }) - void testCastDecimalToVariantKeepsUnscaledValueAndScale( - final BigDecimal decimal, final int precision, final int scale) { - assertThat(fromDecimal(DecimalData.fromBigDecimal(decimal, precision, scale))) - .isEqualTo(BUILDER.of(decimal)); - } - @Test void testPrintRendersLikeCast() { final Variant variant = @@ -216,43 +84,6 @@ class VariantCastUtilsTest { .isEqualTo("2021-09-24 14:34:56.123456"); } - /** Places the bytes inside a larger segment, the way a row field points into its row. */ - private static StringData binaryString(final byte[] utf8) { - final byte[] row = new byte[utf8.length + 2]; - System.arraycopy(utf8, 0, row, 1, utf8.length); - return BinaryStringData.fromAddress( - new MemorySegment[] {MemorySegmentFactory.wrap(row)}, 1, utf8.length); - } - - private static StringData offHeapString(final byte[] utf8) { - final MemorySegment segment = - MemorySegmentFactory.wrapOffHeapMemory(ByteBuffer.allocateDirect(utf8.length + 2)); - segment.put(1, utf8); - return BinaryStringData.fromAddress(new MemorySegment[] {segment}, 1, utf8.length); - } - - /** Places the bytes in the first of two segments, the way a short field sits in a large row. */ - private static StringData firstOfTwoSegments(final byte[] utf8) { - final byte[] first = new byte[utf8.length + 2]; - System.arraycopy(utf8, 0, first, 1, utf8.length); - final MemorySegment[] segments = { - MemorySegmentFactory.wrap(first), MemorySegmentFactory.wrap(new byte[first.length]) - }; - return BinaryStringData.fromAddress(segments, 1, utf8.length); - } - - /** Splits the bytes over two segments of the same size, the way a large row spans pages. */ - private static StringData splitString(final byte[] utf8) { - final int segmentSize = utf8.length / 2 + 1; - final byte[] row = new byte[2 * segmentSize]; - System.arraycopy(utf8, 0, row, 1, utf8.length); - final MemorySegment[] segments = { - MemorySegmentFactory.wrap(Arrays.copyOfRange(row, 0, segmentSize)), - MemorySegmentFactory.wrap(Arrays.copyOfRange(row, segmentSize, row.length)) - }; - return BinaryStringData.fromAddress(segments, 1, utf8.length); - } - /** Replaces the header of the array's second element. */ private static Variant withHeader(final BinaryVariant array, final byte header) { final byte[] value = array.getValue().clone();
