[GitHub] [flink] godfreyhe commented on a change in pull request #13299: [FLINK-19072][table-planner] Import Temporal Table join rule for Stream

[GitBox <gi...@apache.org>]

godfreyhe commented on a change in pull request #13299:
URL: https://github.com/apache/flink/pull/13299#discussion_r504660985



##########
File path: flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUniqueKeysTest.scala
##########
@@ -70,6 +70,16 @@ class FlinkRelMdUniqueKeysTest extends FlinkRelMdHandlerTestBase {
     assertEquals(uniqueKeys(Array(0)), mq.getUniqueKeys(logicalFilter).toSet)
   }
 
+  @Test
+  def testGetUniqueKeysOnWatermark(): Unit = {
+    assertEquals(uniqueKeys(Array(0)), mq.getUniqueKeys(logicalWatermarkAssigner).toSet)
+  }
+
+  @Test
+  def testGetUniqueKeysOnRowtimeDeduplicate(): Unit = {

Review comment:
       move this close to `testGetUniqueKeysOnStreamExecDeduplicate` method

##########
File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/LogicalCorrelateToJoinFromTemporalTableRule.scala
##########
@@ -17,19 +17,32 @@
  */
 package org.apache.flink.table.planner.plan.rules.logical
 
-import org.apache.flink.table.planner.plan.nodes.physical.stream.{StreamExecLookupJoin, StreamExecTemporalJoin}
-
 import org.apache.calcite.plan.RelOptRule.{any, operand}
+import org.apache.calcite.plan.hep.{HepPlanner, HepRelVertex}
 import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelOptRuleOperand}
 import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.logical.{LogicalCorrelate, LogicalFilter, LogicalSnapshot}
-import org.apache.calcite.rex.{RexCorrelVariable, RexFieldAccess, RexInputRef, RexNode, RexShuttle}
+import org.apache.calcite.rel.core.TableScan
+import org.apache.calcite.rel.logical._
+import org.apache.calcite.rex._
+
+import org.apache.flink.table.api.ValidationException
+import org.apache.flink.table.connector.source.LookupTableSource
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory.isRowtimeIndicatorType
+import org.apache.flink.table.planner.plan.nodes.logical.{FlinkLogicalLegacyTableSourceScan, FlinkLogicalTableSourceScan}
+import org.apache.flink.table.planner.plan.nodes.physical.stream.{StreamExecLookupJoin, StreamExecTemporalJoin}
+import org.apache.flink.table.planner.plan.schema.{LegacyTableSourceTable, TableSourceTable, TimeIndicatorRelDataType}
+import org.apache.flink.table.planner.plan.utils.TemporalJoinUtil
+import org.apache.flink.table.sources.LookupableTableSource
+
+import scala.collection.JavaConversions._
+import scala.collection.JavaConverters._
 
 /**
   * The initial temporal table join (FOR SYSTEM_TIME AS OF) is a Correlate, rewrite it into a Join
   * to make join condition can be pushed-down. The join will be translated into
-  * [[StreamExecLookupJoin]] in physical and might be translated into [[StreamExecTemporalJoin]]
-  * in the future.
+  * [[StreamExecLookupJoin]] in physical or translated into [[StreamExecTemporalJoin]].

Review comment:
       The join will be translated into [[StreamExecLookupJoin]] or [[StreamExecTemporalJoin]] in physical.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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