raminqaf commented on code in PR #28758: URL: https://github.com/apache/flink/pull/28758#discussion_r3690105826
########## 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: Done -- 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]
