From 3afaf3d7197e8a6cc7a5d2d41ef45a920071dfe7 Mon Sep 17 00:00:00 2001
From: yluan <yluan@xiaohongshu.com>
Date: Tue, 8 Jun 2021 13:59:05 +0800
Subject: [PATCH] try to produce lookup join filter pushdown expensive issue

---
 .../src/test/resources/log4j2-test.properties |  4 +-
 .../src/test/resources/logback-test.xml       |  2 +-
 .../flink/table/api/stream/ExplainTest.scala  |  2 +-
 .../plan/stream/sql/join/LookupJoinTest.scala | 98 ++++++++++++++++++-
 .../planner/plan/utils/lookupFunctions.scala  |  6 ++
 .../table/planner/utils/TableTestBase.scala   |  1 +
 6 files changed, 106 insertions(+), 7 deletions(-)

diff --git a/flink-table/flink-table-planner-blink/src/test/resources/log4j2-test.properties b/flink-table/flink-table-planner-blink/src/test/resources/log4j2-test.properties
index 835c2ec9a3d..1577307cf48 100644
--- a/flink-table/flink-table-planner-blink/src/test/resources/log4j2-test.properties
+++ b/flink-table/flink-table-planner-blink/src/test/resources/log4j2-test.properties
@@ -18,11 +18,11 @@
 
 # Set root logger level to OFF to not flood build logs
 # set manually to INFO for debugging purposes
-rootLogger.level = OFF
+rootLogger.level = INFO
 rootLogger.appenderRef.test.ref = TestLogger
 
 appender.testlogger.name = TestLogger
 appender.testlogger.type = CONSOLE
-appender.testlogger.target = SYSTEM_ERR
+appender.testlogger.target = SYSTEM_OUT
 appender.testlogger.layout.type = PatternLayout
 appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/logback-test.xml b/flink-table/flink-table-planner-blink/src/test/resources/logback-test.xml
index b99489e3d23..4b871189311 100644
--- a/flink-table/flink-table-planner-blink/src/test/resources/logback-test.xml
+++ b/flink-table/flink-table-planner-blink/src/test/resources/logback-test.xml
@@ -23,7 +23,7 @@
         </encoder>
     </appender>
 
-    <root level="WARN">
+    <root level="INFO">
         <appender-ref ref="STDOUT"/>
     </root>
 </configuration>
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/api/stream/ExplainTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/api/stream/ExplainTest.scala
index cadbaa5f445..f1b0b5296d1 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/api/stream/ExplainTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/api/stream/ExplainTest.scala
@@ -116,7 +116,7 @@ class ExplainTest(extended: Boolean) extends TableTestBase {
       "upsertSink2", upsertSink2)
     stmtSet.addInsert("upsertSink2", table2)
 
-    util.verifyExplain(stmtSet, extraDetails: _*)
+    util.verifyExplain(stmtSet, ExplainDetail.JSON_EXECUTION_PLAN)
   }
 
   @Test
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala
index 4d417456b90..c5a5a63a286 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala
@@ -24,17 +24,18 @@ import org.apache.flink.core.testutils.FlinkMatchers.containsMessage
 import org.apache.flink.streaming.api.datastream.DataStream
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
 import org.apache.flink.table.api._
+import org.apache.flink.table.api.internal.TableEnvironmentInternal
 import org.apache.flink.table.data.RowData
 import org.apache.flink.table.descriptors.ConnectorDescriptorValidator.CONNECTOR_TYPE
 import org.apache.flink.table.descriptors.{CustomConnectorDescriptor, DescriptorProperties, Schema}
 import org.apache.flink.table.factories.TableSourceFactory
 import org.apache.flink.table.functions.{AsyncTableFunction, TableFunction, UserDefinedFunction}
-import org.apache.flink.table.planner.plan.utils._
+import org.apache.flink.table.planner.plan.utils.{AsyncTableFunction1, _}
 import org.apache.flink.table.planner.utils.TableTestBase
 import org.apache.flink.table.sources._
 import org.apache.flink.table.types.DataType
+import org.apache.flink.table.types.logical.{BigIntType, IntType, VarCharType}
 import org.apache.flink.table.utils.EncodingUtils
-
 import org.junit.Assert.{assertThat, assertTrue, fail}
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
@@ -44,7 +45,6 @@ import _root_.java.lang.{Boolean => JBoolean}
 import _root_.java.sql.Timestamp
 import _root_.java.util
 import _root_.java.util.{ArrayList => JArrayList, Collection => JCollection, HashMap => JHashMap, List => JList, Map => JMap}
