[FLINK-6094] [table] Use the lexicographic smallest attribute as the …

…common group id

This closes apache#5273.

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/util/UpdatingPlanChecker.scala

@@ -39,8 +39,13 @@ object UpdatingPlanChecker {
 
  /** Extracts the unique keys of the table produced by the plan. */
  def getUniqueKeyFields(plan: RelNode): Option[Array[String]] = {
+ getUniqueKeyGroups(plan).map(_.map(_._1).toArray)
+ }
+
+ /** Extracts the unique keys and groups of the table produced by the plan. */
+ def getUniqueKeyGroups(plan: RelNode): Option[Seq[(String, String)]] = {
  val keyExtractor = new UniqueKeyExtractor
- keyExtractor.visit(plan).map(_.map(_._1).toArray)
+ keyExtractor.visit(plan)
  }
 
  private class AppendOnlyValidator extends RelVisitor {
@@ -101,7 +106,7 @@ object UpdatingPlanChecker {
  .filter(io => inputKeys.get.map(e => e._1).contains(io._1))
 
  val inputKeysMap = inputKeys.get.toMap
- val inOutGroups = inputKeysAndOutput
+ val inOutGroups = inputKeysAndOutput.sorted.reverse
  .map(e => (inputKeysMap(e._1), e._2))
  .toMap
 

flink-libraries/flink-table/src/test/scala/org/apache/flink/table/plan/UpdatingPlanCheckerTest.scala

@@ -83,6 +83,9 @@ class UpdatingPlanCheckerTest {
  .select('a as 'a1, 'a as 'a2, 'b.count)
 
  util.verifyTableUniqueKey(resultTable, Seq("a1", "a2"))
+ // both a1 and a2 belong to the same group, i.e., a1. We use the lexicographic smallest
+ // attribute as the common group id
+ util.verifyTableKeyGroups(resultTable, Seq(("a1", "a1"), ("a2", "a1")))
  }
 
  @Test
@@ -210,13 +213,25 @@ class UpdatePlanCheckerUtil extends StreamTableTestUtil {
  verifyTableUniqueKey(tableEnv.sql(query), expected)
  }
 
- def verifyTableUniqueKey(resultTable: Table, expected: Seq[String]): Unit = {
+ def getKeyGroups(resultTable: Table): Option[Seq[(String, String)]] = {
  val relNode = resultTable.getRelNode
  val optimized = tableEnv.optimize(relNode, updatesAsRetraction = false)
- val actual = UpdatingPlanChecker.getUniqueKeyFields(optimized)
+ UpdatingPlanChecker.getUniqueKeyGroups(optimized)
+ }
 
+ def verifyTableKeyGroups(resultTable: Table, expected: Seq[(String, String)]): Unit = {
+ val actual = getKeyGroups(resultTable)
+ if (actual.isDefined) {
+ assertEquals(expected.sorted, actual.get.sorted)
+ } else {
+ assertEquals(expected.sorted, Nil)
+ }
+ }
+
+ def verifyTableUniqueKey(resultTable: Table, expected: Seq[String]): Unit = {
+ val actual = getKeyGroups(resultTable).map(_.map(_._1))
  if (actual.isDefined) {
- assertEquals(expected.sorted, actual.get.toSeq.sorted)
+ assertEquals(expected.sorted, actual.get.sorted)
  } else {
  assertEquals(expected.sorted, Nil)
  }