twalthr commented on code in PR #28758:
URL: https://github.com/apache/flink/pull/28758#discussion_r3684320053
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java:
##########
@@ -1544,35 +1544,55 @@ Stream<CastTestSpecBuilder> testCases() {
.fromCase(BITMAP(), DEFAULT_BITMAP,
DEFAULT_BITMAP.toBytes())
.fromCase(BITMAP(), Bitmap.empty(),
Bitmap.empty().toBytes())
.fromCase(BITMAP(), null, null),
- // From VARIANT to primitive types. Numeric targets are
lenient: a variant holding
- // any numeric kind converts to the requested numeric type
(widening and narrowing).
- // Non-numeric targets are strict: the stored kind must match,
otherwise the cast
- // fails and TRY_CAST returns null.
+ // From VARIANT to primitive types. A cast succeeds only when
the target holds the
+ // stored value unaltered: an integer widens or narrows while
it stays in range, a
+ // DECIMAL has to fit the precision and scale, and a timestamp
the precision. FLOAT
+ // and DOUBLE are approximate, so they take any numeric kind
and reject only a
+ // magnitude out of range. Reading one kind as another is
never implicit.
CastTestSpecBuilder.testCastTo(BOOLEAN())
.fromCase(VARIANT(), Variant.newBuilder().of(true),
true)
.fromCase(VARIANT(), Variant.newBuilder().of(false),
false)
.fail(VARIANT(), Variant.newBuilder().of(1),
TableRuntimeException.class),
CastTestSpecBuilder.testCastTo(TINYINT())
.fromCase(VARIANT(), Variant.newBuilder().of((byte)
42), (byte) 42)
+ // a wider integer kind narrows while the value is in
range
.fromCase(VARIANT(), Variant.newBuilder().of(42),
(byte) 42)
+ // out of range is rejected instead of wrapping
+ .fail(VARIANT(), Variant.newBuilder().of(1000),
TableRuntimeException.class)
.fail(VARIANT(), Variant.newBuilder().of("x"),
TableRuntimeException.class),
CastTestSpecBuilder.testCastTo(SMALLINT())
.fromCase(VARIANT(), Variant.newBuilder().of((short)
42), (short) 42)
.fromCase(VARIANT(), Variant.newBuilder().of((byte)
42), (short) 42)
+ .fromCase(VARIANT(), Variant.newBuilder().of(1000),
(short) 1000)
+ .fail(
+ VARIANT(),
+ Variant.newBuilder().of(40000),
+ TableRuntimeException.class)
.fail(
VARIANT(),
Variant.newBuilder().of(true),
TableRuntimeException.class),
CastTestSpecBuilder.testCastTo(INT())
- // widening: a JSON integer is stored in the smallest
type but still casts
- // up
+ .fromCase(VARIANT(), Variant.newBuilder().of(42), 42)
+ // every integer kind converts as long as the value
fits
.fromCase(VARIANT(), Variant.newBuilder().of((byte)
42), 42)
.fromCase(VARIANT(), Variant.newBuilder().of((short)
42), 42)
- .fromCase(VARIANT(), Variant.newBuilder().of(42), 42)
.fromCase(VARIANT(), Variant.newBuilder().of(42L), 42)
- // narrowing from a floating point or decimal value
truncates
- .fromCase(VARIANT(), Variant.newBuilder().of(3.9d), 3)
- .fromCase(VARIANT(), Variant.newBuilder().of(new
BigDecimal("7.2")), 7)
+ .fail(
+ VARIANT(),
+ Variant.newBuilder().of(2147483648L),
+ TableRuntimeException.class)
+ // an approximate or decimal kind is not read as an
integer, whether or not
+ // the value happens to be integral
+ .fail(VARIANT(), Variant.newBuilder().of(7.0d),
TableRuntimeException.class)
+ .fail(
+ VARIANT(),
+ Variant.newBuilder().of(new BigDecimal("7.0")),
+ TableRuntimeException.class)
Review Comment:
This one we could rediscuss, right?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java:
##########
@@ -138,75 +138,203 @@ Stream<TestSetSpec> getTestSetSpecs() {
}
private static List<TestSetSpec> variantCasts() {
- // A variant is produced with PARSE_JSON so that the source column
stays a STRING literal;
- // there is no VARIANT literal to feed a source column directly. A
JSON integer is stored in
- // the smallest integer type that fits (42 -> TINYINT), and numeric
casts are lenient, so it
- // still widens to INT.
+ // A variant is produced with PARSE_JSON since there is no VARIANT
literal. Numeric casts
+ // succeed only when the value is preserved exactly, otherwise CAST
fails and TRY_CAST
+ // returns NULL.
return List.of(
TestSetSpec.forExpression("Cast a VARIANT produced by
PARSE_JSON to a primitive")
.onFieldsWithData("unused")
.andDataTypes(STRING())
+ // An integer converts to any integer target while the
value stays in
+ // range, and to FLOAT or DOUBLE which are approximate
by definition.
+ .testResult(
+ call("PARSE_JSON", "42").cast(TINYINT()),
+ "CAST(PARSE_JSON('42') AS TINYINT)",
+ (byte) 42,
+ TINYINT().notNull())
+ .testResult(
+ call("PARSE_JSON", "42").cast(SMALLINT()),
+ "CAST(PARSE_JSON('42') AS SMALLINT)",
+ (short) 42,
+ SMALLINT().notNull())
.testResult(
call("PARSE_JSON", "42").cast(INT()),
"CAST(PARSE_JSON('42') AS INT)",
42,
INT().notNull())
- // Integer overflow wraps around (Java narrowing),
like a regular numeric
- // cast.
.testResult(
- call("PARSE_JSON", "40000").cast(SMALLINT()),
- "CAST(PARSE_JSON('40000') AS SMALLINT)",
- (short) -25536,
+ call("PARSE_JSON", "42").cast(BIGINT()),
+ "CAST(PARSE_JSON('42') AS BIGINT)",
+ 42L,
+ BIGINT().notNull())
+ .testResult(
+ call("PARSE_JSON", "42").cast(FLOAT()),
+ "CAST(PARSE_JSON('42') AS FLOAT)",
+ 42.0f,
+ FLOAT().notNull())
+ .testResult(
+ call("PARSE_JSON", "42").cast(DOUBLE()),
+ "CAST(PARSE_JSON('42') AS DOUBLE)",
+ 42.0d,
+ DOUBLE().notNull())
+ // An out-of-range value is rejected rather than
wrapped.
+ .testResult(
+ call("PARSE_JSON", "1000").cast(SMALLINT()),
+ "CAST(PARSE_JSON('1000') AS SMALLINT)",
+ (short) 1000,
SMALLINT().notNull())
+ .testTableApiRuntimeError(
+ call("PARSE_JSON", "1000").cast(TINYINT()),
"overflowed")
+ .testSqlRuntimeError("CAST(PARSE_JSON('1000') AS
TINYINT)", "overflowed")
.testResult(
- call("PARSE_JSON", "128").cast(TINYINT()),
- "CAST(PARSE_JSON('128') AS TINYINT)",
- (byte) -128,
- TINYINT().notNull())
+ call("PARSE_JSON", "1000").tryCast(TINYINT()),
+ "TRY_CAST(PARSE_JSON('1000') AS TINYINT)",
+ null,
+ TINYINT())
+ // A decimal is not read as an integer, since that
would drop digits.
+ .testTableApiRuntimeError(
+ call("PARSE_JSON", "123.456").cast(INT()),
"Cannot cast a VARIANT")
.testResult(
- call("PARSE_JSON", "2147483648").cast(INT()),
- "CAST(PARSE_JSON('2147483648') AS INT)",
- -2147483648,
- INT().notNull())
+ call("PARSE_JSON", "123.456").tryCast(INT()),
+ "TRY_CAST(PARSE_JSON('123.456') AS INT)",
+ null,
+ INT())
+ // A DECIMAL target has to hold the value exactly.
.testResult(
- call("PARSE_JSON",
"9223372036854775808").cast(BIGINT()),
- "CAST(PARSE_JSON('9223372036854775808') AS
BIGINT)",
- -9223372036854775808L,
- BIGINT().notNull())
- // An out-of-range floating point value saturates when
cast to an integer,
- // and a fractional value is truncated toward zero.
+ call("PARSE_JSON", "123.456").cast(DECIMAL(6,
3)),
+ "CAST(PARSE_JSON('123.456') AS DECIMAL(6, 3))",
+ new BigDecimal("123.456"),
+ DECIMAL(6, 3).notNull())
+ .testTableApiRuntimeError(
+ call("PARSE_JSON", "123.456").cast(DECIMAL(6,
2)), "lose precision")
.testResult(
- call("PARSE_JSON", "1e20").cast(INT()),
- "CAST(PARSE_JSON('1e20') AS INT)",
- 2147483647,
- INT().notNull())
+ call("PARSE_JSON",
"123.456").tryCast(DECIMAL(6, 2)),
+ "TRY_CAST(PARSE_JSON('123.456') AS DECIMAL(6,
2))",
+ null,
+ DECIMAL(6, 2))
+ .testTableApiRuntimeError(
+ call("PARSE_JSON", "123.456").cast(DECIMAL(5,
3)), "overflowed")
.testResult(
- call("PARSE_JSON", "3.9").cast(INT()),
- "CAST(PARSE_JSON('3.9') AS INT)",
- 3,
- INT().notNull())
- // A value beyond the FLOAT range becomes infinity.
+ call("PARSE_JSON",
"123.456").tryCast(DECIMAL(5, 3)),
+ "TRY_CAST(PARSE_JSON('123.456') AS DECIMAL(5,
3))",
+ null,
+ DECIMAL(5, 3))
+ // An integer is exact, so it reaches a DECIMAL that
has room for it.
+ .testResult(
+ call("PARSE_JSON", "42").cast(DECIMAL(5, 2)),
+ "CAST(PARSE_JSON('42') AS DECIMAL(5, 2))",
+ new BigDecimal("42.00"),
+ DECIMAL(5, 2).notNull())
+ // A decimal reaches an approximate target, where
losing digits is expected.
.testResult(
- call("PARSE_JSON", "1e40").cast(FLOAT()),
- "CAST(PARSE_JSON('1e40') AS FLOAT)",
- Float.POSITIVE_INFINITY,
+ call("PARSE_JSON", "123.456").cast(FLOAT()),
+ "CAST(PARSE_JSON('123.456') AS FLOAT)",
+ 123.456f,
FLOAT().notNull())
- // DECIMAL overflow yields NULL instead of wrapping;
TRY_CAST surfaces it.
.testResult(
- call("PARSE_JSON",
"123.456").tryCast(DECIMAL(4, 2)),
- "TRY_CAST(PARSE_JSON('123.456') AS DECIMAL(4,
2))",
+ call("PARSE_JSON", "123.456").cast(DOUBLE()),
+ "CAST(PARSE_JSON('123.456') AS DOUBLE)",
+ 123.456d,
+ DOUBLE().notNull())
+ // A magnitude the target cannot represent is still
rejected.
+ .testTableApiRuntimeError(
+ call("PARSE_JSON", "1e40").cast(FLOAT()),
"overflowed")
+ .testResult(
+ call("PARSE_JSON", "1e40").tryCast(FLOAT()),
+ "TRY_CAST(PARSE_JSON('1e40') AS FLOAT)",
null,
- DECIMAL(4, 2))
+ FLOAT())
.testResult(
- call("PARSE_JSON", "42").cast(BIGINT()),
- "CAST(PARSE_JSON('42') AS BIGINT)",
- 42L,
- BIGINT().notNull())
+ call("PARSE_JSON", "1e20").cast(DOUBLE()),
+ "CAST(PARSE_JSON('1e20') AS DOUBLE)",
+ 1e20,
+ DOUBLE().notNull())
.testResult(
call("PARSE_JSON", "true").cast(BOOLEAN()),
"CAST(PARSE_JSON('true') AS BOOLEAN)",
true,
BOOLEAN().notNull())
+ // CAST returns the raw scalar value (string unquoted)
+ .testResult(
+ call("PARSE_JSON", "\"foo\"").cast(STRING()),
+ "CAST(PARSE_JSON('\"foo\"') AS STRING)",
+ "foo",
+ STRING().notNull())
+ .testResult(
+ call("PARSE_JSON", "123.456").cast(STRING()),
+ "CAST(PARSE_JSON('123.456') AS STRING)",
+ "123.456",
+ STRING().notNull())
+ .testResult(
+ call("PARSE_JSON", "true").cast(STRING()),
+ "CAST(PARSE_JSON('true') AS STRING)",
+ "true",
+ STRING().notNull())
Review Comment:
This we did not discuss. Let's not allow this for now. Only string can be
cast to string.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java:
##########
@@ -0,0 +1,257 @@
+/*
+ * 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.runtime.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.types.variant.Variant;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.time.Instant;
+import java.time.LocalDateTime;
+
+/**
+ * Runtime helpers for casting a {@code VARIANT} value to a SQL type.
+ *
+ * <p>A cast succeeds only when the target holds the stored value without
altering it, so a value is
+ * never wrapped, rounded, truncated, or padded to make it fit. {@code FLOAT}
and {@code DOUBLE} are
+ * the exception: they are approximate by definition, so they accept any
numeric kind and reject
+ * only a magnitude they cannot represent at all.
+ */
+@Internal
+public final class VariantCastUtils {
+
+ private VariantCastUtils() {}
+
+ /**
+ * Reads an integer variant as a {@code long} and checks it against the
target range. Only the
+ * integer kinds are accepted, so an approximate or decimal value is
rejected rather than
+ * rounded.
+ */
+ public static long toIntegral(Variant variant, long min, long max, String
targetType) {
+ switch (variant.getType()) {
+ case TINYINT:
+ case SMALLINT:
+ case INT:
+ case BIGINT:
+ break;
+ default:
+ throw unsupportedKind(variant, targetType);
+ }
+ final long value = ((Number) variant.get()).longValue();
+ if (value < min || value > max) {
+ throw overflow(value, targetType);
+ }
+ return value;
+ }
+
+ /**
+ * Reads any numeric variant as a {@code float}. Dropping decimal digits
is expected of an
+ * approximate type, but a magnitude outside the {@code FLOAT} range is
rejected.
+ */
+ public static float toFloat(Variant variant) {
+ final float value = numeric(variant, "FLOAT").floatValue();
+ if (!Float.isFinite(value)) {
+ throw overflow(variant.get(), "FLOAT");
+ }
+ return value;
+ }
+
+ /** Reads any numeric variant as a {@code double}. See {@link
#toFloat(Variant)}. */
+ public static double toDouble(Variant variant) {
+ final double value = numeric(variant, "DOUBLE").doubleValue();
+ if (!Double.isFinite(value)) {
+ throw overflow(variant.get(), "DOUBLE");
+ }
+ return value;
+ }
+
+ /**
+ * Reads an integer or decimal variant as the target {@code DECIMAL}. The
value has to fit the
+ * precision and scale without rounding, although trailing zeros may be
appended to reach the
+ * scale.
+ */
+ public static DecimalData toDecimal(Variant variant, int precision, int
scale) {
+ final String targetType = String.format("DECIMAL(%d, %d)", precision,
scale);
Review Comment:
only format on error, keep the hot path performant
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java:
##########
@@ -0,0 +1,257 @@
+/*
+ * 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.runtime.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.types.variant.Variant;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.time.Instant;
+import java.time.LocalDateTime;
+
+/**
+ * Runtime helpers for casting a {@code VARIANT} value to a SQL type.
+ *
+ * <p>A cast succeeds only when the target holds the stored value without
altering it, so a value is
+ * never wrapped, rounded, truncated, or padded to make it fit. {@code FLOAT}
and {@code DOUBLE} are
+ * the exception: they are approximate by definition, so they accept any
numeric kind and reject
+ * only a magnitude they cannot represent at all.
+ */
+@Internal
+public final class VariantCastUtils {
+
+ private VariantCastUtils() {}
+
+ /**
+ * Reads an integer variant as a {@code long} and checks it against the
target range. Only the
+ * integer kinds are accepted, so an approximate or decimal value is
rejected rather than
+ * rounded.
+ */
+ public static long toIntegral(Variant variant, long min, long max, String
targetType) {
+ switch (variant.getType()) {
+ case TINYINT:
+ case SMALLINT:
+ case INT:
+ case BIGINT:
+ break;
+ default:
+ throw unsupportedKind(variant, targetType);
+ }
+ final long value = ((Number) variant.get()).longValue();
+ if (value < min || value > max) {
+ throw overflow(value, targetType);
+ }
+ return value;
+ }
+
+ /**
+ * Reads any numeric variant as a {@code float}. Dropping decimal digits
is expected of an
+ * approximate type, but a magnitude outside the {@code FLOAT} range is
rejected.
+ */
+ public static float toFloat(Variant variant) {
+ final float value = numeric(variant, "FLOAT").floatValue();
+ if (!Float.isFinite(value)) {
+ throw overflow(variant.get(), "FLOAT");
+ }
+ return value;
+ }
+
+ /** Reads any numeric variant as a {@code double}. See {@link
#toFloat(Variant)}. */
+ public static double toDouble(Variant variant) {
+ final double value = numeric(variant, "DOUBLE").doubleValue();
+ if (!Double.isFinite(value)) {
+ throw overflow(variant.get(), "DOUBLE");
+ }
+ return value;
+ }
+
+ /**
+ * Reads an integer or decimal variant as the target {@code DECIMAL}. The
value has to fit the
+ * precision and scale without rounding, although trailing zeros may be
appended to reach the
+ * scale.
+ */
+ public static DecimalData toDecimal(Variant variant, int precision, int
scale) {
+ final String targetType = String.format("DECIMAL(%d, %d)", precision,
scale);
+ final BigDecimal value;
+ switch (variant.getType()) {
+ case TINYINT:
+ case SMALLINT:
+ case INT:
+ case BIGINT:
+ value = BigDecimal.valueOf(((Number)
variant.get()).longValue());
+ break;
+ case DECIMAL:
+ value = variant.getDecimal();
+ break;
+ default:
+ throw unsupportedKind(variant, targetType);
+ }
+ // The integral part must fit the digits the target reserves for it.
+ if (value.precision() - value.scale() > precision - scale) {
+ throw overflow(value, targetType);
+ }
+ final BigDecimal rescaled;
+ try {
+ // UNNECESSARY throws unless the value fits the target scale
exactly.
+ rescaled = value.setScale(scale, RoundingMode.UNNECESSARY);
+ } catch (ArithmeticException e) {
+ throw lossyCast(value, targetType);
+ }
+ final DecimalData decimal = DecimalData.fromBigDecimal(rescaled,
precision, scale);
+ if (decimal == null) {
+ throw overflow(value, targetType);
+ }
+ return decimal;
+ }
+
+ /**
+ * Reads a timestamp variant as the target {@code TIMESTAMP}. A variant
keeps microseconds, so
+ * the value is accepted only when its fractional seconds fit the target
precision.
+ */
+ public static TimestampData toTimestamp(Variant variant, int precision) {
+ if (variant.getType() != Variant.Type.TIMESTAMP) {
+ throw unsupportedKind(variant, String.format("TIMESTAMP(%d)",
precision));
+ }
+ final LocalDateTime value = variant.getDateTime();
+ checkFractionFits(value.getNano(), precision, value, "TIMESTAMP");
+ return TimestampData.fromLocalDateTime(value);
+ }
+
+ /** Reads a timestamp with local time zone variant. See {@link
#toTimestamp(Variant, int)}. */
+ public static TimestampData toTimestampLtz(Variant variant, int precision)
{
+ if (variant.getType() != Variant.Type.TIMESTAMP_LTZ) {
+ throw unsupportedKind(variant, String.format("TIMESTAMP_LTZ(%d)",
precision));
+ }
+ final Instant value = variant.getInstant();
+ checkFractionFits(value.getNano(), precision, value, "TIMESTAMP_LTZ");
+ return TimestampData.fromInstant(value);
+ }
+
+ /**
+ * Reads a binary variant, enforcing {@code targetLength} strictly with no
padding or truncation
+ * ({@code BINARY} requires an exact length, {@code VARBINARY} an upper
bound).
+ */
+ public static byte[] toBytes(Variant variant, int targetLength, boolean
fixedLength) {
+ final byte[] value = variant.getBytes();
+ final boolean fits =
+ fixedLength ? value.length == targetLength : value.length <=
targetLength;
+ if (!fits) {
+ throw new TableRuntimeException(
+ String.format(
+ "The VARIANT binary value of length %d does not
fit %s(%d); VARIANT "
+ + "casts do not pad or truncate.",
+ value.length, fixedLength ? "BINARY" :
"VARBINARY", targetLength));
+ }
+ return value;
+ }
+
+ /**
+ * Casts a scalar {@code VARIANT} to its raw string value, enforcing
{@code targetLength}
+ * strictly with no padding or truncation ({@code CHAR} requires an exact
length, {@code
+ * VARCHAR} an upper bound).
+ */
+ public static String toStringValue(Variant variant, int targetLength,
boolean charTarget) {
+ final String targetType =
+ String.format("%s(%d)", charTarget ? "CHAR" : "VARCHAR",
targetLength);
+ switch (variant.getType()) {
+ case OBJECT:
+ case ARRAY:
+ case BYTES:
+ throw new TableRuntimeException(
+ String.format(
+ "Cannot cast a VARIANT %s value to a character
string. Use the "
+ + "JSON_STRING function to obtain its
JSON representation.",
+ variant.getType()));
+ case NULL:
+ // Only reachable for a NOT NULL target. A nullable target
maps a null-valued
+ // variant to SQL NULL before this method is called.
+ throw new TableRuntimeException(
+ String.format(
+ "Cannot cast a VARIANT null value to %s
because the target does not "
+ + "accept NULL.",
+ targetType));
+ default:
+ // Scalars are cast to their raw string value.
+ }
+ final String value = variant.get().toString();
Review Comment:
this is again JSON and not SQL. e.g. `TRUE` vs `true` for variant boolean to
string. as mentioned above. only string variant can cast to string. if you want
to support more, come up with a proper string representation for each variant
type. incl TS_LTZ in users time zone.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java:
##########
@@ -0,0 +1,257 @@
+/*
+ * 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.runtime.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.types.variant.Variant;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.time.Instant;
+import java.time.LocalDateTime;
+
+/**
+ * Runtime helpers for casting a {@code VARIANT} value to a SQL type.
+ *
+ * <p>A cast succeeds only when the target holds the stored value without
altering it, so a value is
+ * never wrapped, rounded, truncated, or padded to make it fit. {@code FLOAT}
and {@code DOUBLE} are
+ * the exception: they are approximate by definition, so they accept any
numeric kind and reject
+ * only a magnitude they cannot represent at all.
+ */
+@Internal
+public final class VariantCastUtils {
+
+ private VariantCastUtils() {}
+
+ /**
+ * Reads an integer variant as a {@code long} and checks it against the
target range. Only the
+ * integer kinds are accepted, so an approximate or decimal value is
rejected rather than
+ * rounded.
+ */
+ public static long toIntegral(Variant variant, long min, long max, String
targetType) {
+ switch (variant.getType()) {
+ case TINYINT:
+ case SMALLINT:
+ case INT:
+ case BIGINT:
+ break;
+ default:
+ throw unsupportedKind(variant, targetType);
+ }
+ final long value = ((Number) variant.get()).longValue();
+ if (value < min || value > max) {
+ throw overflow(value, targetType);
+ }
+ return value;
+ }
+
+ /**
+ * Reads any numeric variant as a {@code float}. Dropping decimal digits
is expected of an
+ * approximate type, but a magnitude outside the {@code FLOAT} range is
rejected.
+ */
+ public static float toFloat(Variant variant) {
+ final float value = numeric(variant, "FLOAT").floatValue();
+ if (!Float.isFinite(value)) {
+ throw overflow(variant.get(), "FLOAT");
+ }
+ return value;
+ }
+
+ /** Reads any numeric variant as a {@code double}. See {@link
#toFloat(Variant)}. */
+ public static double toDouble(Variant variant) {
+ final double value = numeric(variant, "DOUBLE").doubleValue();
+ if (!Double.isFinite(value)) {
+ throw overflow(variant.get(), "DOUBLE");
+ }
+ return value;
+ }
+
+ /**
+ * Reads an integer or decimal variant as the target {@code DECIMAL}. The
value has to fit the
+ * precision and scale without rounding, although trailing zeros may be
appended to reach the
+ * scale.
+ */
+ public static DecimalData toDecimal(Variant variant, int precision, int
scale) {
+ final String targetType = String.format("DECIMAL(%d, %d)", precision,
scale);
+ final BigDecimal value;
+ switch (variant.getType()) {
+ case TINYINT:
+ case SMALLINT:
+ case INT:
+ case BIGINT:
+ value = BigDecimal.valueOf(((Number)
variant.get()).longValue());
+ break;
+ case DECIMAL:
+ value = variant.getDecimal();
+ break;
+ default:
+ throw unsupportedKind(variant, targetType);
+ }
+ // The integral part must fit the digits the target reserves for it.
+ if (value.precision() - value.scale() > precision - scale) {
+ throw overflow(value, targetType);
+ }
+ final BigDecimal rescaled;
+ try {
+ // UNNECESSARY throws unless the value fits the target scale
exactly.
+ rescaled = value.setScale(scale, RoundingMode.UNNECESSARY);
+ } catch (ArithmeticException e) {
+ throw lossyCast(value, targetType);
+ }
+ final DecimalData decimal = DecimalData.fromBigDecimal(rescaled,
precision, scale);
+ if (decimal == null) {
+ throw overflow(value, targetType);
+ }
+ return decimal;
+ }
+
+ /**
+ * Reads a timestamp variant as the target {@code TIMESTAMP}. A variant
keeps microseconds, so
+ * the value is accepted only when its fractional seconds fit the target
precision.
+ */
+ public static TimestampData toTimestamp(Variant variant, int precision) {
+ if (variant.getType() != Variant.Type.TIMESTAMP) {
+ throw unsupportedKind(variant, String.format("TIMESTAMP(%d)",
precision));
+ }
+ final LocalDateTime value = variant.getDateTime();
+ checkFractionFits(value.getNano(), precision, value, "TIMESTAMP");
+ return TimestampData.fromLocalDateTime(value);
+ }
+
+ /** Reads a timestamp with local time zone variant. See {@link
#toTimestamp(Variant, int)}. */
+ public static TimestampData toTimestampLtz(Variant variant, int precision)
{
+ if (variant.getType() != Variant.Type.TIMESTAMP_LTZ) {
+ throw unsupportedKind(variant, String.format("TIMESTAMP_LTZ(%d)",
precision));
+ }
+ final Instant value = variant.getInstant();
+ checkFractionFits(value.getNano(), precision, value, "TIMESTAMP_LTZ");
+ return TimestampData.fromInstant(value);
+ }
+
+ /**
+ * Reads a binary variant, enforcing {@code targetLength} strictly with no
padding or truncation
+ * ({@code BINARY} requires an exact length, {@code VARBINARY} an upper
bound).
+ */
+ public static byte[] toBytes(Variant variant, int targetLength, boolean
fixedLength) {
+ final byte[] value = variant.getBytes();
+ final boolean fits =
+ fixedLength ? value.length == targetLength : value.length <=
targetLength;
+ if (!fits) {
+ throw new TableRuntimeException(
+ String.format(
+ "The VARIANT binary value of length %d does not
fit %s(%d); VARIANT "
+ + "casts do not pad or truncate.",
+ value.length, fixedLength ? "BINARY" :
"VARBINARY", targetLength));
+ }
+ return value;
+ }
+
+ /**
+ * Casts a scalar {@code VARIANT} to its raw string value, enforcing
{@code targetLength}
+ * strictly with no padding or truncation ({@code CHAR} requires an exact
length, {@code
+ * VARCHAR} an upper bound).
+ */
+ public static String toStringValue(Variant variant, int targetLength,
boolean charTarget) {
+ final String targetType =
Review Comment:
same comment as above
--
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]
