fhueske commented on code in PR #29130:
URL: https://github.com/apache/flink/pull/29130#discussion_r4525448065
##########
flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java:
##########
@@ -1099,6 +1086,43 @@ public Builder partitionBy(String... columnNames) {
return this;
}
+ /**
+ * Configures the changelog mode this table argument's input
carries, as reported by
+ * {@link TableSemantics#changelogMode()}.
+ *
+ * <p>Defaults to {@link ChangelogMode#insertOnly()}. An argument
that doesn't declare
+ * {@code ArgumentTrait.SUPPORT_UPDATES} only accepts {@link
+ * ChangelogMode#insertOnly()}.
+ *
+ * @param mode the changelog mode this argument's input carries
+ */
+ public Builder changelogMode(ChangelogMode mode) {
+ this.changelogMode = checkNotNull(mode, "mode must not be
null");
+ return this;
+ }
+
+ /**
+ * Declares a candidate upsert key for this table argument, i.e. a
set of columns that
+ * uniquely identifies a row in the argument's changelog.
+ *
+ * <p>Call this multiple times to declare more than one candidate
key. Each call adds a
+ * further candidate. This mirrors {@link
TableSemantics#upsertKeyColumns()}, which
+ * reports a list of candidate keys derived from the input's
metadata.
+ *
+ * <p>Required when using upsert input modes ({@link
ChangelogMode#upsert(boolean)})
+ * with a {@code SET_SEMANTIC_TABLE} argument: the argument's
partition columns must
+ * equal one declared candidate exactly. If they equal none, the
PTF receives updates
+ * with UPDATE_BEFORE and full deletes instead, as in {@link
ChangelogMode#all()}.
Review Comment:
```suggestion
* equal one declared candidate exactly. If they equal none, the
harness throws an exception.
```
The current logic throws a `IllegalStateException`.
##########
flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java:
##########
@@ -1099,6 +1086,43 @@ public Builder partitionBy(String... columnNames) {
return this;
}
+ /**
+ * Configures the changelog mode this table argument's input
carries, as reported by
+ * {@link TableSemantics#changelogMode()}.
+ *
+ * <p>Defaults to {@link ChangelogMode#insertOnly()}. An argument
that doesn't declare
+ * {@code ArgumentTrait.SUPPORT_UPDATES} only accepts {@link
+ * ChangelogMode#insertOnly()}.
+ *
+ * @param mode the changelog mode this argument's input carries
+ */
+ public Builder changelogMode(ChangelogMode mode) {
+ this.changelogMode = checkNotNull(mode, "mode must not be
null");
+ return this;
+ }
+
+ /**
+ * Declares a candidate upsert key for this table argument, i.e. a
set of columns that
+ * uniquely identifies a row in the argument's changelog.
+ *
+ * <p>Call this multiple times to declare more than one candidate
key. Each call adds a
+ * further candidate. This mirrors {@link
TableSemantics#upsertKeyColumns()}, which
+ * reports a list of candidate keys derived from the input's
metadata.
+ *
+ * <p>Required when using upsert input modes ({@link
ChangelogMode#upsert(boolean)})
+ * with a {@code SET_SEMANTIC_TABLE} argument: the argument's
partition columns must
+ * equal one declared candidate exactly. If they equal none, the
PTF receives updates
+ * with UPDATE_BEFORE and full deletes instead, as in {@link
ChangelogMode#all()}.
+ *
+ * @param columnNames the columns forming one candidate upsert key
+ */
+ public Builder upsertKey(String... columnNames) {
+ checkNotNull(columnNames, "columnNames must not be null");
+ checkArgument(columnNames.length > 0, "Must specify at least
one column");
+ this.upsertKeys.add(columnNames);
Review Comment:
nit: could do a defensive copy of the Array here.
--
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]
