[flink] branch release-1.5 updated: [FLINK-10642] [table] Fix CodeGen split errors when maxGeneratedCodeLength equals 1

[twalthr]

This is an automated email from the ASF dual-hosted git repository.

twalthr pushed a commit to branch release-1.5
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.5 by this push:
     new 1d8bc86  [FLINK-10642] [table] Fix CodeGen split errors when 
maxGeneratedCodeLength equals 1
1d8bc86 is described below

commit 1d8bc86d152cde2941750c99f6cfd2507c96afdb
Author: xueyu <278006...@qq.com>
AuthorDate: Tue Oct 23 00:16:48 2018 +0800

    [FLINK-10642] [table] Fix CodeGen split errors when maxGeneratedCodeLength 
equals 1
    
    This closes #6900.
---
 .../org/apache/flink/table/codegen/CodeGenerator.scala    |  4 ++--
 .../flink/table/runtime/batch/table/CalcITCase.scala      | 15 +++++++++++++++
 2 files changed, 17 insertions(+), 2 deletions(-)

diff --git 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
index f0b6793..57e3809 100644
--- 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
+++ 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
@@ -1059,13 +1059,13 @@ abstract class CodeGenerator(
 
         // declaration
         val resultTypeTerm = primitiveTypeTermForTypeInfo(expr.resultType)
-        if (nullCheck) {
+        if (nullCheck && !expr.nullTerm.equals(NEVER_NULL) && 
!expr.nullTerm.equals(ALWAYS_NULL)) {
           reusableMemberStatements.add(s"private boolean ${expr.nullTerm};")
         }
         reusableMemberStatements.add(s"private $resultTypeTerm 
${expr.resultTerm};")
 
         // assignment
-        if (nullCheck) {
+        if (nullCheck && !expr.nullTerm.equals(NEVER_NULL) && 
!expr.nullTerm.equals(ALWAYS_NULL)) {
           reusablePerRecordStatements.add(s"this.${expr.nullTerm} = 
${expr.nullTerm};")
         }
         reusablePerRecordStatements.add(s"this.${expr.resultTerm} = 
${expr.resultTerm};")
diff --git 
a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/batch/table/CalcITCase.scala
 
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/batch/table/CalcITCase.scala
index 71a87d2..0631539 100644
--- 
a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/batch/table/CalcITCase.scala
+++ 
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/batch/table/CalcITCase.scala
@@ -574,6 +574,21 @@ class CalcITCase(
     TestBaseUtils.compareResultAsText(results.asJava, expected)
   }
 
+  @Test
+  def testSplitFieldsOnCustomType(): Unit = {
+    val env = ExecutionEnvironment.getExecutionEnvironment
+    val tEnv = TableEnvironment.getTableEnvironment(env, config)
+    tEnv.getConfig.setMaxGeneratedCodeLength(1)  // splits fields
+
+    val ds = CollectionDataSets.getCustomTypeDataSet(env)
+    val filterDs = ds.toTable(tEnv, 'myInt as 'i, 'myLong as 'l, 'myString as 
's)
+      .filter('s.like("%a%") && 's.charLength() > 12)
+      .select('i, 'l, 's.charLength())
+
+    val expected = "3,3,25\n" + "3,5,14\n"
+    val results = filterDs.toDataSet[Row].collect()
+    TestBaseUtils.compareResultAsText(results.asJava, expected)
+  }
 }
 
 object CalcITCase {