raminqaf commented on code in PR #28758: URL: https://github.com/apache/flink/pull/28758#discussion_r3704703986
########## flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java: ########## @@ -0,0 +1,340 @@ +/* + * 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.table.utils.DateTimeUtils; +import org.apache.flink.types.variant.Variant; + +import java.math.BigDecimal; +import java.math.RoundingMode; +import java.time.Instant; +import java.time.LocalDateTime; +import java.util.TimeZone; + +/** + * 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. Any numeric kind therefore reaches + * an integer target as long as the value is integral and in range. {@code FLOAT} and {@code DOUBLE} + * are the exception to exactness: 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 { + + /** + * The magnitude 2^63, the exclusive bound for a {@code double} that still fits a {@code long}. + * Taken from {@link Long#MIN_VALUE} because that is exactly -2^63, whereas widening {@link + * Long#MAX_VALUE} would reach the same number only by rounding up. + */ + private static final double LONG_MAGNITUDE_LIMIT = -(double) Long.MIN_VALUE; + + /** A variant stores a timestamp with microsecond precision. */ + private static final int TIMESTAMP_PRECISION = 6; + + private VariantCastUtils() {} + + /** + * Reads a numeric variant as a {@code long} and checks it against the target range. An + * approximate or decimal value is accepted only when it is already integral, so nothing is + * rounded away. + */ + public static long toIntegral(Variant variant, long min, long max, String targetType) { + final long value; + switch (variant.getType()) { + case TINYINT: + case SMALLINT: + case INT: + case BIGINT: + value = ((Number) variant.get()).longValue(); + break; + case FLOAT: + case DOUBLE: + final double approximate = ((Number) variant.get()).doubleValue(); + // Below 2^63 the narrowing conversion stays exact. The comparison is negated so + // that NaN fails it too. + if (!(Math.abs(approximate) < LONG_MAGNITUDE_LIMIT)) { + throw overflow(approximate, targetType); + } + value = (long) approximate; + if (value != approximate) { + throw lossyCast(approximate, targetType); + } + break; + case DECIMAL: + final BigDecimal decimal = variant.getDecimal(); + final BigDecimal integral; + try { + // UNNECESSARY throws unless the value is already integral. + integral = decimal.setScale(0, RoundingMode.UNNECESSARY); + } catch (ArithmeticException e) { + throw lossyCast(decimal, targetType); + } + try { + // longValueExact rejects a value that does not fit a long instead of returning + // its low-order bits. + value = integral.longValueExact(); + } catch (ArithmeticException e) { + throw overflow(decimal, targetType); + } + break; + default: + throw unsupportedKind(variant, targetType); + } + 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 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, decimalTarget(precision, scale)); + } + // The integral part must fit the digits the target reserves for it. + if (value.precision() - value.scale() > precision - scale) { + throw overflow(value, decimalTarget(precision, scale)); + } + 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, decimalTarget(precision, scale)); + } + final DecimalData decimal = DecimalData.fromBigDecimal(rescaled, precision, scale); + if (decimal == null) { + throw overflow(value, decimalTarget(precision, scale)); + } + return decimal; + } + + private static String decimalTarget(int precision, int scale) { + return String.format("DECIMAL(%d, %d)", precision, scale); + } + + /** + * 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 a character string, rendering the value the way a regular + * SQL cast of the stored kind would. {@code targetLength} is enforced strictly with no padding + * or truncation ({@code CHAR} requires an exact length, {@code VARCHAR} an upper bound). + * + * @param sessionZone the session time zone, applied to a {@code TIMESTAMP_LTZ} value + */ + public static String toStringValue( + Variant variant, TimeZone sessionZone, int targetLength, boolean charTarget) { + final String value; + switch (variant.getType()) { + case BOOLEAN: + value = variant.getBoolean() ? "TRUE" : "FALSE"; + break; + case TINYINT: + case SMALLINT: + case INT: + case BIGINT: + case FLOAT: + case DOUBLE: + case DECIMAL: + value = variant.get().toString(); + break; + case STRING: + value = variant.getString(); Review Comment: I implemented in a way that the user casts Bytes -> String and if it is broken it gets a proper error message and sees the bytes. Unfortunately, if we have a STRING and want to cast it to Bytes, this will fail. If we want to allow this for debugging reasons, which I think is very useful to just see the bytes. One way to solve this is to introduce a `getStringBytes` in `BinaryVariant` that fetches the Binary form of String. This will cause that the implementation diverges from Spark. Can be also done in a Follow-up PR. -- 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]