-
 import _root_.scala.collection.JavaConversions._
 
 /**
@@ -55,6 +55,9 @@ import _root_.scala.collection.JavaConversions._
 class LookupJoinTest(legacyTableSource: Boolean) extends TableTestBase with Serializable {
 
   private val util = streamTestUtil()
+  val STRING = new VarCharType(VarCharType.MAX_LENGTH)
+  val LONG = new BigIntType()
+  val INT = new IntType()
 
   @Before
   def before(): Unit ={
@@ -314,6 +317,95 @@ class LookupJoinTest(legacyTableSource: Boolean) extends TableTestBase with Seri
     util.verifyExecPlan(sql)
   }
 
+  @Test
+  def testJoinTemporalTableWithLookupFilterPushDown(): Unit = {
+    createLookupTable("LookupTableAsync1", new AsyncTableFunction1)
+
+    val sql =
+      """
+        |SELECT * FROM MyTable AS T
+        |LEFT JOIN LookupTableAsync1 FOR SYSTEM_TIME AS OF T.proctime AS D
+        |ON T.a = D.id AND D.age = 10
+      """.stripMargin
+
+    util.verifyExecPlan(sql)
+  }
+
+  /**
+   * testcase with 2 insert
+   * lookup with 2 filter condition
+   */
+  @Test
+  def testJoinTemporalTableWithMultisinkWithFilterPushDown1(): Unit = {
+    createLookupTable("LookupTableAsync1", new AsyncTableFunction1)
+
+    util.tableEnv.createTemporaryView("v_vvv", util.tableEnv.sqlQuery(
+      """
+        |SELECT * FROM MyTable AS T
+        |JOIN LookupTableAsync1 FOR SYSTEM_TIME AS OF T.proctime AS D
+        |ON T.a = D.id
+        |""".stripMargin))
+
+    val stmtSet = util.tableEnv.createStatementSet()
+
+    val appendSink1 = util.createRetractTableSink(
+      Array("a", "b", "id", "name"),
+      Array(INT, STRING, INT, STRING))
+    util.tableEnv.asInstanceOf[TableEnvironmentInternal].registerTableSinkInternal(
+      "appendSink1", appendSink1)
+    stmtSet.addInsert("appendSink1", util.tableEnv.sqlQuery(
+      """
+        |SELECT a,b,id,name FROM v_vvv
+        |WHERE age = 10
+        |""".stripMargin))
+
+    util.verifyExecPlan(stmtSet)
+    // util.verifyExplain(stmtSet)
+  }
+
+  /**
+   * testcase with 1 insert
+   * lookup with 2 filter condition
+   */
+  @Test
+  def testJoinTemporalTableWithMultisinkWithFilterPushDown2(): Unit = {
+    createLookupTable("LookupTableAsync1", new AsyncTableFunction1)
+
+    util.tableEnv.createTemporaryView("v_vvv", util.tableEnv.sqlQuery(
+      """
+        |SELECT * FROM MyTable AS T
+        |JOIN LookupTableAsync1 FOR SYSTEM_TIME AS OF T.proctime AS D
+        |ON T.a = D.id
+        |""".stripMargin))
+
+    val stmtSet = util.tableEnv.createStatementSet()
+
+    val appendSink1 = util.createRetractTableSink(
+      Array("a", "b", "id", "name"),
+      Array(INT, STRING, INT, STRING))
+    util.tableEnv.asInstanceOf[TableEnvironmentInternal].registerTableSinkInternal(
+      "appendSink1", appendSink1)
+    stmtSet.addInsert("appendSink1", util.tableEnv.sqlQuery(
+      """
+        |SELECT a,b,id,name FROM v_vvv
+        |WHERE age = 10
+        |""".stripMargin))
+
+    val appendSink2 = util.createRetractTableSink(
+      Array("a", "b", "id", "name"),
+      Array(INT, STRING, INT, STRING))
+    util.tableEnv.asInstanceOf[TableEnvironmentInternal].registerTableSinkInternal(
+      "appendSink2", appendSink2)
+    stmtSet.addInsert("appendSink2", util.tableEnv.sqlQuery(
+      """
+        |SELECT a,b,id,name FROM v_vvv
+        |WHERE age = 30
+        |""".stripMargin))
+
+    // util.verifyExecPlan(stmtSet)
+    util.verifyExplain(stmtSet)
+  }
+
   @Test
   def testJoinTemporalTableWithCalcPushDown(): Unit = {
     val sql =
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/utils/lookupFunctions.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/utils/lookupFunctions.scala
index 89d3cb025b2..cf5c62b3ddb 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/utils/lookupFunctions.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/utils/lookupFunctions.scala
@@ -57,6 +57,12 @@ class TableFunctionWithRow extends TableFunction[Row] {
   }
 }
 
+@SerialVersionUID(1L)
+class AsyncTableFunction1 extends AsyncTableFunction[RowData] {
+  def eval(resultFuture: CompletableFuture[JCollection[RowData]], a: Integer): Unit = {
+  }
+}
+
 @SerialVersionUID(1L)
 class InvalidAsyncTableFunctionEvalSignature1 extends AsyncTableFunction[RowData] {
   def eval(a: Integer, b: StringData, c: LocalDateTime): Unit = {
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
index 14b696973db..bb0d72eebd0 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
@@ -917,6 +917,7 @@ abstract class TableTestUtilBase(test: TableTestBase, isStreamingMode: Boolean)
   }
 
   private def doVerifyExplain(explainResult: String, needReplaceEstimatedCost: Boolean): Unit = {
+    println(explainResult)
     val actual = if (needReplaceEstimatedCost) {
       replaceEstimatedCost(explainResult)
     } else {
-- 
2.23.0

