[FLINK-18632][table-planner-blink] Assign the missing RowKind when to…

…RetractStream with POJO type Co-authored-by: luoziyu <[email protected]> This closes apache#12955
spotify · Jul 24, 2020 · fc36235 · fc36235
1 parent ede4da5
commit fc36235
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 0 deletions.
diff --git a/...anner-blink/src/main/scala/org/apache/flink/table/planner/codegen/SinkCodeGenerator.scala b/...anner-blink/src/main/scala/org/apache/flink/table/planner/codegen/SinkCodeGenerator.scala
@@ -94,6 +94,7 @@ object SinkCodeGenerator {
  afterIndexModify = CodeGenUtils.newName("afterIndexModify")
  s"""
  |${conversion.code}
+ |${conversion.resultTerm}.setRowKind(${inputTerm}.getRowKind());
  |${classOf[RowData].getCanonicalName} $afterIndexModify = ${conversion.resultTerm};
  |""".stripMargin
  case _ =>

diff --git a/...cala/org/apache/flink/table/planner/runtime/stream/sql/StreamTableEnvironmentITCase.scala b/...cala/org/apache/flink/table/planner/runtime/stream/sql/StreamTableEnvironmentITCase.scala
@@ -114,4 +114,27 @@ class StreamTableEnvironmentITCase extends StreamingTestBase {
  "(true,Person{name='Jack', age=3})")
  assertEquals(expected.sorted, sink.getResults.sorted)
  }
+
+ @Test
+ def testRetractMsgWithPojoType(): Unit = {
+ val orders = env.fromCollection(Seq(
+ new Order(1L, new ProductItem("beer", 10L), 1),
+ new Order(1L, new ProductItem("beer", 10L), 2)
+ ))
+
+ val table = tEnv.fromDataStream(orders, 'user, 'product, 'amount)
+
+ val sink = new StringSink[(Boolean, Order)]()
+ tEnv.sqlQuery(s"""|SELECT user, product, sum(amount) as amount
+ |FROM $table
+ |GROUP BY user, product
+ |""".stripMargin).toRetractStream[Order].addSink(sink)
+ env.execute()
+
+ val expected = List(
+ "(true,Order{user=1, product='Product{name='beer', id=10}', amount=1})",
+ "(false,Order{user=1, product='Product{name='beer', id=10}', amount=1})",
+ "(true,Order{user=1, product='Product{name='beer', id=10}', amount=3})")
+ assertEquals(expected.sorted, sink.getResults.sorted)
+ }
 }