AHeise commented on code in PR #29311:
URL: https://github.com/apache/flink/pull/29311#discussion_r4144608428
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java:
##########
@@ -540,6 +551,111 @@ private static Number numeric(Variant variant, String
targetType) {
}
}
+ 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) {
+ return BUILDER.of(value.toBigDecimal());
+ }
+
+ public static Variant fromString(StringData value) {
+ return BUILDER.of(value.toString());
+ }
+
+ public static Variant fromBytes(byte[] value) {
+ return BUILDER.of(value);
+ }
+
+ 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) {
Review Comment:
nit: This is correct for pre-epoch values because `nanoOfMillisecond` is
never negative, but no test covers it on the micros path. Could you add a
`TIMESTAMP(6)` case for `1969-12-31T23:59:59.999999` (micros `-1`)?
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java:
##########
@@ -403,6 +403,33 @@ public final class LogicalTypeCasts {
.explicitFromFamily(EXACT_NUMERIC, CHARACTER_STRING)
.build();
+ //
-----------------------------------------------------------------------------------------
+ // VARIANT type
+ //
-----------------------------------------------------------------------------------------
+
+ // Only a type with a VARIANT kind that holds its value without loss
casts to VARIANT.
+ castTo(VARIANT)
Review Comment:
This also enables casts into VARIANT one level down in constructed types.
`supportsConstructedCasting` checks each child with `supportsCasting`, so
`supportsExplicitCast(ARRAY<INT>, ARRAY<VARIANT>)` flips from false to true.
`SqlCastFunction` sends ARRAY/ROW/MAP sources to `LogicalTypeCasts`, and
`ArrayToArrayCastRule`/`RowToRowCastRule`/`MapToMapAndMultisetToMultisetCastRule`
resolve the element cast to the new rule. So `CAST(ARRAY[1, 2] AS
ARRAY<VARIANT>)`, `CAST(ROW(1) AS ROW<a VARIANT>)` and `CAST(MAP['a', 1] AS
MAP<STRING, VARIANT>)` should all pass validation now.
Is that intended as part of this PR or of FLINK-40826? If it's in, it
deserves tests (`LogicalTypeCastsTest`, IT) and a line in the docs. If not, we
should reject it explicitly for now. Adding it later is additive, the same
argument as for explicit-only.
##########
docs/content/docs/sql/reference/data-types.md:
##########
@@ -1677,6 +1678,45 @@ CAST(o AS MAP<STRING, VARIANT>) -- values kept as
variants, the variant null in
CAST(o AS MAP<INT, STRING>) -- fails at validation, a MAP key must be a
character string
```
+A scalar value can also be cast to a `VARIANT` with `CAST` or `TRY_CAST`. Only
a type that a
+`VARIANT` kind holds without loss is supported, and any other type, such as
`INTERVAL`, `RAW`, or
+`BITMAP`, is rejected at validation. The value keeps the kind of its SQL type:
+
+| Input type | Stored `VARIANT` kind
|
+|--------------------------------------------|-------------------------------------------------|
+| `BOOLEAN` | `BOOLEAN`
|
+| `TINYINT`, `SMALLINT`, `INTEGER`, `BIGINT` | `TINYINT`, `SMALLINT`, `INT`,
`BIGINT` |
+| `FLOAT`, `DOUBLE`, `DECIMAL` | `FLOAT`, `DOUBLE`, `DECIMAL`
|
+| `CHAR`, `VARCHAR`, `STRING` | `STRING`
|
+| `BINARY`, `VARBINARY`, `BYTES` | `BYTES`
|
+| `DATE`, `TIME`, `UUID` | `DATE`, `TIME`, `UUID`
|
+| `TIMESTAMP`, `TIMESTAMP_LTZ` | `TIMESTAMP`, `TIMESTAMP_LTZ`
|
+
+- An integer keeps the width of its SQL type, so a `BIGINT` is stored as a
`BIGINT` even when the
+ value would fit a smaller kind. `PARSE_JSON('1')` instead picks the smallest
kind, a `TINYINT`.
+ Either way it casts back to any integer type that holds the value.
+- A character string is stored as a `STRING` and is never parsed. Use
`PARSE_JSON` to parse JSON text.
+- A `TIMESTAMP(p)` or `TIMESTAMP_LTZ(p)` keeps its declared precision. Up to a
precision of 6 it is
Review Comment:
Keeping the declared precision is consistent, but it makes a per-row runtime
failure for common sentinel values such as `9999-12-31 23:59:59` in a
`TIMESTAMP(9)` column. Did you consider falling back to micros when the value
has no sub-micro digits? I see that it mixes kinds by value, which the PR
avoids on purpose. I'm not pushing for it, just want the trade-off to be a
conscious one. Maybe someone else has an opinion.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java:
##########
@@ -432,6 +435,201 @@ private static List<TestSetSpec> variantPrimitiveCasts() {
TINYINT()));
}
+ private static List<TestSetSpec> primitiveToVariantCasts() {
+ final VariantBuilder builder = Variant.newBuilder();
+ final LocalDateTime nanos =
LocalDateTime.parse("2026-09-25T10:15:30.123456789");
+ final Instant instant = Instant.parse("2026-09-25T10:15:30.123Z");
+ return List.of(
+ // A value keeps the kind of its SQL type, so an integer keeps
its width, and a
+ // string is wrapped rather than parsed.
+ CastTestSpecBuilder.testCastTo(VARIANT())
+ .fromCase(BOOLEAN(), true, builder.of(true))
+ .fromCase(INT(), 42, builder.of(42))
+ .fromCase(BIGINT(), 1L, builder.of(1L))
+ .fromCase(BIGINT(), 10000000000L,
builder.of(10000000000L))
+ .fromCase(DOUBLE(), 1.5d, builder.of(1.5d))
+ .fromCase(
+ DECIMAL(4, 2),
+ new BigDecimal("12.50"),
+ builder.of(new BigDecimal("12.50")))
+ .fromCase(STRING(), "{\"a\":1}",
builder.of("{\"a\":1}"))
+ .fromCase(
+ DATE(),
+ LocalDate.parse("2026-09-25"),
+ builder.of(LocalDate.parse("2026-09-25")))
+ .fromCase(TIMESTAMP(9), nanos, builder.of(nanos))
+ .fromCase(TIMESTAMP_LTZ(3), instant,
builder.of(instant))
+ .fromCase(UUID(), DEFAULT_UUID,
builder.of(DEFAULT_UUID))
+ .fromCase(INT(), null, null)
+ // a type without a VARIANT kind is rejected at
validation
+ .failValidation(INTERVAL(MONTH()), Period.ofMonths(2))
+ .build(),
+ TestSetSpec.forExpression("Cast a primitive to VARIANT and
back")
Review Comment:
nit: The round trip skips TINYINT, SMALLINT, CHAR(n) and BINARY(n).
`CastRulesTest` checks the stored kind, but not the way back through
`VariantToPrimitiveCastRule`. Especially CHAR(n): are the trailing spaces kept?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java:
##########
@@ -540,6 +551,111 @@ private static Number numeric(Variant variant, String
targetType) {
}
}
+ 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) {
+ return BUILDER.of(value.toBigDecimal());
+ }
+
+ public static Variant fromString(StringData value) {
Review Comment:
nit: This runs per record. `toString()` decodes UTF-8, `appendString`
encodes it again, and `build()` copies the result once more. A `byte[]`
overload on `BinaryVariantInternalBuilder` fed from `BinaryStringData#toBytes`
would skip the round trip. The same goes for `fromDecimal` on compact decimals.
Fine as a follow-up.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/PrimitiveToVariantCastRule.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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;
+
+/**
Review Comment:
nit: This repeats the Javadoc of `VariantCastUtils#fromTimestamp` almost
verbatim. I'd keep the details there and link to it.
##########
docs/content/docs/sql/reference/data-types.md:
##########
@@ -1677,6 +1678,45 @@ CAST(o AS MAP<STRING, VARIANT>) -- values kept as
variants, the variant null in
CAST(o AS MAP<INT, STRING>) -- fails at validation, a MAP key must be a
character string
```
+A scalar value can also be cast to a `VARIANT` with `CAST` or `TRY_CAST`. Only
a type that a
+`VARIANT` kind holds without loss is supported, and any other type, such as
`INTERVAL`, `RAW`, or
+`BITMAP`, is rejected at validation. The value keeps the kind of its SQL type:
+
+| Input type | Stored `VARIANT` kind
|
+|--------------------------------------------|-------------------------------------------------|
+| `BOOLEAN` | `BOOLEAN`
|
+| `TINYINT`, `SMALLINT`, `INTEGER`, `BIGINT` | `TINYINT`, `SMALLINT`, `INT`,
`BIGINT` |
+| `FLOAT`, `DOUBLE`, `DECIMAL` | `FLOAT`, `DOUBLE`, `DECIMAL`
|
+| `CHAR`, `VARCHAR`, `STRING` | `STRING`
|
+| `BINARY`, `VARBINARY`, `BYTES` | `BYTES`
|
+| `DATE`, `TIME`, `UUID` | `DATE`, `TIME`, `UUID`
|
+| `TIMESTAMP`, `TIMESTAMP_LTZ` | `TIMESTAMP`, `TIMESTAMP_LTZ`
|
+
+- An integer keeps the width of its SQL type, so a `BIGINT` is stored as a
`BIGINT` even when the
+ value would fit a smaller kind. `PARSE_JSON('1')` instead picks the smallest
kind, a `TINYINT`.
+ Either way it casts back to any integer type that holds the value.
+- A character string is stored as a `STRING` and is never parsed. Use
`PARSE_JSON` to parse JSON text.
+- A `TIMESTAMP(p)` or `TIMESTAMP_LTZ(p)` keeps its declared precision. Up to a
precision of 6 it is
+ 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 `NaN` or infinite `FLOAT` or `DOUBLE` is stored as is, although
`PARSE_JSON` rejects them. A
Review Comment:
`JSON_STRING` isn't the only strict consumer. `RowDataToJsonConverters`
(json format) and `RawFormatSerializationSchema` both call `Variant#toJson()`,
which fails on NaN and infinity. So `INSERT INTO json_sink SELECT CAST(d AS
VARIANT) ...` fails the job on the first NaN row. Before this PR that needed an
Avro source. It's fine to allow NaN, but I'd mention the sink behavior here so
users aren't surprised at runtime.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/PrimitiveToVariantCastRule.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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;
+
+/**
+ * Primitive type to {@link LogicalTypeRoot#VARIANT} cast rule.
+ *
+ * <p>The value keeps the kind of its SQL type, so a {@code BIGINT} is stored
as a {@code BIGINT}
+ * even when it would fit a smaller one. Unlike {@code PARSE_JSON}, a {@code
NaN} or infinite {@code
+ * FLOAT} or {@code DOUBLE} is accepted, since a {@code VARIANT} is not
limited to what JSON can
+ * express. A timestamp keeps its declared precision, so a {@code
TIMESTAMP(p)} with {@code p} above
+ * 6 is stored with nanoseconds. That kind only covers 1677-09-21 to
2262-04-11, so such a cast
+ * fails for a value outside that range. Every other cast never fails.
+ */
+class PrimitiveToVariantCastRule extends
AbstractExpressionCodeGeneratorCastRule<Object, Variant> {
+
+ static final PrimitiveToVariantCastRule INSTANCE = new
PrimitiveToVariantCastRule();
+
+ /** The highest precision a VARIANT stores with microseconds, above it
nanoseconds are used. */
+ private static final int MAX_MICROS_PRECISION = 6;
Review Comment:
nit: `MAX_MICROS_PRECISION` duplicates
`VariantCastUtils.TIMESTAMP_PRECISION`. Could we expose one and use it in both
places?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/PrimitiveToVariantCastRule.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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;
+
+/**
+ * Primitive type to {@link LogicalTypeRoot#VARIANT} cast rule.
+ *
+ * <p>The value keeps the kind of its SQL type, so a {@code BIGINT} is stored
as a {@code BIGINT}
+ * even when it would fit a smaller one. Unlike {@code PARSE_JSON}, a {@code
NaN} or infinite {@code
+ * FLOAT} or {@code DOUBLE} is accepted, since a {@code VARIANT} is not
limited to what JSON can
+ * express. A timestamp keeps its declared precision, so a {@code
TIMESTAMP(p)} with {@code p} above
+ * 6 is stored with nanoseconds. That kind only covers 1677-09-21 to
2262-04-11, so such a cast
+ * fails for a value outside that range. Every other cast never fails.
+ */
+class PrimitiveToVariantCastRule extends
AbstractExpressionCodeGeneratorCastRule<Object, Variant> {
+
+ static final PrimitiveToVariantCastRule INSTANCE = new
PrimitiveToVariantCastRule();
+
+ /** The highest precision a VARIANT stores with microseconds, above it
nanoseconds are used. */
+ private static final int MAX_MICROS_PRECISION = 6;
+
+ 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;
+ }
+ }
+
+ @Override
+ public boolean canFail(LogicalType inputLogicalType, LogicalType
targetLogicalType) {
Review Comment:
`canFail` only covers timestamps with p > 6, but
`BinaryVariantInternalBuilder.checkCapacity` throws
`VARIANT_SIZE_LIMIT_EXCEPTION` once the value passes `SIZE_LIMIT` (16 MiB).
`appendString` and `appendBinary` both hit it. So `TRY_CAST(big_string AS
VARIANT)` throws instead of returning `NULL`, since `ScalarOperatorGens` only
wraps the cast when `canFail` is true. `CAST` fails with a bare
`VARIANT_SIZE_LIMIT` message.
Could we return true for CHAR/VARCHAR/BINARY/VARBINARY whose declared length
can exceed the limit? Then a `VARCHAR(100)` stays non-failing. The size error
should also become a `TableRuntimeException` with a proper message, like the
timestamp one. A test with a >16 MiB string for both `CAST` and `TRY_CAST`
would pin it. The Javadoc ("Every other cast never fails") and the commit
message need the same adjustment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
