[FLINK-10642] [table] Fix CodeGen split errors when maxGeneratedCodeLength equals 1

This closes apache#6900.

Author: xueyumusic

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala

@@ -1061,13 +1061,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};")

flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/batch/table/CalcITCase.scala

@@ -578,6 +578,22 @@ class CalcITCase(
     val expected = List("a,a,d,d,e,e", "x,x,z,z,z,z").mkString("\n")
     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 {