This is an automated email from the ASF dual-hosted git repository. twalthr pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4891ee93bc0f0e3b1f8bf6190963ea5931b0dc2f Author: slinkydeveloper <[email protected]> AuthorDate: Mon Nov 8 17:37:18 2021 +0100 [FLINK-24779][table-planner] Add numeric casting rules Signed-off-by: slinkydeveloper <[email protected]> This closes #17738. --- .../functions/casting/CastRuleProvider.java | 10 +- .../casting/ExpressionCodeGeneratorCastRule.java | 3 +- .../AbstractExpressionCodeGeneratorCastRule.java | 4 +- .../AbstractNullAwareCodeGeneratorCastRule.java | 14 +- .../functions/casting/rules/CastRuleUtils.java | 66 +++++ ...CastRule.java => DecimalToDecimalCastRule.java} | 27 +- ...java => DecimalToNumericPrimitiveCastRule.java} | 35 ++- .../functions/casting/rules/IdentityCastRule.java | 18 ++ .../casting/rules/NumericPrimitiveCastRule.java | 79 ++++++ .../rules/NumericPrimitiveToDecimalCastRule.java | 75 ++++++ .../flink/table/planner/codegen/CodeGenUtils.scala | 111 ++------- .../table/planner/codegen/ExprCodeGenerator.scala | 4 +- .../planner/codegen/calls/BuiltInMethods.scala | 27 +- .../table/planner/codegen/calls/IfCallGen.scala | 32 ++- .../planner/codegen/calls/ScalarOperatorGens.scala | 113 +++------ .../planner/functions/casting/CastRulesTest.java | 272 ++++++++++++++++++++- .../apache/flink/table/data/DecimalDataUtils.java | 24 -- 17 files changed, 683 insertions(+), 231 deletions(-) diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java index 2684e79..a926630 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java @@ -25,15 +25,18 @@ import org.apache.flink.table.planner.functions.casting.rules.ArrayToStringCastR import org.apache.flink.table.planner.functions.casting.rules.BinaryToStringCastRule; import org.apache.flink.table.planner.functions.casting.rules.BooleanToStringCastRule; import org.apache.flink.table.planner.functions.casting.rules.DateToStringCastRule; +import org.apache.flink.table.planner.functions.casting.rules.DecimalToDecimalCastRule; +import org.apache.flink.table.planner.functions.casting.rules.DecimalToNumericPrimitiveCastRule; import org.apache.flink.table.planner.functions.casting.rules.IdentityCastRule; import org.apache.flink.table.planner.functions.casting.rules.IntervalToStringCastRule; import org.apache.flink.table.planner.functions.casting.rules.MapToStringCastRule; +import org.apache.flink.table.planner.functions.casting.rules.NumericPrimitiveCastRule; +import org.apache.flink.table.planner.functions.casting.rules.NumericPrimitiveToDecimalCastRule; import org.apache.flink.table.planner.functions.casting.rules.NumericToStringCastRule; import org.apache.flink.table.planner.functions.casting.rules.RawToStringCastRule; import org.apache.flink.table.planner.functions.casting.rules.RowToStringCastRule; import org.apache.flink.table.planner.functions.casting.rules.TimeToStringCastRule; import org.apache.flink.table.planner.functions.casting.rules.TimestampToStringCastRule; -import org.apache.flink.table.planner.functions.casting.rules.UpcastToBigIntCastRule; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.LogicalTypeFamily; import org.apache.flink.table.types.logical.LogicalTypeRoot; @@ -56,7 +59,10 @@ public class CastRuleProvider { static { INSTANCE // Numeric rules - .addRule(UpcastToBigIntCastRule.INSTANCE) + .addRule(DecimalToDecimalCastRule.INSTANCE) + .addRule(NumericPrimitiveToDecimalCastRule.INSTANCE) + .addRule(DecimalToNumericPrimitiveCastRule.INSTANCE) + .addRule(NumericPrimitiveCastRule.INSTANCE) // To string rules .addRule(NumericToStringCastRule.INSTANCE) .addRule(BooleanToStringCastRule.INSTANCE) diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/ExpressionCodeGeneratorCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/ExpressionCodeGeneratorCastRule.java index a41db96..68b2aba 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/ExpressionCodeGeneratorCastRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/ExpressionCodeGeneratorCastRule.java @@ -34,7 +34,8 @@ public interface ExpressionCodeGeneratorCastRule<IN, OUT> extends CodeGeneratorC * Generate a Java expression performing the casting. This expression can be wrapped in another * expression, or assigned to a variable or returned from a function. * - * <p>NOTE: the {@code inputTerm} is always either a primitive or a non-null object. + * <p>NOTE: the {@code inputTerm} is always either a primitive or a non-null object, while the + * expression result is either a primitive or a nullable object. */ String generateExpression( CodeGeneratorCastRule.Context context, diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/AbstractExpressionCodeGeneratorCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/AbstractExpressionCodeGeneratorCastRule.java index aba38f0..eaf5fc4 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/AbstractExpressionCodeGeneratorCastRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/AbstractExpressionCodeGeneratorCastRule.java @@ -30,8 +30,8 @@ import org.apache.flink.table.types.logical.utils.LogicalTypeUtils; import java.util.Collections; -import static org.apache.flink.table.planner.codegen.CodeGenUtils.box; -import static org.apache.flink.table.planner.codegen.CodeGenUtils.unbox; +import static org.apache.flink.table.planner.functions.casting.rules.CastRuleUtils.box; +import static org.apache.flink.table.planner.functions.casting.rules.CastRuleUtils.unbox; /** * Base class for cast rules that supports code generation, requiring only an expression to perform diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/AbstractNullAwareCodeGeneratorCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/AbstractNullAwareCodeGeneratorCastRule.java index c58d1cd..9826286 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/AbstractNullAwareCodeGeneratorCastRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/AbstractNullAwareCodeGeneratorCastRule.java @@ -24,6 +24,7 @@ import org.apache.flink.table.planner.functions.casting.CastRulePredicate; import org.apache.flink.table.planner.functions.casting.CodeGeneratorCastRule; import org.apache.flink.table.types.logical.LogicalType; +import static org.apache.flink.table.planner.codegen.CodeGenUtils.isPrimitiveNullable; import static org.apache.flink.table.planner.codegen.CodeGenUtils.primitiveDefaultValue; import static org.apache.flink.table.planner.codegen.CodeGenUtils.primitiveTypeTermForType; @@ -59,8 +60,7 @@ public abstract class AbstractNullAwareCodeGeneratorCastRule<IN, OUT> LogicalType targetType) { final CastRuleUtils.CodeWriter writer = new CastRuleUtils.CodeWriter(); - // Result of a casting can be null only and only if the input is null - final boolean isResultNullable = inputType.isNullable(); + final boolean isResultNullable = inputType.isNullable() || isPrimitiveNullable(targetType); String nullTerm; if (isResultNullable) { nullTerm = context.declareVariable("boolean", "isNull"); @@ -81,7 +81,15 @@ public abstract class AbstractNullAwareCodeGeneratorCastRule<IN, OUT> if (isResultNullable) { writer.ifStmt( "!" + nullTerm, - thenWriter -> thenWriter.appendBlock(castCodeBlock), + thenWriter -> { + thenWriter.appendBlock(castCodeBlock); + + // If the result type is not primitive, + // then perform another null check + if (isPrimitiveNullable(targetType)) { + thenWriter.assignStmt(nullTerm, returnTerm + " == null"); + } + }, elseWriter -> elseWriter.assignStmt(returnTerm, primitiveDefaultValue(targetType))); } else { diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/CastRuleUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/CastRuleUtils.java index c32d447..437884f 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/CastRuleUtils.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/CastRuleUtils.java @@ -21,6 +21,7 @@ package org.apache.flink.table.planner.functions.casting.rules; import org.apache.flink.table.planner.codegen.CodeGenUtils; import org.apache.flink.table.planner.functions.casting.CastCodeBlock; import org.apache.flink.table.planner.functions.casting.CastRule; +import org.apache.flink.table.types.logical.DistinctType; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.utils.EncodingUtils; @@ -41,6 +42,10 @@ final class CastRuleUtils { static final String NULL_STR_LITERAL = strLiteral("null"); static final String EMPTY_STR_LITERAL = "\"\""; + static String staticCall(Class<?> clazz, String methodName, Object... args) { + return methodCall(className(clazz), methodName, args); + } + static String staticCall(Method staticMethod, Object... args) { return functionCall(CodeGenUtils.qualifyMethod(staticMethod), args); } @@ -80,6 +85,67 @@ final class CastRuleUtils { return "\"" + EncodingUtils.escapeJava(str) + "\""; } + static String cast(String target, String expression) { + return "((" + target + ")(" + expression + "))"; + } + + static String castToPrimitive(LogicalType target, String expression) { + return cast(primitiveTypeTermForType(target), expression); + } + + static String unbox(String term, LogicalType type) { + switch (type.getTypeRoot()) { + case BOOLEAN: + return methodCall(term, "booleanValue"); + case TINYINT: + return methodCall(term, "byteValue"); + case SMALLINT: + return methodCall(term, "shortValue"); + case INTEGER: + case DATE: + case TIME_WITHOUT_TIME_ZONE: + case INTERVAL_YEAR_MONTH: + return methodCall(term, "intValue"); + case BIGINT: + case INTERVAL_DAY_TIME: + return methodCall(term, "longValue"); + case FLOAT: + return methodCall(term, "floatValue"); + case DOUBLE: + return methodCall(term, "doubleValue"); + case DISTINCT_TYPE: + return unbox(term, ((DistinctType) type).getSourceType()); + } + return term; + } + + static String box(String term, LogicalType type) { + switch (type.getTypeRoot()) { + // ordered by type root definition + case BOOLEAN: + return staticCall(Boolean.class, "valueOf", term); + case TINYINT: + return staticCall(Byte.class, "valueOf", term); + case SMALLINT: + return staticCall(Short.class, "valueOf", term); + case INTEGER: + case DATE: + case TIME_WITHOUT_TIME_ZONE: + case INTERVAL_YEAR_MONTH: + return staticCall(Integer.class, "valueOf", term); + case BIGINT: + case INTERVAL_DAY_TIME: + return staticCall(Long.class, "valueOf", term); + case FLOAT: + return staticCall(Float.class, "valueOf", term); + case DOUBLE: + return staticCall(Double.class, "valueOf", term); + case DISTINCT_TYPE: + box(term, ((DistinctType) type).getSourceType()); + } + return term; + } + static final class CodeWriter { StringBuilder builder = new StringBuilder(); diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/UpcastToBigIntCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/DecimalToDecimalCastRule.java similarity index 60% copy from flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/UpcastToBigIntCastRule.java copy to flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/DecimalToDecimalCastRule.java index 4d002f3..11700fd 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/UpcastToBigIntCastRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/DecimalToDecimalCastRule.java @@ -19,24 +19,28 @@ package org.apache.flink.table.planner.functions.casting.rules; import org.apache.flink.annotation.Internal; +import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.planner.codegen.calls.BuiltInMethods; import org.apache.flink.table.planner.functions.casting.CastRulePredicate; import org.apache.flink.table.planner.functions.casting.CodeGeneratorCastRule; +import org.apache.flink.table.types.logical.DecimalType; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.LogicalTypeRoot; -/** Upcasting to long for smaller types. */ +import static org.apache.flink.table.planner.functions.casting.rules.CastRuleUtils.staticCall; + +/** {@link LogicalTypeRoot#DECIMAL} to {@link LogicalTypeRoot#DECIMAL} cast rule. */ @Internal -public class UpcastToBigIntCastRule extends AbstractExpressionCodeGeneratorCastRule<Object, Long> { +public class DecimalToDecimalCastRule + extends AbstractExpressionCodeGeneratorCastRule<DecimalData, DecimalData> { - public static final UpcastToBigIntCastRule INSTANCE = new UpcastToBigIntCastRule(); + public static final DecimalToDecimalCastRule INSTANCE = new DecimalToDecimalCastRule(); - private UpcastToBigIntCastRule() { + private DecimalToDecimalCastRule() { super( CastRulePredicate.builder() - .input(LogicalTypeRoot.TINYINT) - .input(LogicalTypeRoot.SMALLINT) - .input(LogicalTypeRoot.INTEGER) - .target(LogicalTypeRoot.BIGINT) + .input(LogicalTypeRoot.DECIMAL) + .target(LogicalTypeRoot.DECIMAL) .build()); } @@ -46,6 +50,11 @@ public class UpcastToBigIntCastRule extends AbstractExpressionCodeGeneratorCastR String inputTerm, LogicalType inputLogicalType, LogicalType targetLogicalType) { - return "((long)(" + inputTerm + "))"; + final DecimalType targetDecimalType = (DecimalType) targetLogicalType; + return staticCall( + BuiltInMethods.DECIMAL_TO_DECIMAL(), + inputTerm, + targetDecimalType.getPrecision(), + targetDecimalType.getScale()); } } diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/UpcastToBigIntCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/DecimalToNumericPrimitiveCastRule.java similarity index 51% rename from flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/UpcastToBigIntCastRule.java rename to flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/DecimalToNumericPrimitiveCastRule.java index 4d002f3..c9f4927 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/UpcastToBigIntCastRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/DecimalToNumericPrimitiveCastRule.java @@ -19,24 +19,37 @@ package org.apache.flink.table.planner.functions.casting.rules; import org.apache.flink.annotation.Internal; +import org.apache.flink.table.data.DecimalData; import org.apache.flink.table.planner.functions.casting.CastRulePredicate; import org.apache.flink.table.planner.functions.casting.CodeGeneratorCastRule; import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.LogicalTypeFamily; import org.apache.flink.table.types.logical.LogicalTypeRoot; -/** Upcasting to long for smaller types. */ +import java.lang.reflect.Method; + +import static org.apache.flink.table.planner.codegen.calls.BuiltInMethods.DECIMAL_TO_DOUBLE; +import static org.apache.flink.table.planner.codegen.calls.BuiltInMethods.DECIMAL_TO_INTEGRAL; +import static org.apache.flink.table.planner.functions.casting.rules.CastRuleUtils.castToPrimitive; +import static org.apache.flink.table.planner.functions.casting.rules.CastRuleUtils.staticCall; + +/** + * {@link LogicalTypeRoot#DECIMAL} to {@link LogicalTypeFamily#INTEGER_NUMERIC} and {@link + * LogicalTypeFamily#APPROXIMATE_NUMERIC} cast rule. + */ @Internal -public class UpcastToBigIntCastRule extends AbstractExpressionCodeGeneratorCastRule<Object, Long> { +public class DecimalToNumericPrimitiveCastRule + extends AbstractExpressionCodeGeneratorCastRule<DecimalData, Number> { - public static final UpcastToBigIntCastRule INSTANCE = new UpcastToBigIntCastRule(); + public static final DecimalToNumericPrimitiveCastRule INSTANCE = + new DecimalToNumericPrimitiveCastRule(); - private UpcastToBigIntCastRule() { + private DecimalToNumericPrimitiveCastRule() { super( CastRulePredicate.builder() - .input(LogicalTypeRoot.TINYINT) - .input(LogicalTypeRoot.SMALLINT) - .input(LogicalTypeRoot.INTEGER) - .target(LogicalTypeRoot.BIGINT) + .input(LogicalTypeRoot.DECIMAL) + .target(LogicalTypeFamily.INTEGER_NUMERIC) + .target(LogicalTypeFamily.APPROXIMATE_NUMERIC) .build()); } @@ -46,6 +59,10 @@ public class UpcastToBigIntCastRule extends AbstractExpressionCodeGeneratorCastR String inputTerm, LogicalType inputLogicalType, LogicalType targetLogicalType) { - return "((long)(" + inputTerm + "))"; + Method method = + targetLogicalType.is(LogicalTypeFamily.INTEGER_NUMERIC) + ? DECIMAL_TO_INTEGRAL() + : DECIMAL_TO_DOUBLE(); + return castToPrimitive(targetLogicalType, staticCall(method, inputTerm)); } } diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/IdentityCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/IdentityCastRule.java index 184520f..f769a71 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/IdentityCastRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/IdentityCastRule.java @@ -25,6 +25,7 @@ import org.apache.flink.table.planner.functions.casting.CodeGeneratorCastRule; import org.apache.flink.table.planner.functions.casting.ExpressionCodeGeneratorCastRule; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.LogicalTypeFamily; +import org.apache.flink.table.types.logical.LogicalTypeRoot; import org.apache.flink.table.types.logical.utils.LogicalTypeCasts; /** @@ -49,6 +50,23 @@ public class IdentityCastRule extends AbstractCodeGeneratorCastRule<Object, Obje && targetLogicalType.is(LogicalTypeFamily.CHARACTER_STRING)) { return true; } + + // INTERVAL_YEAR_MONTH and INTEGER uses the same primitive int type + if ((inputLogicalType.is(LogicalTypeRoot.INTERVAL_YEAR_MONTH) + && targetLogicalType.is(LogicalTypeRoot.INTEGER)) + || (inputLogicalType.is(LogicalTypeRoot.INTEGER) + && targetLogicalType.is(LogicalTypeRoot.INTERVAL_YEAR_MONTH))) { + return true; + } + + // INTERVAL_DAY_TIME and BIGINT uses the same primitive long type + if ((inputLogicalType.is(LogicalTypeRoot.INTERVAL_DAY_TIME) + && targetLogicalType.is(LogicalTypeRoot.BIGINT)) + || (inputLogicalType.is(LogicalTypeRoot.BIGINT) + && targetLogicalType.is(LogicalTypeRoot.INTERVAL_DAY_TIME))) { + return true; + } + return LogicalTypeCasts.supportsAvoidingCast(inputLogicalType, targetLogicalType); } diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/NumericPrimitiveCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/NumericPrimitiveCastRule.java new file mode 100644 index 0000000..6e20861 --- /dev/null +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/NumericPrimitiveCastRule.java @@ -0,0 +1,79 @@ +/* + * 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.rules; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.planner.functions.casting.CastRulePredicate; +import org.apache.flink.table.planner.functions.casting.CodeGeneratorCastRule; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.LogicalTypeFamily; + +import static org.apache.flink.table.types.logical.LogicalTypeFamily.APPROXIMATE_NUMERIC; +import static org.apache.flink.table.types.logical.LogicalTypeFamily.INTEGER_NUMERIC; +import static org.apache.flink.table.types.logical.LogicalTypeRoot.BIGINT; +import static org.apache.flink.table.types.logical.LogicalTypeRoot.INTEGER; +import static org.apache.flink.table.types.logical.LogicalTypeRoot.INTERVAL_DAY_TIME; +import static org.apache.flink.table.types.logical.LogicalTypeRoot.INTERVAL_YEAR_MONTH; + +/** + * Cast rule for {@link LogicalTypeFamily#INTEGER_NUMERIC} and {@link + * LogicalTypeFamily#APPROXIMATE_NUMERIC} and {@link LogicalTypeFamily#INTERVAL} conversions. + */ +@Internal +public class NumericPrimitiveCastRule + extends AbstractExpressionCodeGeneratorCastRule<Number, Number> { + + public static final NumericPrimitiveCastRule INSTANCE = new NumericPrimitiveCastRule(); + + private NumericPrimitiveCastRule() { + super(CastRulePredicate.builder().predicate(NumericPrimitiveCastRule::matches).build()); + } + + private static boolean matches(LogicalType input, LogicalType target) { + // Exclude identity casting + if (input.is(target.getTypeRoot())) { + return false; + } + + // Conversions between primitive numerics + if ((input.is(INTEGER_NUMERIC) || input.is(APPROXIMATE_NUMERIC)) + && (target.is(INTEGER_NUMERIC) || target.is(APPROXIMATE_NUMERIC))) { + return true; + } + + // Conversions between Interval year month (int) and bigint (long) + if ((input.is(INTERVAL_YEAR_MONTH) && target.is(BIGINT)) + || (input.is(BIGINT) && target.is(INTERVAL_YEAR_MONTH))) { + return true; + } + + // Conversions between Interval day time (long) and integer (int) + return (input.is(INTERVAL_DAY_TIME) && target.is(INTEGER)) + || (input.is(INTEGER) && target.is(INTERVAL_DAY_TIME)); + } + + @Override + public String generateExpression( + CodeGeneratorCastRule.Context context, + String inputTerm, + LogicalType inputLogicalType, + LogicalType targetLogicalType) { + return CastRuleUtils.castToPrimitive(targetLogicalType, inputTerm); + } +} diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/NumericPrimitiveToDecimalCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/NumericPrimitiveToDecimalCastRule.java new file mode 100644 index 0000000..df62905 --- /dev/null +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/rules/NumericPrimitiveToDecimalCastRule.java @@ -0,0 +1,75 @@ +/* + * 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.rules; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.planner.functions.casting.CastRulePredicate; +import org.apache.flink.table.planner.functions.casting.CodeGeneratorCastRule; +import org.apache.flink.table.types.logical.DecimalType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.LogicalTypeFamily; +import org.apache.flink.table.types.logical.LogicalTypeRoot; + +import static org.apache.flink.table.planner.codegen.calls.BuiltInMethods.DOUBLE_TO_DECIMAL; +import static org.apache.flink.table.planner.codegen.calls.BuiltInMethods.INTEGRAL_TO_DECIMAL; +import static org.apache.flink.table.planner.functions.casting.rules.CastRuleUtils.cast; +import static org.apache.flink.table.planner.functions.casting.rules.CastRuleUtils.staticCall; + +/** + * {@link LogicalTypeFamily#INTEGER_NUMERIC} and {@link LogicalTypeFamily#APPROXIMATE_NUMERIC} to + * {@link LogicalTypeRoot#DECIMAL} cast rule. + */ +@Internal +public class NumericPrimitiveToDecimalCastRule + extends AbstractExpressionCodeGeneratorCastRule<Number, DecimalData> { + + public static final NumericPrimitiveToDecimalCastRule INSTANCE = + new NumericPrimitiveToDecimalCastRule(); + + private NumericPrimitiveToDecimalCastRule() { + super( + CastRulePredicate.builder() + .input(LogicalTypeFamily.INTEGER_NUMERIC) + .input(LogicalTypeFamily.APPROXIMATE_NUMERIC) + .target(LogicalTypeRoot.DECIMAL) + .build()); + } + + @Override + public String generateExpression( + CodeGeneratorCastRule.Context context, + String inputTerm, + LogicalType inputLogicalType, + LogicalType targetLogicalType) { + final DecimalType targetDecimalType = (DecimalType) targetLogicalType; + if (inputLogicalType.is(LogicalTypeFamily.INTEGER_NUMERIC)) { + return staticCall( + INTEGRAL_TO_DECIMAL(), + cast("long", inputTerm), + targetDecimalType.getPrecision(), + targetDecimalType.getScale()); + } + return staticCall( + DOUBLE_TO_DECIMAL(), + cast("double", inputTerm), + targetDecimalType.getPrecision(), + targetDecimalType.getScale()); + } +} diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala index f63f3aa..22bb463 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala @@ -195,38 +195,6 @@ object CodeGenUtils { case _ => boxedTypeTermForType(t) } - /** - * Execute primitive unboxing. - */ - def unbox(term: String, ty: LogicalType): String = ty.getTypeRoot match { - // ordered by type root definition - case BOOLEAN => s"$term.booleanValue()" - case TINYINT => s"$term.byteValue()" - case SMALLINT => s"$term.shortValue()" - case INTEGER | DATE | TIME_WITHOUT_TIME_ZONE | INTERVAL_YEAR_MONTH => s"$term.intValue()" - case BIGINT | INTERVAL_DAY_TIME => s"$term.longValue()" - case FLOAT => s"$term.floatValue()" - case DOUBLE => s"$term.doubleValue()" - case DISTINCT_TYPE => unbox(term, ty.asInstanceOf[DistinctType].getSourceType) - case _ => term - } - - /** - * Execute primitive unboxing. - */ - def box(term: String, ty: LogicalType): String = ty.getTypeRoot match { - // ordered by type root definition - case BOOLEAN => s"Boolean.valueOf($term)" - case TINYINT => s"Byte.valueOf($term)" - case SMALLINT => s"Short.valueOf($term)" - case INTEGER | DATE | TIME_WITHOUT_TIME_ZONE | INTERVAL_YEAR_MONTH => s"Integer.valueOf($term)" - case BIGINT | INTERVAL_DAY_TIME => s"Long.valueOf($term)" - case FLOAT => s"Float.valueOf($term)" - case DOUBLE => s"Double.valueOf($term)" - case DISTINCT_TYPE => unbox(term, ty.asInstanceOf[DistinctType].getSourceType) - case _ => term - } - @tailrec def boxedTypeTermForType(t: LogicalType): String = t.getTypeRoot match { // ordered by type root definition @@ -254,6 +222,22 @@ object CodeGenUtils { } /** + * Returns true if [[primitiveDefaultValue()]] returns a nullable Java type, that is, + * a non primitive type. + */ + @tailrec + def isPrimitiveNullable(t: LogicalType): Boolean = t.getTypeRoot match { + // ordered by type root definition + case BOOLEAN | TINYINT | SMALLINT | INTEGER | + DATE | TIME_WITHOUT_TIME_ZONE | INTERVAL_YEAR_MONTH | + BIGINT | INTERVAL_DAY_TIME | FLOAT | DOUBLE => false + + case DISTINCT_TYPE => isPrimitiveNullable(t.asInstanceOf[DistinctType].getSourceType) + + case _ => true + } + + /** * Gets the default value for a primitive type, and null for generic types */ @tailrec @@ -328,69 +312,6 @@ object CodeGenUtils { throw new IllegalArgumentException("Illegal type: " + t) } - // ---------------------------------------------------------------------------------------------- - - // Cast numeric type to another numeric type with larger range. - // This function must be in sync with [[NumericOrDefaultReturnTypeInference]]. - def getNumericCastedResultTerm(expr: GeneratedExpression, targetType: LogicalType): String = { - (expr.resultType.getTypeRoot, targetType.getTypeRoot) match { - case _ if isInteroperable(expr.resultType, targetType) => expr.resultTerm - - // byte -> other numeric types - case (TINYINT, SMALLINT) => s"(short) ${expr.resultTerm}" - case (TINYINT, INTEGER) => s"(int) ${expr.resultTerm}" - case (TINYINT, BIGINT) => s"(long) ${expr.resultTerm}" - case (TINYINT, DECIMAL) => - val dt = targetType.asInstanceOf[DecimalType] - s"$DECIMAL_UTIL.castFrom(" + - s"${expr.resultTerm}, ${dt.getPrecision}, ${dt.getScale})" - case (TINYINT, FLOAT) => s"(float) ${expr.resultTerm}" - case (TINYINT, DOUBLE) => s"(double) ${expr.resultTerm}" - - // short -> other numeric types - case (SMALLINT, INTEGER) => s"(int) ${expr.resultTerm}" - case (SMALLINT, BIGINT) => s"(long) ${expr.resultTerm}" - case (SMALLINT, DECIMAL) => - val dt = targetType.asInstanceOf[DecimalType] - s"$DECIMAL_UTIL.castFrom(" + - s"${expr.resultTerm}, ${dt.getPrecision}, ${dt.getScale})" - case (SMALLINT, FLOAT) => s"(float) ${expr.resultTerm}" - case (SMALLINT, DOUBLE) => s"(double) ${expr.resultTerm}" - - // int -> other numeric types - case (INTEGER, BIGINT) => s"(long) ${expr.resultTerm}" - case (INTEGER, DECIMAL) => - val dt = targetType.asInstanceOf[DecimalType] - s"$DECIMAL_UTIL.castFrom(" + - s"${expr.resultTerm}, ${dt.getPrecision}, ${dt.getScale})" - case (INTEGER, FLOAT) => s"(float) ${expr.resultTerm}" - case (INTEGER, DOUBLE) => s"(double) ${expr.resultTerm}" - - // long -> other numeric types - case (BIGINT, DECIMAL) => - val dt = targetType.asInstanceOf[DecimalType] - s"$DECIMAL_UTIL.castFrom(" + - s"${expr.resultTerm}, ${dt.getPrecision}, ${dt.getScale})" - case (BIGINT, FLOAT) => s"(float) ${expr.resultTerm}" - case (BIGINT, DOUBLE) => s"(double) ${expr.resultTerm}" - - // decimal -> other numeric types - case (DECIMAL, DECIMAL) => - val dt = targetType.asInstanceOf[DecimalType] - s"$DECIMAL_UTIL.castToDecimal(" + - s"${expr.resultTerm}, ${dt.getPrecision}, ${dt.getScale})" - case (DECIMAL, FLOAT) => - s"$DECIMAL_UTIL.castToFloat(${expr.resultTerm})" - case (DECIMAL, DOUBLE) => - s"$DECIMAL_UTIL.castToDouble(${expr.resultTerm})" - - // float -> other numeric types - case (FLOAT, DOUBLE) => s"(double) ${expr.resultTerm}" - - case _ => null - } - } - // -------------------------- Method & Enum --------------------------------------- def qualifyMethod(method: Method): String = diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala index 6895214..6009cc1 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala @@ -823,13 +823,13 @@ class ExprCodeGenerator(ctx: CodeGeneratorContext, nullableInput: Boolean) operands.foreach { operand => requireComparable(operand) } - generateGreatestLeast(resultType, operands) + generateGreatestLeast(ctx, resultType, operands) case BuiltInFunctionDefinitions.LEAST => operands.foreach { operand => requireComparable(operand) } - generateGreatestLeast(resultType, operands, greatest = false) + generateGreatestLeast(ctx, resultType, operands, greatest = false) case BuiltInFunctionDefinitions.JSON_STRING => new JsonStringCallGen(call).generate(ctx, operands, resultType) diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala index 55f820e..15132b0 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala @@ -18,15 +18,14 @@ package org.apache.flink.table.planner.codegen.calls -import org.apache.flink.table.data.{DecimalData, TimestampData} +import org.apache.flink.table.data.{DecimalData, DecimalDataUtils, TimestampData} import org.apache.flink.table.runtime.functions._ import org.apache.flink.table.utils.DateTimeUtils import org.apache.flink.table.utils.DateTimeUtils.TimeUnitRange import org.apache.calcite.linq4j.tree.Types import org.apache.calcite.runtime.{JsonFunctions, SqlFunctions} -import org.apache.calcite.sql.{SqlJsonConstructorNullClause, SqlJsonExistsErrorBehavior, SqlJsonQueryEmptyOrErrorBehavior, SqlJsonQueryWrapperBehavior, SqlJsonValueEmptyOrErrorBehavior} -import org.apache.calcite.sql.{SqlJsonConstructorNullClause, SqlJsonExistsErrorBehavior, SqlJsonQueryEmptyOrErrorBehavior, SqlJsonQueryWrapperBehavior, SqlJsonValueEmptyOrErrorBehavior} +import org.apache.calcite.sql.{SqlJsonExistsErrorBehavior, SqlJsonQueryEmptyOrErrorBehavior, SqlJsonQueryWrapperBehavior, SqlJsonValueEmptyOrErrorBehavior} import org.apache.flink.table.data.binary.BinaryStringData import java.lang.reflect.Method @@ -548,4 +547,26 @@ object BuiltInMethods { val BINARY_STRING_DATA_FROM_STRING = Types.lookupMethod(classOf[BinaryStringData], "fromString", classOf[String]) + // DecimalData functions + + val DECIMAL_TO_DECIMAL = Types.lookupMethod( + classOf[DecimalDataUtils], + "castToDecimal", classOf[DecimalData], classOf[Int], classOf[Int]) + + val DECIMAL_TO_INTEGRAL = Types.lookupMethod( + classOf[DecimalDataUtils], + "castToIntegral", classOf[DecimalData]) + + val DECIMAL_TO_DOUBLE = Types.lookupMethod( + classOf[DecimalDataUtils], + "doubleValue", classOf[DecimalData]) + + val INTEGRAL_TO_DECIMAL = Types.lookupMethod( + classOf[DecimalDataUtils], + "castFrom", classOf[Long], classOf[Int], classOf[Int]) + + val DOUBLE_TO_DECIMAL = Types.lookupMethod( + classOf[DecimalDataUtils], + "castFrom", classOf[Long], classOf[Int], classOf[Int]) + } diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/IfCallGen.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/IfCallGen.scala index fcc62c3..af8061c 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/IfCallGen.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/IfCallGen.scala @@ -18,8 +18,11 @@ package org.apache.flink.table.planner.codegen.calls -import org.apache.flink.table.planner.codegen.CodeGenUtils.{primitiveDefaultValue, primitiveTypeTermForType} -import org.apache.flink.table.planner.codegen.{CodeGenUtils, CodeGeneratorContext, GeneratedExpression} +import org.apache.flink.table.planner.codegen.CodeGenUtils.{className, primitiveDefaultValue, primitiveTypeTermForType} +import org.apache.flink.table.planner.codegen.calls.ScalarOperatorGens.toCastContext +import org.apache.flink.table.planner.codegen.{CodeGenException, CodeGenUtils, CodeGeneratorContext, GeneratedExpression} +import org.apache.flink.table.planner.functions.casting.{CastRuleProvider, ExpressionCodeGeneratorCastRule} +import org.apache.flink.table.runtime.types.PlannerTypeUtils.isInteroperable import org.apache.flink.table.types.logical.LogicalType /** @@ -36,8 +39,8 @@ class IfCallGen() extends CallGenerator { // Inferred return type is ARG1. Must be the same as ARG2. // This is a temporary solution which introduce type cast in codegen. // Not elegant, but can allow IF function to handle different numeric type arguments. - val castedResultTerm1 = CodeGenUtils.getNumericCastedResultTerm(operands(1), returnType) - val castedResultTerm2 = CodeGenUtils.getNumericCastedResultTerm(operands(2), returnType) + val castedResultTerm1 = normalizeArgument(ctx, operands(1), returnType) + val castedResultTerm2 = normalizeArgument(ctx, operands(2), returnType) if (castedResultTerm1 == null || castedResultTerm2 == null) { throw new Exception(String.format("Unsupported operand types: IF(boolean, %s, %s)", operands(1).resultType, operands(2).resultType)) @@ -51,6 +54,7 @@ class IfCallGen() extends CallGenerator { val resultCode = s""" + |// --- Start code generated by ${className[IfCallGen]} |${operands.head.code} |$resultTerm = $resultDefault; |if (${operands.head.resultTerm}) { @@ -66,8 +70,28 @@ class IfCallGen() extends CallGenerator { | } | $nullTerm = ${operands(2).nullTerm}; |} + |// --- End code generated by ${className[IfCallGen]} """.stripMargin GeneratedExpression(resultTerm, nullTerm, resultCode, returnType) } + + /** + * This function will return the argument term casted if an expression casting can be performed, + * or null if no casting can be performed + */ + private def normalizeArgument( + ctx: CodeGeneratorContext, expr: GeneratedExpression, targetType: LogicalType): String = { + val rule = CastRuleProvider.resolve(expr.resultType, targetType) + rule match { + case codeGeneratorCastRule: ExpressionCodeGeneratorCastRule[_, _] => + codeGeneratorCastRule.generateExpression( + toCastContext(ctx), + expr.resultTerm, + expr.resultType, + targetType + ) + case _ => null + } + } } diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala index 6532bd7..3021776 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala @@ -22,13 +22,14 @@ import org.apache.flink.table.api.{DataTypes, ValidationException} import org.apache.flink.table.connector.sink.DynamicTableSink.DataStructureConverter import org.apache.flink.table.data.binary.BinaryArrayData import org.apache.flink.table.data.conversion.DataStructureConverters -import org.apache.flink.table.planner.functions.casting.{CastRule, CastRuleProvider, CodeGeneratorCastRule} +import org.apache.flink.table.planner.functions.casting.{CastRule, CastRuleProvider, CodeGeneratorCastRule, ExpressionCodeGeneratorCastRule} import org.apache.flink.table.data.util.{DataFormatConverters, MapDataUtil} import org.apache.flink.table.data.writer.{BinaryArrayWriter, BinaryRowWriter} import org.apache.flink.table.planner.codegen.CodeGenUtils.{binaryRowFieldSetAccess, binaryRowSetNull, binaryWriterWriteField, binaryWriterWriteNull, _} import org.apache.flink.table.planner.codegen.GenerateUtils._ import org.apache.flink.table.planner.codegen.GeneratedExpression.{ALWAYS_NULL, NEVER_NULL, NO_CODE} import org.apache.flink.table.planner.codegen.{CodeGenException, CodeGenUtils, CodeGeneratorContext, GenerateUtils, GeneratedExpression} +import org.apache.flink.table.planner.functions.casting.rules.{AbstractExpressionCodeGeneratorCastRule, IdentityCastRule} import org.apache.flink.table.planner.utils.JavaScalaConversionUtil.toScala import org.apache.flink.table.runtime.functions.SqlFunctionUtils import org.apache.flink.table.runtime.types.LogicalTypeDataTypeConverter.fromLogicalTypeToDataType @@ -81,19 +82,19 @@ object ScalarOperatorGens { val leftCasting = operator match { case "%" => if (isInteroperable(left.resultType, right.resultType)) { - numericCasting(left.resultType, resultType) + numericCasting(ctx, left.resultType, resultType) } else { val castedType = if (isDecimal(left.resultType)) { new BigIntType() } else { left.resultType } - numericCasting(left.resultType, castedType) + numericCasting(ctx, left.resultType, castedType) } - case _ => numericCasting(left.resultType, resultType) + case _ => numericCasting(ctx, left.resultType, resultType) } - val rightCasting = numericCasting(right.resultType, resultType) + val rightCasting = numericCasting(ctx, right.resultType, resultType) val resultTypeTerm = primitiveTypeTermForType(resultType) generateOperatorIfNotNull(ctx, resultType, left, right) { @@ -117,7 +118,7 @@ object ScalarOperatorGens { // use it as is during calculation. def castToDec(t: LogicalType): String => String = t match { case _: DecimalType => (operandTerm: String) => s"$operandTerm" - case _ => numericCasting(t, resultType) + case _ => numericCasting(ctx, t, resultType) } val methods = Map( "+" -> "add", @@ -358,7 +359,7 @@ object ScalarOperatorGens { // we need to normalize the values for the hash set val castNumeric = widerType match { case Some(t) => (value: GeneratedExpression) => - numericCasting(value.resultType, t)(value.resultTerm) + numericCasting(ctx, value.resultType, t)(value.resultTerm) case None => (value: GeneratedExpression) => value.resultTerm } @@ -951,19 +952,8 @@ object ScalarOperatorGens { } // Generate the code block - val castContext = new CodeGeneratorCastRule.Context { - override def getSessionTimeZoneTerm: String = ctx.addReusableSessionTimeZone() - override def declareVariable(ty: String, variablePrefix: String): String = - ctx.addReusableLocalVariable(ty, variablePrefix) - override def declareTypeSerializer(ty: LogicalType): String = - ctx.addReusableTypeSerializer(ty) - override def declareClassField(ty: String, field: String, init: String): String = { - ctx.addReusableMember(s"private $ty $field = $init;") - field - } - } val castCodeBlock = codeGeneratorCastRule.generateCodeBlock( - castContext, + toCastContext(ctx), operand.resultTerm, operand.nullTerm, inputType, @@ -1203,13 +1193,6 @@ object ScalarOperatorGens { operandTerm => s"$operandTerm != 0" } - // between NUMERIC TYPE | Decimal - case (_, _) if isNumeric(operand.resultType) && isNumeric(targetType) => - val operandCasting = numericCasting(operand.resultType, targetType) - generateUnaryOperatorIfNotNull(ctx, targetType, operand) { - operandTerm => s"${operandCasting(operandTerm)}" - } - // Date -> Timestamp case (DATE, TIMESTAMP_WITHOUT_TIME_ZONE) => generateUnaryOperatorIfNotNull(ctx, targetType, operand) { @@ -1795,6 +1778,7 @@ object ScalarOperatorGens { * returns NULL if any argument is NULL. */ def generateGreatestLeast( + ctx: CodeGeneratorContext, resultType: LogicalType, elements: Seq[GeneratedExpression], greatest: Boolean = true) @@ -1806,7 +1790,7 @@ object ScalarOperatorGens { def castIfNumeric(t: GeneratedExpression): String = { if (isNumeric(widerType.get)) { - s"${numericCasting(t.resultType, widerType.get).apply(t.resultTerm)}" + s"${numericCasting(ctx, t.resultType, widerType.get).apply(t.resultTerm)}" } else { s"${t.resultTerm}" } @@ -2247,58 +2231,41 @@ object ScalarOperatorGens { expr.copy(resultType = targetType) } + /** + * @deprecated You should use [[generateCast()]] + */ + @deprecated private def numericCasting( + ctx: CodeGeneratorContext, operandType: LogicalType, resultType: LogicalType): String => String = { - val resultTypeTerm = primitiveTypeTermForType(resultType) - - def decToPrimMethod(targetType: LogicalType): String = targetType.getTypeRoot match { - case TINYINT => "castToByte" - case SMALLINT => "castToShort" - case INTEGER => "castToInt" - case BIGINT => "castToLong" - case FLOAT => "castToFloat" - case DOUBLE => "castToDouble" - case BOOLEAN => "castToBoolean" - case _ => throw new CodeGenException(s"Unsupported decimal casting type: '$targetType'") + // All numeric rules are assumed to be instance of AbstractExpressionCodeGeneratorCastRule + val rule = CastRuleProvider.resolve(operandType, resultType) + rule match { + case codeGeneratorCastRule: ExpressionCodeGeneratorCastRule[_, _] => + operandTerm => codeGeneratorCastRule.generateExpression( + toCastContext(ctx), + operandTerm, + operandType, + resultType + ) + case _ => + throw new CodeGenException(s"Unsupported casting from $operandType to $resultType.") } + } - // no casting necessary - if (isInteroperable(operandType, resultType)) { - operandTerm => s"$operandTerm" - } - // decimal to decimal, may have different precision/scale - else if (isDecimal(resultType) && isDecimal(operandType)) { - val dt = resultType.asInstanceOf[DecimalType] - operandTerm => - s"$DECIMAL_UTIL.castToDecimal($operandTerm, ${dt.getPrecision}, ${dt.getScale})" - } - // non_decimal_numeric to decimal - else if (isDecimal(resultType) && isNumeric(operandType)) { - val dt = resultType.asInstanceOf[DecimalType] - operandTerm => - s"$DECIMAL_UTIL.castFrom($operandTerm, ${dt.getPrecision}, ${dt.getScale})" - } - // decimal to non_decimal_numeric - else if (isNumeric(resultType) && isDecimal(operandType) ) { - operandTerm => - s"$DECIMAL_UTIL.${decToPrimMethod(resultType)}($operandTerm)" - } - // numeric to numeric - // TODO: Create a wrapper layer that handles type conversion between numeric. - else if (isNumeric(operandType) && isNumeric(resultType)) { - val resultTypeValue = resultTypeTerm + "Value()" - val boxedTypeTerm = boxedTypeTermForType(operandType) - operandTerm => - s"(new $boxedTypeTerm($operandTerm)).$resultTypeValue" - } - // result type is time interval and operand type is integer - else if (isTimeInterval(resultType) && isInteger(operandType)){ - operandTerm => s"(($resultTypeTerm) $operandTerm)" - } - else { - throw new CodeGenException(s"Unsupported casting from $operandType to $resultType.") + def toCastContext(ctx: CodeGeneratorContext): CodeGeneratorCastRule.Context = { + new CodeGeneratorCastRule.Context { + override def getSessionTimeZoneTerm: String = ctx.addReusableSessionTimeZone() + override def declareVariable(ty: String, variablePrefix: String): String = + ctx.addReusableLocalVariable(ty, variablePrefix) + override def declareTypeSerializer(ty: LogicalType): String = + ctx.addReusableTypeSerializer(ty) + override def declareClassField(ty: String, field: String, init: String): String = { + ctx.addReusableMember(s"private $ty $field = $init;") + field + } } } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java index 63dbbe9..587fc5a 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java @@ -69,6 +69,7 @@ import static org.apache.flink.table.api.DataTypes.MAP; import static org.apache.flink.table.api.DataTypes.MONTH; import static org.apache.flink.table.api.DataTypes.RAW; import static org.apache.flink.table.api.DataTypes.ROW; +import static org.apache.flink.table.api.DataTypes.SECOND; import static org.apache.flink.table.api.DataTypes.SMALLINT; import static org.apache.flink.table.api.DataTypes.STRING; import static org.apache.flink.table.api.DataTypes.TIME; @@ -92,6 +93,19 @@ class CastRulesTest { CastRule.Context.create( ZoneId.of("CET"), Thread.currentThread().getContextClassLoader()); + private static final byte DEFAULT_POSITIVE_TINY_INT = (byte) 5; + private static final byte DEFAULT_NEGATIVE_TINY_INT = (byte) -5; + private static final short DEFAULT_POSITIVE_SMALL_INT = (short) 12345; + private static final short DEFAULT_NEGATIVE_SMALL_INT = (short) -12345; + private static final int DEFAULT_POSITIVE_INT = 1234567; + private static final int DEFAULT_NEGATIVE_INT = -1234567; + private static final long DEFAULT_POSITIVE_BIGINT = 12345678901L; + private static final long DEFAULT_NEGATIVE_BIGINT = -12345678901L; + private static final float DEFAULT_POSITIVE_FLOAT = 123.456f; + private static final float DEFAULT_NEGATIVE_FLOAT = -123.456f; + private static final double DEFAULT_POSITIVE_DOUBLE = 123.456789d; + private static final double DEFAULT_NEGATIVE_DOUBLE = -123.456789d; + private static final int DATE = DateTimeUtils.localDateToUnixDate(LocalDate.parse("2021-09-24")); private static final int TIME = @@ -108,11 +122,232 @@ class CastRulesTest { Stream<CastTestSpecBuilder> testCases() { return Stream.of( + CastTestSpecBuilder.testCastTo(TINYINT()) + .fromCase(TINYINT(), null, null) + .fromCase( + DECIMAL(4, 3), + DecimalData.fromBigDecimal(new BigDecimal("9.87"), 4, 3), + (byte) 9) + // https://issues.apache.org/jira/browse/FLINK-24420 - Check out of range + // instead of overflow + .fromCase( + DECIMAL(10, 3), + DecimalData.fromBigDecimal(new BigDecimal("9123.87"), 10, 3), + (byte) -93) + .fromCase(TINYINT(), DEFAULT_POSITIVE_TINY_INT, DEFAULT_POSITIVE_TINY_INT) + .fromCase(TINYINT(), DEFAULT_NEGATIVE_TINY_INT, DEFAULT_NEGATIVE_TINY_INT) + .fromCase(SMALLINT(), (short) 32, (byte) 32) + .fromCase(SMALLINT(), DEFAULT_POSITIVE_SMALL_INT, (byte) 57) + .fromCase(SMALLINT(), DEFAULT_NEGATIVE_SMALL_INT, (byte) -57) + .fromCase(INT(), -12, (byte) -12) + .fromCase(INT(), DEFAULT_POSITIVE_INT, (byte) -121) + .fromCase(INT(), DEFAULT_NEGATIVE_INT, (byte) 121) + .fromCase(BIGINT(), DEFAULT_POSITIVE_BIGINT, (byte) 53) + .fromCase(BIGINT(), DEFAULT_NEGATIVE_BIGINT, (byte) -53) + .fromCase(FLOAT(), DEFAULT_POSITIVE_FLOAT, (byte) 123) + .fromCase(FLOAT(), DEFAULT_NEGATIVE_FLOAT, (byte) -123) + .fromCase(DOUBLE(), DEFAULT_POSITIVE_DOUBLE, (byte) 123) + .fromCase(DOUBLE(), DEFAULT_NEGATIVE_DOUBLE, (byte) -123), + CastTestSpecBuilder.testCastTo(SMALLINT()) + .fromCase(SMALLINT(), null, null) + .fromCase( + DECIMAL(4, 3), + DecimalData.fromBigDecimal(new BigDecimal("9.87"), 4, 3), + (short) 9) + // https://issues.apache.org/jira/browse/FLINK-24420 - Check out of range + // instead of overflow + .fromCase( + DECIMAL(10, 3), + DecimalData.fromBigDecimal(new BigDecimal("91235.87"), 10, 3), + (short) 25699) + .fromCase( + TINYINT(), + DEFAULT_POSITIVE_TINY_INT, + (short) DEFAULT_POSITIVE_TINY_INT) + .fromCase( + TINYINT(), + DEFAULT_NEGATIVE_TINY_INT, + (short) DEFAULT_NEGATIVE_TINY_INT) + .fromCase( + SMALLINT(), DEFAULT_POSITIVE_SMALL_INT, DEFAULT_POSITIVE_SMALL_INT) + .fromCase( + SMALLINT(), DEFAULT_NEGATIVE_SMALL_INT, DEFAULT_NEGATIVE_SMALL_INT) + .fromCase(SMALLINT(), (short) 32780, (short) -32756) + .fromCase(INT(), DEFAULT_POSITIVE_INT, (short) -10617) + .fromCase(INT(), DEFAULT_NEGATIVE_INT, (short) 10617) + .fromCase(INT(), -12, (short) -12) + .fromCase(BIGINT(), 123L, (short) 123) + .fromCase(BIGINT(), DEFAULT_POSITIVE_BIGINT, (short) 7221) + .fromCase(BIGINT(), DEFAULT_NEGATIVE_BIGINT, (short) -7221) + .fromCase(FLOAT(), DEFAULT_POSITIVE_FLOAT, (short) 123) + .fromCase(FLOAT(), DEFAULT_NEGATIVE_FLOAT, (short) -123) + .fromCase(FLOAT(), 123456.78f, (short) -7616) + .fromCase(DOUBLE(), DEFAULT_POSITIVE_DOUBLE, (short) 123) + .fromCase(DOUBLE(), DEFAULT_NEGATIVE_DOUBLE, (short) -123) + .fromCase(DOUBLE(), 123456.7890d, (short) -7616), + CastTestSpecBuilder.testCastTo(INT()) + .fromCase( + DECIMAL(4, 3), + DecimalData.fromBigDecimal(new BigDecimal("9.87"), 4, 3), + 9) + // https://issues.apache.org/jira/browse/FLINK-24420 - Check out of range + // instead of overflow + .fromCase( + DECIMAL(20, 3), + DecimalData.fromBigDecimal( + new BigDecimal("3276913443134.87"), 20, 3), + -146603714) + .fromCase( + TINYINT(), + DEFAULT_POSITIVE_TINY_INT, + (int) DEFAULT_POSITIVE_TINY_INT) + .fromCase( + TINYINT(), + DEFAULT_NEGATIVE_TINY_INT, + (int) DEFAULT_NEGATIVE_TINY_INT) + .fromCase( + SMALLINT(), + DEFAULT_POSITIVE_SMALL_INT, + (int) DEFAULT_POSITIVE_SMALL_INT) + .fromCase( + SMALLINT(), + DEFAULT_NEGATIVE_SMALL_INT, + (int) DEFAULT_NEGATIVE_SMALL_INT) + .fromCase(INT(), DEFAULT_POSITIVE_INT, DEFAULT_POSITIVE_INT) + .fromCase(INT(), DEFAULT_NEGATIVE_INT, DEFAULT_NEGATIVE_INT) + .fromCase(BIGINT(), 123L, 123) + .fromCase(BIGINT(), DEFAULT_POSITIVE_BIGINT, -539222987) + .fromCase(BIGINT(), DEFAULT_NEGATIVE_BIGINT, 539222987) + .fromCase(FLOAT(), DEFAULT_POSITIVE_FLOAT, 123) + .fromCase(FLOAT(), DEFAULT_NEGATIVE_FLOAT, -123) + .fromCase(FLOAT(), 9234567891.12f, 2147483647) + .fromCase(DOUBLE(), DEFAULT_POSITIVE_DOUBLE, 123) + .fromCase(DOUBLE(), DEFAULT_NEGATIVE_DOUBLE, -123) + .fromCase(DOUBLE(), 9234567891.12345d, 2147483647) + .fromCase(INTERVAL(YEAR(), MONTH()), 123, 123) + .fromCase(INTERVAL(DAY(), SECOND()), 123L, 123), CastTestSpecBuilder.testCastTo(BIGINT()) - .fromCase(BIGINT(), 10L, 10L) - .fromCase(INT(), 10, 10L) - .fromCase(SMALLINT(), (short) 10, 10L) - .fromCase(TINYINT(), (byte) 10, 10L), + .fromCase(BIGINT(), null, null) + .fromCase( + DECIMAL(4, 3), + DecimalData.fromBigDecimal(new BigDecimal("9.87"), 4, 3), + 9L) + .fromCase( + DECIMAL(20, 3), + DecimalData.fromBigDecimal( + new BigDecimal("3276913443134.87"), 20, 3), + 3276913443134L) + .fromCase( + TINYINT(), + DEFAULT_POSITIVE_TINY_INT, + (long) DEFAULT_POSITIVE_TINY_INT) + .fromCase( + TINYINT(), + DEFAULT_NEGATIVE_TINY_INT, + (long) DEFAULT_NEGATIVE_TINY_INT) + .fromCase( + SMALLINT(), + DEFAULT_POSITIVE_SMALL_INT, + (long) DEFAULT_POSITIVE_SMALL_INT) + .fromCase( + SMALLINT(), + DEFAULT_NEGATIVE_SMALL_INT, + (long) DEFAULT_NEGATIVE_SMALL_INT) + .fromCase(INT(), DEFAULT_POSITIVE_INT, (long) DEFAULT_POSITIVE_INT) + .fromCase(INT(), DEFAULT_NEGATIVE_INT, (long) DEFAULT_NEGATIVE_INT) + .fromCase(BIGINT(), DEFAULT_POSITIVE_BIGINT, DEFAULT_POSITIVE_BIGINT) + .fromCase(BIGINT(), DEFAULT_NEGATIVE_BIGINT, DEFAULT_NEGATIVE_BIGINT) + .fromCase(FLOAT(), DEFAULT_POSITIVE_FLOAT, 123L) + .fromCase(FLOAT(), DEFAULT_NEGATIVE_FLOAT, -123L) + .fromCase(FLOAT(), 9234567891.12f, 9234568192L) + .fromCase(DOUBLE(), DEFAULT_POSITIVE_DOUBLE, 123L) + .fromCase(DOUBLE(), DEFAULT_NEGATIVE_DOUBLE, -123L) + .fromCase(DOUBLE(), 9234567891.12345d, 9234567891L), + CastTestSpecBuilder.testCastTo(FLOAT()) + .fromCase(FLOAT(), null, null) + .fromCase( + DECIMAL(4, 3), + DecimalData.fromBigDecimal(new BigDecimal("9.87"), 4, 3), + 9.87f) + // https://issues.apache.org/jira/browse/FLINK-24420 - Check out of range + // instead of overflow + .fromCase( + DECIMAL(20, 3), + DecimalData.fromBigDecimal( + new BigDecimal("3276913443134.87"), 20, 3), + 3.27691351E12f) + .fromCase( + TINYINT(), + DEFAULT_POSITIVE_TINY_INT, + (float) DEFAULT_POSITIVE_TINY_INT) + .fromCase( + TINYINT(), + DEFAULT_NEGATIVE_TINY_INT, + (float) DEFAULT_NEGATIVE_TINY_INT) + .fromCase( + SMALLINT(), + DEFAULT_POSITIVE_SMALL_INT, + (float) DEFAULT_POSITIVE_SMALL_INT) + .fromCase( + SMALLINT(), + DEFAULT_NEGATIVE_SMALL_INT, + (float) DEFAULT_NEGATIVE_SMALL_INT) + .fromCase(INT(), DEFAULT_POSITIVE_INT, (float) DEFAULT_POSITIVE_INT) + .fromCase(INT(), DEFAULT_NEGATIVE_INT, (float) DEFAULT_NEGATIVE_INT) + .fromCase( + BIGINT(), DEFAULT_POSITIVE_BIGINT, (float) DEFAULT_POSITIVE_BIGINT) + .fromCase( + BIGINT(), DEFAULT_NEGATIVE_BIGINT, (float) DEFAULT_NEGATIVE_BIGINT) + .fromCase(FLOAT(), DEFAULT_POSITIVE_FLOAT, DEFAULT_POSITIVE_FLOAT) + .fromCase(FLOAT(), DEFAULT_NEGATIVE_FLOAT, DEFAULT_NEGATIVE_FLOAT) + .fromCase(FLOAT(), 9234567891.12f, 9234567891.12f) + .fromCase(DOUBLE(), DEFAULT_POSITIVE_DOUBLE, 123.456789f) + .fromCase(DOUBLE(), DEFAULT_NEGATIVE_DOUBLE, -123.456789f) + .fromCase(DOUBLE(), 1239234567891.1234567891234d, 1.23923451E12f), + CastTestSpecBuilder.testCastTo(DOUBLE()) + .fromCase(DOUBLE(), null, null) + .fromCase( + DECIMAL(4, 3), + DecimalData.fromBigDecimal(new BigDecimal("9.87"), 4, 3), + 9.87d) + .fromCase( + DECIMAL(20, 3), + DecimalData.fromBigDecimal( + new BigDecimal("3276913443134.87"), 20, 3), + 3.27691344313487E12d) + .fromCase( + DECIMAL(30, 20), + DecimalData.fromBigDecimal( + new BigDecimal("123456789.123456789123456789"), 30, 20), + 1.2345678912345679E8d) + .fromCase( + TINYINT(), + DEFAULT_POSITIVE_TINY_INT, + (double) DEFAULT_POSITIVE_TINY_INT) + .fromCase( + TINYINT(), + DEFAULT_NEGATIVE_TINY_INT, + (double) DEFAULT_NEGATIVE_TINY_INT) + .fromCase( + SMALLINT(), + DEFAULT_POSITIVE_SMALL_INT, + (double) DEFAULT_POSITIVE_SMALL_INT) + .fromCase( + SMALLINT(), + DEFAULT_NEGATIVE_SMALL_INT, + (double) DEFAULT_NEGATIVE_SMALL_INT) + .fromCase(INT(), DEFAULT_POSITIVE_INT, (double) DEFAULT_POSITIVE_INT) + .fromCase(INT(), DEFAULT_NEGATIVE_INT, (double) DEFAULT_NEGATIVE_INT) + .fromCase( + BIGINT(), DEFAULT_POSITIVE_BIGINT, (double) DEFAULT_POSITIVE_BIGINT) + .fromCase( + BIGINT(), DEFAULT_NEGATIVE_BIGINT, (double) DEFAULT_NEGATIVE_BIGINT) + .fromCase(FLOAT(), DEFAULT_POSITIVE_FLOAT, 123.45600128173828d) + .fromCase(FLOAT(), DEFAULT_NEGATIVE_FLOAT, -123.45600128173828) + .fromCase(FLOAT(), 9234567891.12f, 9.234568192E9) + .fromCase(DOUBLE(), DEFAULT_POSITIVE_DOUBLE, DEFAULT_POSITIVE_DOUBLE) + .fromCase(DOUBLE(), DEFAULT_NEGATIVE_DOUBLE, DEFAULT_NEGATIVE_DOUBLE) + .fromCase(DOUBLE(), 1239234567891.1234567891234d, 1.2392345678911235E12d), CastTestSpecBuilder.testCastTo(STRING()) .fromCase(STRING(), null, null) .fromCase( @@ -236,6 +471,35 @@ class CastRulesTest { RawValueData.fromObject( LocalDateTime.parse("2020-11-11T18:08:01.123")), StringData.fromString("2020-11-11T18:08:01.123")), + CastTestSpecBuilder.testCastTo(DECIMAL(5, 3)) + .fromCase( + DECIMAL(4, 3), + DecimalData.fromBigDecimal(new BigDecimal("9.87"), 4, 3), + DecimalData.fromBigDecimal(new BigDecimal("9.870"), 5, 3)) + .fromCase( + TINYINT(), + (byte) -1, + DecimalData.fromBigDecimal(new BigDecimal("-1.000"), 5, 3)) + .fromCase( + SMALLINT(), + (short) 3, + DecimalData.fromBigDecimal(new BigDecimal("3.000"), 5, 3)) + .fromCase( + INT(), + 42, + DecimalData.fromBigDecimal(new BigDecimal("42.000"), 5, 3)) + .fromCase( + BIGINT(), + 8L, + DecimalData.fromBigDecimal(new BigDecimal("8.000"), 5, 3)) + .fromCase( + FLOAT(), + -12.345f, + DecimalData.fromBigDecimal(new BigDecimal("-12.345"), 5, 3)) + .fromCase( + DOUBLE(), + 12.678d, + DecimalData.fromBigDecimal(new BigDecimal("12.678"), 5, 3)), CastTestSpecBuilder.testCastTo(ARRAY(STRING().nullable())) .fromCase( ARRAY(TIMESTAMP().nullable()), diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/DecimalDataUtils.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/DecimalDataUtils.java index 9ea7055..bb9ba00 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/DecimalDataUtils.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/DecimalDataUtils.java @@ -173,30 +173,6 @@ public final class DecimalDataUtils { return bd.longValue(); } - public static long castToLong(DecimalData dec) { - return castToIntegral(dec); - } - - public static int castToInt(DecimalData dec) { - return (int) castToIntegral(dec); - } - - public static short castToShort(DecimalData dec) { - return (short) castToIntegral(dec); - } - - public static byte castToByte(DecimalData dec) { - return (byte) castToIntegral(dec); - } - - public static float castToFloat(DecimalData dec) { - return (float) doubleValue(dec); - } - - public static double castToDouble(DecimalData dec) { - return doubleValue(dec); - } - public static DecimalData castToDecimal(DecimalData dec, int precision, int scale) { return fromBigDecimal(dec.toBigDecimal(), precision, scale); }
