[flink] 02/04: [hotfix][table-planner] Fix primitives/boxed primitives behaviour of ExpressionEvaluator

[tw...@apache.org]

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 b51abc7c8c230507749b237a52fb9ba3eddf9e9c
Author: slinkydeveloper <fr...@gmail.com>
AuthorDate: Mon Nov 8 17:34:55 2021 +0100

    [hotfix][table-planner] Fix primitives/boxed primitives behaviour of ExpressionEvaluator
    
    Signed-off-by: slinkydeveloper <fr...@gmail.com>
---
 .../AbstractExpressionCodeGeneratorCastRule.java   | 23 ++++++++++++----
 .../flink/table/planner/codegen/CodeGenUtils.scala | 32 ++++++++++++++++++++++
 2 files changed, 50 insertions(+), 5 deletions(-)

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 f3ed23f..aba38f0 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,10 +30,15 @@ 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;
+
 /**
  * Base class for cast rules that supports code generation, requiring only an expression to perform
  * the cast. If the casting logic requires to generate several statements, look at {@link
  * AbstractNullAwareCodeGeneratorCastRule}.
+ *
+ * <p>NOTE: the {@code inputTerm} is always either a primitive or a non-null object.
  */
 @Internal
 public abstract class AbstractExpressionCodeGeneratorCastRule<IN, OUT>
@@ -63,11 +68,19 @@ public abstract class AbstractExpressionCodeGeneratorCastRule<IN, OUT>
         final String inputArgumentName = "inputValue";
 
         final String expression =
-                generateExpression(
-                        createCodeGeneratorCastRuleContext(context),
-                        inputArgumentName,
-                        inputLogicalType,
-                        targetLogicalType);
+                // We need to wrap the expression in a null check
+                CastRuleUtils.ternaryOperator(
+                        inputArgumentName + " == null",
+                        "null",
+                        // Values are always boxed when passed to ExpressionEvaluator and no auto
+                        // boxing/unboxing is provided, so we need to take care of it manually
+                        box(
+                                generateExpression(
+                                        createCodeGeneratorCastRuleContext(context),
+                                        unbox(inputArgumentName, inputLogicalType),
+                                        inputLogicalType,
+                                        targetLogicalType),
+                                targetLogicalType));
 
         return new CodeGeneratedExpressionCastExecutor<>(
                 CompileUtils.compileExpression(
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 748c583..f63f3aa 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,6 +195,38 @@ 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