[FLINK-7596] fix bug when Set Operation handles ANY type

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.3.4, 1.4.0
• Component/s: Table SQL / API
• Assignee: Ruidong Li
• Reporter: Ruidong Li
• Created: 07/Sep/17 01:50
• Updated: 12/Mar/18 15:39
• Resolved: 12/Mar/18 15:39

Description

If two inputs with Any(GenericRelDataType), when they comes to Set Operation(UNION, MINUS,...), it will cause a TableException with info is "Type is not supported: ANY"
Here is the test case:

@Test
  def testUnion(): Unit = {
    val list = List((1, new NODE), (2, new NODE))
    val list2 = List((3, new NODE), (4, new NODE))
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val tEnv = TableEnvironment.getTableEnvironment(env)
    val s1 = tEnv.fromDataStream(env.fromCollection(list))
    val s2 = tEnv.fromDataStream(env.fromCollection(list2))
    val result = s1.unionAll(s2).toAppendStream[Row]
    result.addSink(new StreamITCase.StringSink[Row])
    env.execute()
  }

  class NODE {
  val x = new util.HashMap[String, String]()
}

This bug happens because Flink doesn't handle createSqlType(ANY) and Calcite doesn't know the differences between ANY and ANY(GenericRelDataType), so the createSqlType(ANY) of Calcite will return a BasicSqlType instead.