[jira] [Commented] (FLINK-10670) Fix Correlate codegen error

["ASF GitHub Bot (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-10670?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16691709#comment-16691709 ] 

ASF GitHub Bot commented on FLINK-10670:
----------------------------------------

asfgit closed pull request #6923: [FLINK-10670] [table] Fix Correlate codegen error
URL: https://github.com/apache/flink/pull/6923
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

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 2816f116ffd..ad36aa6e6ec 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
@@ -109,39 +109,39 @@ abstract class CodeGenerator(
 
   // set of member statements that will be added only once
   // we use a LinkedHashSet to keep the insertion order
-  protected val reusableMemberStatements: mutable.LinkedHashSet[String] =
+  val reusableMemberStatements: mutable.LinkedHashSet[String] =
     mutable.LinkedHashSet[String]()
 
   // set of constructor statements that will be added only once
   // we use a LinkedHashSet to keep the insertion order
-  protected val reusableInitStatements: mutable.LinkedHashSet[String] =
+  val reusableInitStatements: mutable.LinkedHashSet[String] =
     mutable.LinkedHashSet[String]()
 
   // set of open statements for RichFunction that will be added only once
   // we use a LinkedHashSet to keep the insertion order
-  protected val reusableOpenStatements: mutable.LinkedHashSet[String] =
+  val reusableOpenStatements: mutable.LinkedHashSet[String] =
     mutable.LinkedHashSet[String]()
 
   // set of close statements for RichFunction that will be added only once
   // we use a LinkedHashSet to keep the insertion order
-  protected val reusableCloseStatements: mutable.LinkedHashSet[String] =
+  val reusableCloseStatements: mutable.LinkedHashSet[String] =
     mutable.LinkedHashSet[String]()
 
   // set of statements that will be added only once per record;
   // code should only update member variables because local variables are not accessible if
   // the code needs to be split;
   // we use a LinkedHashSet to keep the insertion order
-  protected val reusablePerRecordStatements: mutable.LinkedHashSet[String] =
+  val reusablePerRecordStatements: mutable.LinkedHashSet[String] =
     mutable.LinkedHashSet[String]()
 
   // map of initial input unboxing expressions that will be added only once
   // (inputTerm, index) -> expr
-  protected val reusableInputUnboxingExprs: mutable.Map[(String, Int), GeneratedExpression] =
+  val reusableInputUnboxingExprs: mutable.Map[(String, Int), GeneratedExpression] =
     mutable.Map[(String, Int), GeneratedExpression]()
 
   // set of constructor statements that will be added only once
   // we use a LinkedHashSet to keep the insertion order
-  protected val reusableConstructorStatements: mutable.LinkedHashSet[(String, String)] =
+  val reusableConstructorStatements: mutable.LinkedHashSet[(String, String)] =
     mutable.LinkedHashSet[(String, String)]()
 
   /**
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CollectorCodeGenerator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CollectorCodeGenerator.scala
index 85d858fb75b..bd225c60b6b 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CollectorCodeGenerator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CollectorCodeGenerator.scala
@@ -86,14 +86,24 @@ class CollectorCodeGenerator(
       s"$input2TypeClass $input2Term" // local variable
     }
 
+    val memberCodes = (reusableMemberStatements ++ codeGenerator.reusableMemberStatements)
+      .mkString("", "\n", "\n")
+
+    val initCodes = (reusableInitStatements ++ codeGenerator.reusableInitStatements)
+      .mkString("", "\n", "\n")
+
+    val reusePerRecordCodes =
+      (reusablePerRecordStatements ++ codeGenerator.reusablePerRecordStatements)
+      .mkString("", "\n", "\n")
+
     val funcCode = j"""
       |public class $className extends ${classOf[TableFunctionCollector[_]].getCanonicalName} {
       |
       |  $recordMember
-      |  ${reuseMemberCode()}
+      |  ${memberCodes}
       |
       |  public $className() throws Exception {
-      |    ${reuseInitCode()}
+      |    ${initCodes}
       |  }
       |
       |  @Override
@@ -107,7 +117,7 @@ class CollectorCodeGenerator(
       |    $input1TypeClass $input1Term = ($input1TypeClass) getInput();
       |    $recordAssignment = ($input2TypeClass) record;
       |    ${reuseInputUnboxingCode()}
-      |    ${reusePerRecordCode()}
+      |    ${reusePerRecordCodes}
       |    $bodyCode
       |  }
       |
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCorrelate.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCorrelate.scala
index 3475e1901e9..18e1e5c5011 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCorrelate.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCorrelate.scala
@@ -150,17 +150,9 @@ trait CommonCorrelate {
          |""".stripMargin
     } else {
 
-      // adjust indices of InputRefs to adhere to schema expected by generator
-      val changeInputRefIndexShuttle = new RexShuttle {
-        override def visitInputRef(inputRef: RexInputRef): RexNode = {
-          new RexInputRef(inputSchema.arity + inputRef.getIndex, inputRef.getType)
-        }
-      }
-      // Run generateExpression to add init statements (ScalarFunctions) of condition to generator.
-      //   The generated expression is discarded.
-      generator.generateExpression(condition.get.accept(changeInputRefIndexShuttle))
-
       filterGenerator.input1Term = filterGenerator.input2Term
+      // filterGenerator run generateExpression to add init statements and member fields
+      // then combine with existing statements in generator
       val filterCondition = filterGenerator.generateExpression(condition.get)
       s"""
          |${filterGenerator.reuseInputUnboxingCode()}
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/table/CorrelateITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/table/CorrelateITCase.scala
index f4b1f805974..ef9787e6ed5 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/table/CorrelateITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/table/CorrelateITCase.scala
@@ -23,6 +23,7 @@ import org.apache.flink.api.scala._
 import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
 import org.apache.flink.table.api.scala._
 import org.apache.flink.table.api.{TableEnvironment, Types, ValidationException}
+import org.apache.flink.table.expressions.{CurrentTimestamp, DateFormat}
 import org.apache.flink.table.expressions.utils.{Func18, Func20, RichFunc2}
 import org.apache.flink.table.runtime.utils.{StreamITCase, StreamTestData, _}
 import org.apache.flink.table.utils._
@@ -284,6 +285,27 @@ class CorrelateITCase extends AbstractTestBase {
     )
   }
 
+  @Test
+  def testTableFunctionCollectorInit(): Unit = {
+    val t = testData(env).toTable(tEnv).as('a, 'b, 'c)
+    val func0 = new TableFunc0
+
+    // this case will generate 'timestamp' member field and 'DateFormatter'
+    val result = t
+      .join(func0('c) as('d, 'e))
+      .where(DateFormat(CurrentTimestamp(), "yyyyMMdd") === 'd)
+      .select('c, 'd, 'e)
+      .toAppendStream[Row]
+
+    result.addSink(new StreamITCase.StringSink[Row])
+    env.execute()
+
+    assertEquals(
+      Seq(),
+      StreamITCase.testResults.sorted
+    )
+  }
+
   private def testData(
       env: StreamExecutionEnvironment)
     : DataStream[(Int, Long, String)] = {


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


> Fix Correlate codegen error
> ---------------------------
>
>                 Key: FLINK-10670
>                 URL: https://issues.apache.org/jira/browse/FLINK-10670
>             Project: Flink
>          Issue Type: Bug
>          Components: Table API & SQL
>            Reporter: Ruidong Li
>            Assignee: Ruidong Li
>            Priority: Major
>              Labels: pull-request-available
>
> TableFunctionCollector should handle reuseInitCode and reuseMemberCode



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)