snuyanzin commented on code in PR #28758: URL: https://github.com/apache/flink/pull/28758#discussion_r3718619712
########## flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java: ########## @@ -0,0 +1,369 @@ +/* + * 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.data.binary.StringUtf8Utils; +import org.apache.flink.table.utils.DateTimeUtils; +import org.apache.flink.types.variant.Variant; + +import java.math.BigDecimal; +import java.math.RoundingMode; +import java.nio.charset.StandardCharsets; +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). + * + * <p>A stored binary value has to be well-formed UTF-8, since a character string cannot carry + * bytes that no character maps to. Invalid input is rejected rather than decoded into {@code + * U+FFFD}, which would silently substitute a character the value never held. + * + * @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: + value = variant.get().toString(); + break; + case DECIMAL: + // toPlainString rather than toString, so that a small scale is not rendered in + // scientific notation, matching a regular DECIMAL to string cast. + value = variant.getDecimal().toPlainString(); + break; + case STRING: + value = variant.getString(); + break; + case BYTES: + // SQL reads a binary value as UTF-8, the same as a regular BINARY to string cast. + final byte[] utf8 = variant.getBytes(); + final int invalidAt = + StringUtf8Utils.firstInvalidUtf8ByteIndex(utf8, 0, utf8.length); + if (invalidAt >= 0) { + throw new TableRuntimeException( + String.format( + "Cannot cast the VARIANT binary value to %s because it is not " + + "valid UTF-8; the first invalid byte is at index %d " + + "of %d. Cast to BYTES to inspect the raw value, or " + + "wrap that in MAKE_VALID_UTF8 to replace every " + + "invalid byte with the U+FFFD replacement character.", + characterTarget(targetLength, charTarget), + invalidAt, + utf8.length)); + } + value = new String(utf8, StandardCharsets.UTF_8); + break; + case DATE: + value = DateTimeUtils.formatDate((int) variant.getDate().toEpochDay()); + break; + case TIMESTAMP: + // A wall-clock value needs no zone shift, which is what UTC_ZONE achieves here. A + // variant keeps microseconds, so the precision is always 6. + value = + DateTimeUtils.formatTimestamp( + TimestampData.fromLocalDateTime(variant.getDateTime()), + DateTimeUtils.UTC_ZONE, + TIMESTAMP_PRECISION); + break; + case TIMESTAMP_LTZ: + value = + DateTimeUtils.formatTimestamp( + TimestampData.fromInstant(variant.getInstant()), + sessionZone, + TIMESTAMP_PRECISION); + break; + 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.", + characterTarget(targetLength, charTarget))); + default: + // An object or array has no scalar rendering. + 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())); + } + final boolean fits = + charTarget ? value.length() == targetLength : value.length() <= targetLength; Review Comment: this might be a problem for multibyte characters eg. such query ```sql SELECT CAST('𝕏' AS CHAR(1)), TRY_CAST(PARSE_JSON('"𝕏"') AS CHAR(1)); ``` returns `𝕏` NULL I would expect same result for 2 projections -- 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]
