autophagy commented on code in PR #29130:
URL: https://github.com/apache/flink/pull/29130#discussion_r4452952150
##########
docs/content.zh/docs/dev/table/functions/ptfs.md:
##########
@@ -2620,6 +2620,100 @@ void testTimerWithState() throws Exception {
{{< /tab >}}
{{< /tabs >}}
+#### Testing Updating (Changelog) Tables
+
+A table argument that declares `ArgumentTrait.SUPPORT_UPDATES` receives an
updating changelog rather
+than an insert-only stream. You can configure the changelog mode of a
particular table argument
+via `.changelogMode()` on `TableArgument.Builder`.
+
+When processing elements for a table argument with a specific changelog mode,
the harness will
+reject any row kind that the mode does not contain.
+
+{{< tabs "changelog-input" >}}
+{{< tab "Java" >}}
+```java
+// A row-semantic PTF that accepts an updating input and echoes each row's
kind and value.
+@DataTypeHint("ROW<seen STRING>")
+public class UpdatingConsumerPTF extends ProcessTableFunction<Row> {
+ public void eval(
+ @ArgumentHint({ArgumentTrait.ROW_SEMANTIC_TABLE,
ArgumentTrait.SUPPORT_UPDATES}) Row input) {
+ collect(Row.of(input.getKind() + ":" + input.getFieldAs("value")));
+ }
+}
+
+@Test
+void testUpdatingInput() throws Exception {
+ try (ProcessTableFunctionTestHarness<Row> harness =
+ ProcessTableFunctionTestHarness.ofClass(UpdatingConsumerPTF.class)
+ .withTableArgument(
+ TableArgument.forName("input")
+ .type(DataTypes.of("ROW<value INT>"))
+ .changelogMode(ChangelogMode.all())
+ .build())
+ .build()) {
+
+ harness.processElement(RowKind.INSERT, 10);
+ harness.processElement(RowKind.UPDATE_AFTER, 20);
+ harness.processElement(RowKind.DELETE, 10);
+
+ List<Row> output = harness.getFunctionOutput();
+ assertThat(output).containsExactly(
+ Row.of("INSERT:10"), Row.of("UPDATE_AFTER:20"), Row.of("DELETE:10"));
+ }
+}
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+**Upsert Input**: Upsert changelog modes (`ChangelogMode.upsert(...)`) apply
to `SET_SEMANTIC_TABLE`
+arguments only. You can configure the upsert key with `.upsertKey()` on
`TableArgument.Builder`.
+Call it more than once to declare additional candidate keys. The partition
columns must match one
+of them.
+
+{{< tabs "changelog-upsert" >}}
+{{< tab "Java" >}}
+```java
+@DataTypeHint("ROW<v INT>")
+public class UpsertConsumerPTF extends ProcessTableFunction<Row> {
Review Comment:
It consumes upsert rows but emits insert only (`Row.of` without specifying
the row kind defaults to insert)
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
