[FLINK-10763] Interval join produces wrong result type in Scala API

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.6.2
• Fix Version/s: 1.6.3, 1.7.0, 1.8.0
• Component/s: API / DataStream
• Labels: pull-request-available
• Assignee: Timo Walther
• Reporter: jinhai
• Created: 02/Nov/18 14:50
• Updated: 28/Nov/18 17:40
• Resolved: 28/Nov/18 17:40

Description

When stream is a Scala case class, the TypeInformation will fall back to GenericType in the process function which result in bad performance when union another DataStream.

In the union method of DataStream, the type is first checked for equality.

Here is an example:

object Test {

    def main(args: Array[String]): Unit = {
      val env = StreamExecutionEnvironment.getExecutionEnvironment

      val orderA: DataStream[Order] = env.fromCollection(Seq(
        Order(1L, "beer", 3),
         Order(1L, "diaper", 4),
         Order(3L, "rubber", 2)))

      val orderB: DataStream[Order] = env.fromCollection(Seq(
        new Order(2L, "pen", 3),
        new Order(2L, "rubber", 3),
        new Order(4L, "beer", 1)))

      val orderC: DataStream[Order] = orderA.keyBy(_.user)
        .intervalJoin(orderB.keyBy(_.user))
        .between(Time.seconds(0), Time.seconds(0))
        .process(new ProcessJoinFunction[Order, Order, Order] {
          override def processElement(left: Order, right: Order, ctx: ProcessJoinFunction[Order, Order, Order]#Context, out: Collector[Order]): Unit = {
            out.collect(left)
          }})

      println("C: " + orderC.dataType.toString)
      println("B: " + orderB.dataType.toString)

      orderC.union(orderB).print()

      env.execute()
    }

    case class Order(user: Long, product: String, amount: Int)
}

Here is the Exception:

Exception in thread "main" java.lang.IllegalArgumentException: Cannot union streams of different types: GenericType<com.manbuyun.awesome.flink.Test.Order> and com.manbuyun.awesome.flink.Test$Order(user: Long, product: String, amount: Integer)
 at org.apache.flink.streaming.api.datastream.DataStream.union(DataStream.java:219)
 at org.apache.flink.streaming.api.scala.DataStream.union(DataStream.scala:357)
 at com.manbuyun.awesome.flink.Test$.main(Test.scala:38)
 at com.manbuyun.awesome.flink.Test.main(Test.scala)

 

Issue Links

• links to: GitHub Pull Request #7007
• links to: GitHub Pull Request #7141