diff --git a/docs/content.zh/docs/dev/table/functions/ptfs.md b/docs/content.zh/docs/dev/table/functions/ptfs.md index 5e7289e2ee7d8..2542f9f289376 100644 --- a/docs/content.zh/docs/dev/table/functions/ptfs.md +++ b/docs/content.zh/docs/dev/table/functions/ptfs.md @@ -2620,6 +2620,97 @@ 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()` `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") +public class UpdatingConsumerPTF extends ProcessTableFunction { + 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 harness = + ProcessTableFunctionTestHarness.ofClass(UpdatingConsumerPTF.class) + .withTableArgument( + TableArgument.forName("input") + .type(DataTypes.of("ROW")) + .changelogMode(ChangelogMode.all()) + .build()) + .build()) { + + harness.processElement(RowKind.INSERT, 10); + harness.processElement(RowKind.UPDATE_AFTER, 20); + harness.processElement(RowKind.DELETE, 10); + + List 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") +public class UpsertConsumerPTF extends ProcessTableFunction { + public void eval( + @ArgumentHint({ArgumentTrait.SET_SEMANTIC_TABLE, ArgumentTrait.SUPPORT_UPDATES}) Row input) { + collect(Row.of((Integer) input.getFieldAs("value"))); + } +} + +@Test +void testUpsertInput() throws Exception { + try (ProcessTableFunctionTestHarness harness = + ProcessTableFunctionTestHarness.ofClass(UpsertConsumerPTF.class) + .withTableArgument( + TableArgument.forName("input") + .type(DataTypes.of("ROW")) + .partitionBy("key") + .upsertKey("key") + .changelogMode(ChangelogMode.upsert(true)) + .build()) + .build()) { + + harness.processElement(RowKind.INSERT, "A", 1); + harness.processElement(RowKind.UPDATE_AFTER, "A", 2); + + List output = harness.getFunctionOutput(); + assertThat(output).containsExactly(Row.of(1), Row.of(2)); + } +} +``` +{{< /tab >}} +{{< /tabs >}} + +**Output Changelog Mode**: The harness validates collected rows against the PTF's resolved output +changelog mode (derived from `ChangelogFunction`, or insert-only otherwise) and rejects any row +whose kind that mode does not contain. + +{{< top >}} + #### Optional Partitioning For PTFs with `OPTIONAL_PARTITION_BY`, you can omit `.partitionBy(...)` on `TableArgument.Builder` @@ -2797,5 +2888,4 @@ void testAtomicOutputFunctionOutput() throws Exception { ### PTF Features Unsupported by the TestHarness -- Update traits (`SUPPORTS_UPDATES`, `REQUIRE_UPDATE_BEFORE`) - State TTL (state is supported but TTL expiration is not yet implemented) diff --git a/docs/content/docs/dev/table/functions/ptfs.md b/docs/content/docs/dev/table/functions/ptfs.md index 0939d1fdfb8f4..e0e65c8e04501 100644 --- a/docs/content/docs/dev/table/functions/ptfs.md +++ b/docs/content/docs/dev/table/functions/ptfs.md @@ -2623,6 +2623,97 @@ 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()` `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") +public class UpdatingConsumerPTF extends ProcessTableFunction { + 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 harness = + ProcessTableFunctionTestHarness.ofClass(UpdatingConsumerPTF.class) + .withTableArgument( + TableArgument.forName("input") + .type(DataTypes.of("ROW")) + .changelogMode(ChangelogMode.all()) + .build()) + .build()) { + + harness.processElement(RowKind.INSERT, 10); + harness.processElement(RowKind.UPDATE_AFTER, 20); + harness.processElement(RowKind.DELETE, 10); + + List 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") +public class UpsertConsumerPTF extends ProcessTableFunction { + public void eval( + @ArgumentHint({ArgumentTrait.SET_SEMANTIC_TABLE, ArgumentTrait.SUPPORT_UPDATES}) Row input) { + collect(Row.of((Integer) input.getFieldAs("value"))); + } +} + +@Test +void testUpsertInput() throws Exception { + try (ProcessTableFunctionTestHarness harness = + ProcessTableFunctionTestHarness.ofClass(UpsertConsumerPTF.class) + .withTableArgument( + TableArgument.forName("input") + .type(DataTypes.of("ROW")) + .partitionBy("key") + .upsertKey("key") + .changelogMode(ChangelogMode.upsert(true)) + .build()) + .build()) { + + harness.processElement(RowKind.INSERT, "A", 1); + harness.processElement(RowKind.UPDATE_AFTER, "A", 2); + + List output = harness.getFunctionOutput(); + assertThat(output).containsExactly(Row.of(1), Row.of(2)); + } +} +``` +{{< /tab >}} +{{< /tabs >}} + +**Output Changelog Mode**: The harness validates collected rows against the PTF's resolved output +changelog mode (derived from `ChangelogFunction`, or insert-only otherwise) and rejects any row +whose kind that mode does not contain. + +{{< top >}} + #### Optional Partitioning For PTFs with `OPTIONAL_PARTITION_BY`, you can omit `.partitionBy(...)` on `TableArgument.Builder` @@ -2800,5 +2891,4 @@ void testAtomicOutputFunctionOutput() throws Exception { ### PTF Features Unsupported by the TestHarness -- Update traits (`SUPPORTS_UPDATES`, `REQUIRE_UPDATE_BEFORE`) - State TTL (state is supported but TTL expiration is not yet implemented) diff --git a/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java b/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java index 42034b08b6986..21544e1361057 100644 --- a/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java +++ b/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java @@ -29,6 +29,7 @@ import org.apache.flink.table.data.RowData; import org.apache.flink.table.data.conversion.DataStructureConverter; import org.apache.flink.table.data.conversion.DataStructureConverters; +import org.apache.flink.table.functions.ChangelogFunction; import org.apache.flink.table.functions.FunctionContext; import org.apache.flink.table.functions.FunctionKind; import org.apache.flink.table.functions.ProcessTableFunction; @@ -152,6 +153,8 @@ private static class TableArgumentConverters { @Nullable private final Class rowtimeConversionClass; + private final ChangelogMode outputChangelogMode; + @Nullable private InvocationContext currentInvocation; private ProcessTableFunctionTestHarness( @@ -165,7 +168,8 @@ private ProcessTableFunctionTestHarness( DataType ptfOutputType, TestHarnessStateManager stateManager, TestHarnessTimerManager timerManager, - @Nullable String onTimeColumnName) + @Nullable String onTimeColumnName, + ChangelogMode outputChangelogMode) throws Exception { this.function = function; this.functionContext = functionContext; @@ -199,6 +203,7 @@ private ProcessTableFunctionTestHarness( this.stateManager = stateManager; this.timerManager = timerManager; this.onTimeColumnName = onTimeColumnName; + this.outputChangelogMode = outputChangelogMode; this.functionOutput = new ArrayList<>(); this.output = new ArrayList<>(); this.collector = new HarnessCollector(); @@ -564,13 +569,11 @@ public TableSemantics tableSemanticsFor(String argName) { argName, tableArgNames)); } TableArgumentInfo tableArg = (TableArgumentInfo) argInfo; - int[] partitionIndices = getPartitionColumnIndices(tableArg); int timeColumnIndex = onTimeColumnName != null ? getFieldNames(tableArg.dataType).indexOf(onTimeColumnName) : -1; - return new TestHarnessTableSemantics( - tableArg.dataType, partitionIndices, timeColumnIndex); + return buildTableSemantics(tableArg, timeColumnIndex); } @Override @@ -596,7 +599,7 @@ public void clearAll() { @Override public ChangelogMode getChangelogMode() { - return ChangelogMode.insertOnly(); + return outputChangelogMode; } } @@ -693,8 +696,8 @@ private void checkTimersEnabled() { ArgumentInfo.filterTableArguments(arguments).stream() .anyMatch( t -> - t.isSetSemantic - && t.prependStrategy + t.isSetSemantic() + && t.prependStrategy() != OutputPrependStrategy.ALL_COLUMNS); if (!enabled) { throw new TableRuntimeException( @@ -744,16 +747,24 @@ private static int[] getPartitionColumnIndices(TableArgumentInfo arg) { if (arg.partitionColumnNames == null || arg.partitionColumnNames.length == 0) { return new int[0]; } + return resolveColumnNamesToIndices(arg, arg.partitionColumnNames, "Partition"); + } + + private static int[] resolveColumnNamesToIndices( + TableArgumentInfo arg, String[] columnNames, String kind) { List fieldNames = getFieldNames(arg.dataType); - int[] indices = new int[arg.partitionColumnNames.length]; - for (int i = 0; i < arg.partitionColumnNames.length; i++) { - String colName = arg.partitionColumnNames[i]; + int[] indices = new int[columnNames.length]; + for (int i = 0; i < columnNames.length; i++) { + String colName = columnNames[i]; int index = fieldNames.indexOf(colName); if (index < 0) { throw new IllegalStateException( - "Partition column '" + kind + + " column '" + colName - + "' not found in table argument. " + + "' not found in table argument '" + + arg.name + + "'. " + "Available fields: " + fieldNames); } @@ -762,6 +773,30 @@ private static int[] getPartitionColumnIndices(TableArgumentInfo arg) { return indices; } + static TestHarnessTableSemantics buildTableSemantics( + TableArgumentInfo tableArg, int timeColumnIndex) { + int[] partitionIndices = getPartitionColumnIndices(tableArg); + ChangelogMode mode = tableArg.effectiveChangelogMode(); + List upsertKeyIndices = new ArrayList<>(); + for (String[] candidate : tableArg.upsertKeys) { + upsertKeyIndices.add(resolveColumnNamesToIndices(tableArg, candidate, "Upsert key")); + } + + return new TestHarnessTableSemantics( + tableArg.dataType, partitionIndices, upsertKeyIndices, timeColumnIndex, mode); + } + + static TestHarnessTableSemantics buildTableSemanticsForInference( + TableArgumentInfo tableArg, int timeColumnIndex) { + int[] partitionIndices = getPartitionColumnIndices(tableArg); + return new TestHarnessTableSemantics( + tableArg.dataType, + partitionIndices, + Collections.emptyList(), + timeColumnIndex, + null); + } + @Nullable private Class resolveRowtimeConversionClass(List tableArguments) { if (onTimeColumnName == null) { @@ -800,6 +835,8 @@ private static T convertFromMillis(long millis, Class targetClass) { } private void invokeEval(TableArgumentInfo activeTableArg, Row activeRow) throws Exception { + activeRow = prepareInputRow(activeTableArg, activeRow); + TableArgumentConverters converters = argumentConverters.get(activeTableArg.name); RowData rowData = (RowData) converters.toNamedRow.toInternal(activeRow); @@ -859,6 +896,82 @@ private Row extractPartitionKey(TableArgumentInfo tableArg, Row row) { return Row.of(keyValues); } + /** + * Applies the changelog-mode rules a row must satisfy as it crosses the input boundary: the row + * kind is validated against the argument's consumed mode, then a key-only delete is reshaped to + * carry only its key columns. Both steps belong to the same boundary and must run in this + * order. + */ + private Row prepareInputRow(TableArgumentInfo tableArg, Row row) { + validateInputRowKind(tableArg, row.getKind()); + return stripNonKeyFieldsForKeyOnlyDelete(tableArg, row); + } + + private void validateInputRowKind(TableArgumentInfo tableArg, RowKind rowKind) { + if (rowKind == RowKind.INSERT) { + return; + } + + if (!tableArg.is(StaticArgumentTrait.SUPPORT_UPDATES)) { + throw new IllegalArgumentException( + String.format( + "Row kind %s is not permitted on table argument '%s'. " + + "This argument does not declare SUPPORT_UPDATES.", + rowKind, tableArg.name)); + } + + ChangelogMode mode = tableArg.effectiveChangelogMode(); + if (!mode.contains(rowKind)) { + throw new IllegalArgumentException( + String.format( + "Row kind %s is not permitted on table argument '%s'. " + + "Expected consumed changelog mode: %s", + rowKind, tableArg.name, mode)); + } + } + + private Row stripNonKeyFieldsForKeyOnlyDelete(TableArgumentInfo tableArg, Row row) { + if (row.getKind() != RowKind.DELETE) { + return row; + } + + ChangelogMode mode = tableArg.effectiveChangelogMode(); + if (!mode.keyOnlyDeletes()) { + return row; + } + + // A key-only delete carries only the columns the stream is co-partitioned by, so only the + // partition columns survive here. + String[] keyColumns = tableArg.partitionColumnNames; + if (keyColumns == null || keyColumns.length == 0) { + return row; + } + + Set keyIndices = + Arrays.stream(resolveColumnNamesToIndices(tableArg, keyColumns, "Partition key")) + .boxed() + .collect(Collectors.toSet()); + + // A freshly constructed Row initializes all fields to null, so only the key columns are + // copied back; every other field is already stripped. + Row result = new Row(row.getKind(), row.getArity()); + for (int i = 0; i < row.getArity(); i++) { + if (keyIndices.contains(i)) { + result.setField(i, row.getField(i)); + } + } + return result; + } + + private void validateOutputRowKind(RowKind rowKind) { + if (!outputChangelogMode.contains(rowKind)) { + throw new TableRuntimeException( + String.format( + "Invalid row kind received: %s. Expected produced changelog mode: %s", + rowKind, outputChangelogMode)); + } + } + /** Collector implementation that stores output in the harness. */ private class HarnessCollector implements Collector { @@ -867,6 +980,8 @@ private class HarnessCollector implements Collector { public void collect(OUT record) { Row ptfRow = toPtfOutputRow(record); + validateOutputRowKind(ptfRow.getKind()); + functionOutput.add(outputKind == OutputKind.ROW ? (OUT) ptfRow : record); Row finalRecord; @@ -913,7 +1028,7 @@ private OutputPrependStrategy resolvePrependStrategy() { if (ctx.isEvalInvocation()) { ArgumentInfo argInfo = argumentsByName.get(ctx.tableArgumentName); if (argInfo instanceof TableArgumentInfo) { - return ((TableArgumentInfo) argInfo).prependStrategy; + return ((TableArgumentInfo) argInfo).prependStrategy(); } } return OutputPrependStrategy.NONE; @@ -943,7 +1058,7 @@ private Row prependPartitionKeys(Row ptfRow) { for (ArgumentInfo arg : arguments) { if (arg instanceof TableArgumentInfo) { TableArgumentInfo tableArg = (TableArgumentInfo) arg; - if (tableArg.isSetSemantic && tableArg.partitionColumnNames != null) { + if (tableArg.isPartitioned()) { totalPartitionKeyCount += tableArg.partitionColumnNames.length; } } @@ -958,7 +1073,7 @@ private Row prependPartitionKeys(Row ptfRow) { for (ArgumentInfo arg : arguments) { if (arg instanceof TableArgumentInfo) { TableArgumentInfo tableArg = (TableArgumentInfo) arg; - if (tableArg.isSetSemantic && tableArg.partitionColumnNames != null) { + if (tableArg.isPartitioned()) { for (int i = 0; i < tableArg.partitionColumnNames.length; i++) { result.setField(resultIndex++, partitionKey.getField(i)); } @@ -1006,7 +1121,7 @@ private static Row appendField(Row row, Object value) { } /** Extracts field names from RowType or StructuredType. */ - private static List getFieldNames(DataType dataType) { + static List getFieldNames(DataType dataType) { LogicalType logicalType = dataType.getLogicalType(); if (logicalType instanceof RowType) { return ((RowType) logicalType) @@ -1048,11 +1163,15 @@ public static final class TableArgument { private final String name; @Nullable private final AbstractDataType type; @Nullable private final String[] partitionColumns; + @Nullable private final ChangelogMode changelogMode; + private final List upsertKeys; private TableArgument(Builder builder) { this.name = builder.name; this.type = builder.type; this.partitionColumns = builder.partitionColumns; + this.changelogMode = builder.changelogMode; + this.upsertKeys = builder.upsertKeys; } public static Builder forName(String argumentName) { @@ -1064,6 +1183,8 @@ public static final class Builder { private final String name; @Nullable private AbstractDataType type; @Nullable private String[] partitionColumns; + @Nullable private ChangelogMode changelogMode; + private final List upsertKeys = new ArrayList<>(); private Builder(String argumentName) { this.name = checkNotNull(argumentName, "argumentName must not be null"); @@ -1098,6 +1219,42 @@ public Builder partitionBy(String... columnNames) { return this; } + /** + * Configures the changelog mode this table argument's input carries, as reported by + * {@link TableSemantics#changelogMode()}. + * + *

Required for an argument that declares {@code ArgumentTrait.SUPPORT_UPDATES}. A + * non-updating argument always reports {@link ChangelogMode#insertOnly()}, so + * configuring one has no effect. + * + * @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. + * + *

Call multiple times to declare multiple candidate keys; each call adds one more + * candidate. This mirrors {@link TableSemantics#upsertKeyColumns()}, which reports a + * list of candidate keys derived from the input's metadata. + * + *

Required when using upsert input modes ({@link ChangelogMode#upsert(boolean)}) + * with a {@code SET_SEMANTIC_TABLE} argument: one of the declared candidates must match + * the argument's partition columns. + * + * @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); + return this; + } + public TableArgument build() { return new TableArgument(this); } @@ -1117,6 +1274,8 @@ public static class Builder { private final Map tableArgs = new HashMap<>(); private final Map partitionConfigs = new HashMap<>(); private final Map stateArgs = new HashMap<>(); + private final Map tableArgumentChangelogModes = new HashMap<>(); + private final Map> tableArgumentUpsertKeys = new HashMap<>(); @Nullable private String onTimeColumnName = null; private Builder(Class> functionClass) { @@ -1150,6 +1309,12 @@ public Builder withTableArgument(TableArgument tableArgument) { tableArgument.name, new PartitionConfiguration(tableArgument.partitionColumns)); } + if (tableArgument.changelogMode != null) { + tableArgumentChangelogModes.put(tableArgument.name, tableArgument.changelogMode); + } + if (!tableArgument.upsertKeys.isEmpty()) { + tableArgumentUpsertKeys.put(tableArgument.name, tableArgument.upsertKeys); + } return this; } @@ -1243,6 +1408,17 @@ public ProcessTableFunctionTestHarness build() throws Exception { validatePartitionConsistency(arguments); validateInitialStateKeys(arguments); + // Reject unknown argument names before resolving the output mode; the resolver's + // per-argument lookups silently ignore them. + List tableArgInfos = ArgumentInfo.filterTableArguments(arguments); + PtfChangelogModeValidator changelogValidator = + new PtfChangelogModeValidator( + tableArgInfos, + tableArgumentChangelogModes, + tableArgumentUpsertKeys, + onTimeColumnName); + changelogValidator.validateConfiguration(); + Map argumentConverters = new HashMap<>(); Map stateConverters = new HashMap<>(); createConverters(arguments, argumentConverters, stateConverters, classLoader); @@ -1265,7 +1441,6 @@ public ProcessTableFunctionTestHarness build() throws Exception { // Extract table arguments for output type derivation // SystemTypeInference needs table semantics for pass-through column deduplication - List tableArgInfos = ArgumentInfo.filterTableArguments(arguments); // The system inference yields the full operator output row (partition keys, // pass-through columns, and rowtime); harnessOutputConverter stamps those field names. @@ -1286,25 +1461,12 @@ public ProcessTableFunctionTestHarness build() throws Exception { deriveOutputType( function, dataTypeFactory, baseTypeInference, arguments, tableArgInfos); - // Validate onTimeColumn configuration - if (onTimeColumnName != null) { - boolean foundInAnyTable = - tableArgInfos.stream() - .anyMatch( - t -> getFieldNames(t.dataType).contains(onTimeColumnName)); - checkArgument( - foundInAnyTable, - "withOnTimeColumn references column '%s' which does not exist in any " - + "table argument. Available table arguments and their columns: %s", - onTimeColumnName, - tableArgInfos.stream() - .collect( - Collectors.toMap( - t -> t.name, t -> getFieldNames(t.dataType)))); - } - TestHarnessTimerManager timerManager = new TestHarnessTimerManager(); + ChangelogMode effectiveOutputChangelogMode = + resolveOutputChangelogMode(function, arguments); + changelogValidator.validateResolvedOutputMode(effectiveOutputChangelogMode); + return new ProcessTableFunctionTestHarness<>( function, functionContext, @@ -1316,7 +1478,23 @@ public ProcessTableFunctionTestHarness build() throws Exception { ptfOutputType, stateManager, timerManager, - onTimeColumnName); + onTimeColumnName, + effectiveOutputChangelogMode); + } + + private ChangelogMode resolveOutputChangelogMode( + ProcessTableFunction function, List arguments) { + if (function instanceof ChangelogFunction) { + List tableAndScalarArguments = + arguments.stream() + .filter(arg -> !(arg instanceof StateArgumentInfo)) + .collect(Collectors.toList()); + return new PtfChangelogModeResolver( + (ChangelogFunction) function, tableAndScalarArguments) + .resolve(); + } else { + return ChangelogMode.insertOnly(); + } } /** @@ -1506,7 +1684,7 @@ private void validatePartitionConsistency(List arguments) { for (ArgumentInfo arg : arguments) { if (arg instanceof TableArgumentInfo) { TableArgumentInfo tableArg = (TableArgumentInfo) arg; - if (tableArg.isSetSemantic && tableArg.partitionColumnNames != null) { + if (tableArg.isPartitioned()) { partitionedTables.add(tableArg); } } @@ -1571,7 +1749,7 @@ private void validateInitialStateKeys(List arguments) { arguments.stream() .filter(arg -> arg instanceof TableArgumentInfo) .map(arg -> (TableArgumentInfo) arg) - .filter(t -> t.isSetSemantic && t.partitionColumnNames != null) + .filter(t -> t.isPartitioned()) .findFirst(); if (partitionedTable.isEmpty()) { @@ -1652,7 +1830,7 @@ private TestHarnessStateManager.PartitionKeyInfo extractPartitionKeyInfo( arguments.stream() .filter(arg -> arg instanceof TableArgumentInfo) .map(arg -> (TableArgumentInfo) arg) - .filter(t -> t.isSetSemantic && t.partitionColumnNames != null) + .filter(t -> t.isPartitioned()) .findFirst(); if (partitionedTable.isEmpty()) { @@ -1737,9 +1915,7 @@ private List extractAndValidateTypeInference( Map tableSemanticsMap = new HashMap<>(); for (int i = 0; i < tableArgs.size(); i++) { TableArgumentInfo tArg = tableArgs.get(i); - int[] partitionIndices = getPartitionColumnIndices(tArg); - tableSemanticsMap.put( - i, new TestHarnessTableSemantics(tArg.dataType, partitionIndices)); + tableSemanticsMap.put(i, buildTableSemanticsForInference(tArg, -1)); } TestHarnessCallContext callContext = new TestHarnessCallContext(); @@ -1896,16 +2072,18 @@ private ArgumentInfo buildArgumentInfo(StaticArgument staticArg) { extractAndValidatePartitionColumns(name, dataType, hasOptionalPartitionBy); } - boolean hasPassColumnsThrough = - staticArg.getTraits().contains(StaticArgumentTrait.PASS_COLUMNS_THROUGH); - if (primaryTrait == ArgumentTrait.SCALAR) { ScalarArgumentConfiguration config = scalarArgs.get(name); Object value = config != null ? config.value : null; return new ScalarArgumentInfo(name, dataType, value); } else { return new TableArgumentInfo( - name, dataType, primaryTrait, partitionColumnNames, hasPassColumnsThrough); + name, + dataType, + staticArg.getTraits(), + partitionColumnNames, + tableArgumentChangelogModes.get(name), + tableArgumentUpsertKeys.get(name)); } } @@ -2066,7 +2244,6 @@ private DataType deriveOutputType( TableArgumentInfo tableArg = tableArgsByName.get(argName); if (tableArg != null) { - int[] partitionIndices = getPartitionColumnIndices(tableArg); int timeColumnIndex = -1; if (onTimeColumnName != null) { int idx = getFieldNames(tableArg.dataType).indexOf(onTimeColumnName); @@ -2075,9 +2252,7 @@ private DataType deriveOutputType( } } tableSemanticsMap.put( - i, - new TestHarnessTableSemantics( - tableArg.dataType, partitionIndices, timeColumnIndex)); + i, buildTableSemanticsForInference(tableArg, timeColumnIndex)); } } } @@ -2156,7 +2331,7 @@ private void handleEvalInvocationException( *

Represents validated argument information combining PTF signature, type inference results, * and builder configuration. */ - private abstract static class ArgumentInfo { + abstract static class ArgumentInfo { final String name; final DataType dataType; @@ -2197,32 +2372,56 @@ static class StateArgumentInfo extends ArgumentInfo { } } - /** Table argument with partitioning and output prepending strategy. */ - private static class TableArgumentInfo extends ArgumentInfo { + /** Resolved metadata for a table argument. */ + static class TableArgumentInfo extends ArgumentInfo { final String[] partitionColumnNames; - final boolean isSetSemantic; - final OutputPrependStrategy prependStrategy; + final Set traits; + @Nullable final ChangelogMode changelogMode; + final List upsertKeys; TableArgumentInfo( String name, DataType dataType, - ArgumentTrait primaryTrait, + Set traits, String[] partitionColumnNames, - boolean hasPassColumnsThrough) { + @Nullable ChangelogMode changelogMode, + @Nullable List upsertKeys) { super(name, dataType); this.partitionColumnNames = partitionColumnNames; - this.isSetSemantic = (primaryTrait == ArgumentTrait.SET_SEMANTIC_TABLE); - this.prependStrategy = - hasPassColumnsThrough - ? OutputPrependStrategy.ALL_COLUMNS - : (this.isSetSemantic && partitionColumnNames != null) - ? OutputPrependStrategy.PARTITION_KEYS - : OutputPrependStrategy.NONE; + this.traits = traits; + this.changelogMode = changelogMode; + this.upsertKeys = upsertKeys != null ? upsertKeys : Collections.emptyList(); + } + + boolean is(StaticArgumentTrait trait) { + return traits.contains(trait); + } + + ChangelogMode effectiveChangelogMode() { + return changelogMode != null ? changelogMode : ChangelogMode.insertOnly(); + } + + boolean isSetSemantic() { + return is(StaticArgumentTrait.SET_SEMANTIC_TABLE); + } + + boolean isPartitioned() { + return isSetSemantic() && partitionColumnNames != null; + } + + OutputPrependStrategy prependStrategy() { + if (is(StaticArgumentTrait.PASS_COLUMNS_THROUGH)) { + return OutputPrependStrategy.ALL_COLUMNS; + } else if (isPartitioned()) { + return OutputPrependStrategy.PARTITION_KEYS; + } else { + return OutputPrependStrategy.NONE; + } } } /** Scalar (constant) argument. */ - private static class ScalarArgumentInfo extends ArgumentInfo { + static class ScalarArgumentInfo extends ArgumentInfo { final Object value; ScalarArgumentInfo(String name, DataType dataType, Object value) { diff --git a/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/PtfChangelogModeResolver.java b/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/PtfChangelogModeResolver.java new file mode 100644 index 0000000000000..00a5327a437bf --- /dev/null +++ b/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/PtfChangelogModeResolver.java @@ -0,0 +1,161 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.runtime.functions; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.functions.ChangelogFunction; +import org.apache.flink.table.functions.TableSemantics; +import org.apache.flink.types.RowKind; + +import org.apache.commons.lang3.ClassUtils; + +import java.util.List; +import java.util.Optional; + +/** + * Derives the {@link ChangelogMode} a {@link ChangelogFunction}-implementing PTF would report from + * {@link org.apache.flink.table.functions.ProcessTableFunction.Context#getChangelogMode()}. + */ +@Internal +final class PtfChangelogModeResolver { + + private static final ChangelogMode MOST_PERMISSIVE_UPSERT_HINT = ChangelogMode.upsert(false); + + private final ChangelogFunction function; + private final List arguments; + + PtfChangelogModeResolver( + ChangelogFunction function, + List arguments) { + this.function = function; + this.arguments = arguments; + } + + ChangelogMode resolve() { + ChangelogMode emittedKinds = probeEmittedKinds(); + if (emittedKinds.containsOnly(RowKind.INSERT)) { + return assembleChangelogMode(emittedKinds, false, false); + } + + boolean requiresUpdateBefore = probeRequiresUpdateBefore(emittedKinds); + boolean keyOnlyDeletes = + !requiresUpdateBefore + && emittedKinds.contains(RowKind.DELETE) + && probeUsesKeyOnlyDeletes(emittedKinds); + + return assembleChangelogMode(emittedKinds, requiresUpdateBefore, keyOnlyDeletes); + } + + /** + * Probes with the most permissive upsert hint, which never offers UPDATE_BEFORE, so the + * function's intrinsic output is revealed: an append-only function answers INSERT-only. + */ + private ChangelogMode probeEmittedKinds() { + return function.getChangelogMode(contextFor(MOST_PERMISSIVE_UPSERT_HINT)); + } + + private boolean probeRequiresUpdateBefore(ChangelogMode emittedKinds) { + ChangelogMode hint = containedKindsExceptUpdateBefore(emittedKinds).build(); + return function.getChangelogMode(contextFor(hint)).contains(RowKind.UPDATE_BEFORE); + } + + private boolean probeUsesKeyOnlyDeletes(ChangelogMode emittedKinds) { + ChangelogMode hint = + containedKindsExceptUpdateBefore(emittedKinds).keyOnlyDeletes(true).build(); + return function.getChangelogMode(contextFor(hint)).keyOnlyDeletes(); + } + + private static ChangelogMode.Builder containedKindsExceptUpdateBefore(ChangelogMode mode) { + ChangelogMode.Builder builder = ChangelogMode.newBuilder(); + for (RowKind kind : mode.getContainedKinds()) { + if (kind != RowKind.UPDATE_BEFORE) { + builder.addContainedKind(kind); + } + } + return builder; + } + + private static ChangelogMode assembleChangelogMode( + ChangelogMode emittedKinds, boolean requiresUpdateBefore, boolean keyOnlyDeletes) { + ChangelogMode.Builder builder = containedKindsExceptUpdateBefore(emittedKinds); + if (emittedKinds.contains(RowKind.DELETE)) { + builder.keyOnlyDeletes(keyOnlyDeletes); + } + if (requiresUpdateBefore && emittedKinds.contains(RowKind.UPDATE_AFTER)) { + builder.addContainedKind(RowKind.UPDATE_BEFORE); + } + return builder.build(); + } + + private Optional argumentAt( + int pos, Class type) { + if (pos < 0 || pos >= arguments.size()) { + return Optional.empty(); + } + ProcessTableFunctionTestHarness.ArgumentInfo arg = arguments.get(pos); + if (!type.isInstance(arg)) { + return Optional.empty(); + } + return Optional.of(type.cast(arg)); + } + + private ChangelogFunction.ChangelogContext contextFor(ChangelogMode syntheticRequiredMode) { + return new ChangelogFunction.ChangelogContext() { + @Override + public ChangelogMode getTableChangelogMode(int pos) { + return argumentAt(pos, ProcessTableFunctionTestHarness.TableArgumentInfo.class) + .map(arg -> arg.effectiveChangelogMode()) + .orElse(null); + } + + @Override + public ChangelogMode getRequiredChangelogMode() { + return syntheticRequiredMode; + } + + @Override + public Optional getTableSemantics(int pos) { + return argumentAt(pos, ProcessTableFunctionTestHarness.TableArgumentInfo.class) + .map( + tableArg -> + ProcessTableFunctionTestHarness.buildTableSemantics( + tableArg, -1)); + } + + @Override + public Optional getArgumentValue(int pos, Class clazz) { + return argumentAt(pos, ProcessTableFunctionTestHarness.ScalarArgumentInfo.class) + .flatMap( + arg -> { + Object value = arg.value; + if (value == null) { + return Optional.empty(); + } + Class targetType = ClassUtils.primitiveToWrapper(clazz); + if (targetType.isInstance(value)) { + @SuppressWarnings("unchecked") + T result = (T) targetType.cast(value); + return Optional.of(result); + } + return Optional.empty(); + }); + } + }; + } +} diff --git a/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/PtfChangelogModeValidator.java b/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/PtfChangelogModeValidator.java new file mode 100644 index 0000000000000..56906239dd2a5 --- /dev/null +++ b/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/PtfChangelogModeValidator.java @@ -0,0 +1,345 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.runtime.functions; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.functions.ProcessTableFunction; +import org.apache.flink.table.types.inference.StaticArgumentTrait; +import org.apache.flink.types.RowKind; + +import javax.annotation.Nullable; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * Validates the changelog-mode configuration of the given {@link ProcessTableFunction} currently + * being tested. + * + *

It validates per-argument input modes and upsert keys against the arguments' declared traits. + * It also validates the resolved output mode against on-time compatibility, the set-semantics + * requirement for upsert output, and the pass-through-columns restriction. + */ +@Internal +final class PtfChangelogModeValidator { + + private final List tableArguments; + private final Map tableArgumentChangelogModes; + private final Map> tableArgumentUpsertKeys; + @Nullable private final String onTimeColumnName; + + PtfChangelogModeValidator( + List tableArguments, + Map tableArgumentChangelogModes, + Map> tableArgumentUpsertKeys, + @Nullable String onTimeColumnName) { + this.tableArguments = tableArguments; + this.tableArgumentChangelogModes = tableArgumentChangelogModes; + this.tableArgumentUpsertKeys = tableArgumentUpsertKeys; + this.onTimeColumnName = onTimeColumnName; + } + + /** + * Validates the per-argument changelog configuration: unknown argument names, SUPPORT_UPDATES + * requiring an explicit mode, REQUIRE_UPDATE_BEFORE/REQUIRE_FULL_DELETE trait compatibility, + * upsert mode requiring set semantics with matching partition columns, upsert-key column + * validity, and on-time column existence. + * + * @throws IllegalArgumentException if configuration references unknown arguments + * @throws IllegalStateException if configuration violates documented constraints + */ + void validateConfiguration() { + validateChangelogModeArgsAreKnown(); + validateUpsertKeyColumnsExist(); + // Validated here at config time so a missing column reports "does not exist" instead of the + // later, more confusing on-time/changelog compatibility error. + validateOnTimeColumnExists(); + + for (ProcessTableFunctionTestHarness.TableArgumentInfo tableArg : tableArguments) { + if (tableArg.is(StaticArgumentTrait.SUPPORT_UPDATES)) { + validateUpdatingInput(tableArg); + } else { + validateNonUpdatingInputIsInsertOnly(tableArg); + } + } + } + + private Set tableArgNames() { + return tableArguments.stream().map(t -> t.name).collect(Collectors.toSet()); + } + + private void validateChangelogModeArgsAreKnown() { + Set validTableArgNames = tableArgNames(); + for (String argName : tableArgumentChangelogModes.keySet()) { + if (!validTableArgNames.contains(argName)) { + throw new IllegalArgumentException( + String.format( + "Unknown table argument: '%s'. Available table arguments: %s", + argName, validTableArgNames)); + } + } + } + + private void validateUpsertKeyColumnsExist() { + for (Map.Entry> entry : tableArgumentUpsertKeys.entrySet()) { + String argName = entry.getKey(); + List candidateKeys = entry.getValue(); + + ProcessTableFunctionTestHarness.TableArgumentInfo tableArg = + tableArguments.stream() + .filter(t -> t.name.equals(argName)) + .findFirst() + .orElseThrow( + () -> + new IllegalArgumentException( + String.format( + "Unknown table argument for upsert key: '%s'. " + + "Available table arguments: %s", + argName, tableArgNames()))); + + for (String[] candidateKey : candidateKeys) { + validateUpsertKeyColumnNames(tableArg, candidateKey); + } + } + } + + private void validateOnTimeColumnExists() { + if (onTimeColumnName == null) { + return; + } + boolean foundInAnyTable = + tableArguments.stream() + .anyMatch( + t -> + ProcessTableFunctionTestHarness.getFieldNames(t.dataType) + .contains(onTimeColumnName)); + if (!foundInAnyTable) { + throw new IllegalArgumentException( + String.format( + "withOnTimeColumn references column '%s' which does not exist in any " + + "table argument. Available table arguments and their columns: %s", + onTimeColumnName, + tableArguments.stream() + .collect( + Collectors.toMap( + t -> t.name, + t -> + ProcessTableFunctionTestHarness + .getFieldNames(t.dataType))))); + } + } + + private static void validateUpdatingInput( + ProcessTableFunctionTestHarness.TableArgumentInfo tableArg) { + ChangelogMode mode = requireExplicitChangelogMode(tableArg); + requireUpdateBeforeIfDeclared(tableArg, mode); + requireFullDeleteIfDeclared(tableArg, mode); + if (isUpsertStyleInput(mode)) { + validateUpsertStyleInput(tableArg, mode); + } + } + + private static ChangelogMode requireExplicitChangelogMode( + ProcessTableFunctionTestHarness.TableArgumentInfo tableArg) { + if (tableArg.changelogMode == null) { + throw new IllegalStateException( + String.format( + "Table argument '%s' declares SUPPORT_UPDATES but no changelog mode " + + "was configured. Use .withTableArgumentChangelogMode(\"%s\", ...) " + + "to specify what changelog mode this argument receives.", + tableArg.name, tableArg.name)); + } + return tableArg.changelogMode; + } + + private static void requireUpdateBeforeIfDeclared( + ProcessTableFunctionTestHarness.TableArgumentInfo tableArg, ChangelogMode mode) { + // Insert-only mode is legal even with REQUIRE_UPDATE_BEFORE since it has no updates to + // encode. + if (tableArg.is(StaticArgumentTrait.REQUIRE_UPDATE_BEFORE) + && !mode.containsOnly(RowKind.INSERT) + && !mode.contains(RowKind.UPDATE_BEFORE)) { + throw new IllegalStateException( + String.format( + "Table argument '%s' declares REQUIRE_UPDATE_BEFORE but " + + "configured mode %s does not include UPDATE_BEFORE.", + tableArg.name, mode)); + } + } + + private static void requireFullDeleteIfDeclared( + ProcessTableFunctionTestHarness.TableArgumentInfo tableArg, ChangelogMode mode) { + if (tableArg.is(StaticArgumentTrait.REQUIRE_FULL_DELETE) && mode.keyOnlyDeletes()) { + throw new IllegalStateException( + String.format( + "Table argument '%s' declares REQUIRE_FULL_DELETE but " + + "configured mode %s has keyOnlyDeletes=true.", + tableArg.name, mode)); + } + } + + /** + * An update without UPDATE_BEFORE is upsert-style and needs a co-located key; a retract (with + * UPDATE_BEFORE) carries its own before-image and needs none. + */ + private static boolean isUpsertStyleInput(ChangelogMode mode) { + return !mode.containsOnly(RowKind.INSERT) && !mode.contains(RowKind.UPDATE_BEFORE); + } + + private static void validateUpsertStyleInput( + ProcessTableFunctionTestHarness.TableArgumentInfo tableArg, ChangelogMode mode) { + if (!tableArg.isPartitioned()) { + throw new IllegalStateException( + String.format( + "Table argument '%s' is configured with upsert mode %s, " + + "but this is only possible for SET_SEMANTIC_TABLE arguments " + + "with non-empty PARTITION BY columns. " + + "ROW_SEMANTIC_TABLE arguments or arguments with no " + + "partitioning can only use insertOnly() or all() modes.", + tableArg.name, mode)); + } + List candidateKeys = tableArg.upsertKeys; + if (candidateKeys.isEmpty()) { + throw new IllegalStateException( + String.format( + "Table argument '%s' is configured with upsert mode %s, " + + "but no upsert key was configured. " + + "Use .upsertKey(...) to specify " + + "the upsert key columns.", + tableArg.name, mode)); + } + + // The planner delivers an upsert (no UPDATE_BEFORE) stream to a PTF only when the partition + // columns coincide with one of the input's candidate identity keys; otherwise it must + // retract. Mirror that: partition columns must equal one candidate (order-independent). + Set partitionCols = new HashSet<>(Arrays.asList(tableArg.partitionColumnNames)); + boolean partitionMatchesCandidate = + candidateKeys.stream() + .anyMatch(key -> new HashSet<>(Arrays.asList(key)).equals(partitionCols)); + if (!partitionMatchesCandidate) { + throw new IllegalStateException( + String.format( + "Table argument '%s' partition columns %s do not match any " + + "configured upsert-key candidate %s. " + + "For upsert input modes, partition columns must contain " + + "the exact same set of columns as one candidate upsert " + + "key (order-independent).", + tableArg.name, + Arrays.toString(tableArg.partitionColumnNames), + candidateKeys.stream() + .map(Arrays::toString) + .collect(Collectors.toList()))); + } + } + + private static void validateNonUpdatingInputIsInsertOnly( + ProcessTableFunctionTestHarness.TableArgumentInfo tableArg) { + // A non-SUPPORT_UPDATES argument is contractually insert-only, so any other configured mode + // describes a stream the planner could never deliver here. + ChangelogMode configured = tableArg.changelogMode; + if (configured != null && !configured.equals(ChangelogMode.insertOnly())) { + throw new IllegalStateException( + String.format( + "Table argument '%s' does not declare SUPPORT_UPDATES, so " + + "its changelog mode must be insertOnly(). " + + "Configured mode: %s", + tableArg.name, configured)); + } + } + + /** + * Validates the resolved output changelog mode against on-time compatibility, the set-semantics + * requirement for upsert output, and the pass-through-columns restriction. + * + *

Deliverability and the "non-ChangelogFunction is insert-only" rule are not checked here: + * the mode always originates from {@link PtfChangelogModeResolver} (which only assembles + * deliverable modes) or the insert-only default for a non-{@link + * org.apache.flink.table.functions.ChangelogFunction}, so both hold by construction. + * + * @param outputMode the resolved output changelog mode + * @throws IllegalStateException if the mode violates on-time, set-semantics, or pass-through + * constraints + */ + void validateResolvedOutputMode(ChangelogMode outputMode) { + validateOnTimeCompatibleWithChangelogModes(outputMode); + validateUpsertOutputRequiresSetSemantics(outputMode); + validatePassThroughColumnsCompatibleWithOutputMode(outputMode); + } + + private void validateOnTimeCompatibleWithChangelogModes(ChangelogMode outputMode) { + if (onTimeColumnName == null) { + return; + } + boolean anyUpdatingInput = + tableArguments.stream() + .anyMatch(t -> !t.effectiveChangelogMode().containsOnly(RowKind.INSERT)); + boolean updatingOutput = !outputMode.containsOnly(RowKind.INSERT); + if (anyUpdatingInput || updatingOutput) { + throw new IllegalStateException( + "Time operations using the `on_time` argument are currently not " + + "supported for PTFs that consume or produce updates."); + } + } + + private void validateUpsertOutputRequiresSetSemantics(ChangelogMode outputMode) { + if (outputMode.containsOnly(RowKind.INSERT) || outputMode.contains(RowKind.UPDATE_BEFORE)) { + return; + } + for (ProcessTableFunctionTestHarness.TableArgumentInfo tableArg : tableArguments) { + if (!tableArg.isSetSemantic()) { + throw new IllegalStateException( + String.format( + "PTFs that take table arguments with row semantics don't " + + "support upsert output. Table argument '%s' must " + + "use set semantics.", + tableArg.name)); + } + } + } + + private void validatePassThroughColumnsCompatibleWithOutputMode(ChangelogMode outputMode) { + if (outputMode.containsOnly(RowKind.INSERT)) { + return; + } + for (ProcessTableFunctionTestHarness.TableArgumentInfo tableArg : tableArguments) { + if (tableArg.is(StaticArgumentTrait.PASS_COLUMNS_THROUGH)) { + throw new IllegalStateException( + "Pass-through columns are not supported for PTFs that produce updates."); + } + } + } + + private void validateUpsertKeyColumnNames( + ProcessTableFunctionTestHarness.TableArgumentInfo tableArg, String[] upsertKeyColumns) { + List fieldNames = ProcessTableFunctionTestHarness.getFieldNames(tableArg.dataType); + for (String columnName : upsertKeyColumns) { + if (!fieldNames.contains(columnName)) { + throw new IllegalArgumentException( + String.format( + "Upsert key column '%s' not found in table argument '%s'. " + + "Available fields: %s", + columnName, tableArg.name, fieldNames)); + } + } + } +} diff --git a/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/TestHarnessTableSemantics.java b/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/TestHarnessTableSemantics.java index a271070093a09..05a3ce14dde84 100644 --- a/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/TestHarnessTableSemantics.java +++ b/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/TestHarnessTableSemantics.java @@ -23,7 +23,8 @@ import org.apache.flink.table.functions.TableSemantics; import org.apache.flink.table.types.DataType; -import java.util.Collections; +import javax.annotation.Nullable; + import java.util.List; import java.util.Optional; @@ -34,24 +35,19 @@ class TestHarnessTableSemantics implements TableSemantics { private final int[] partitionByColumns; private final List upsertKeyColumns; private final int timeColumnIndex; - - TestHarnessTableSemantics(DataType dataType, int[] partitionByColumns) { - this(dataType, partitionByColumns, Collections.emptyList(), -1); - } - - TestHarnessTableSemantics(DataType dataType, int[] partitionByColumns, int timeColumnIndex) { - this(dataType, partitionByColumns, Collections.emptyList(), timeColumnIndex); - } + @Nullable private final ChangelogMode changelogMode; TestHarnessTableSemantics( DataType dataType, int[] partitionByColumns, List upsertKeyColumns, - int timeColumnIndex) { + int timeColumnIndex, + @Nullable ChangelogMode changelogMode) { this.dataType = dataType; this.partitionByColumns = partitionByColumns; this.upsertKeyColumns = upsertKeyColumns; this.timeColumnIndex = timeColumnIndex; + this.changelogMode = changelogMode; } @Override @@ -81,7 +77,7 @@ public int timeColumn() { @Override public Optional changelogMode() { - return Optional.empty(); + return Optional.ofNullable(changelogMode); } @Override diff --git a/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarnessChangelogModeTest.java b/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarnessChangelogModeTest.java new file mode 100644 index 0000000000000..6954290f3394f --- /dev/null +++ b/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarnessChangelogModeTest.java @@ -0,0 +1,317 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.runtime.functions; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.TableRuntimeException; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.runtime.functions.ProcessTableFunctionTestHarness.TableArgument; +import org.apache.flink.types.Row; +import org.apache.flink.types.RowKind; + +import org.junit.jupiter.api.Test; + +import java.time.LocalDateTime; +import java.util.List; +import java.util.Optional; + +import static org.apache.flink.table.runtime.functions.PtfTestFunctions.KEY_VALUE_ROW; +import static org.apache.flink.table.runtime.functions.PtfTestFunctions.TIMED_ROW; +import static org.apache.flink.table.runtime.functions.PtfTestFunctions.VALUE_ROW; +import static org.apache.flink.table.runtime.functions.PtfTestFunctions.inputArg; +import static org.apache.flink.table.runtime.functions.PtfTestFunctions.onInput; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Changelog-mode behavior of {@link ProcessTableFunctionTestHarness}: what a configured mode makes + * visible to the function, and which {@link RowKind}s are accepted on the way in and out. Rejection + * of illegal changelog configuration at {@code build()} time lives in {@link + * PtfChangelogModeValidatorTest}; the PTFs under test live in {@link PtfTestFunctions}. + */ +class ProcessTableFunctionTestHarnessChangelogModeTest { + + // ------------------------------------------------------------------------- + // Table argument (input) changelog mode + // ------------------------------------------------------------------------- + + @Test + void testTableArgumentChangelogModeIsPerArgument() throws Exception { + // A configured updating mode must not leak into a sibling argument. + try (ProcessTableFunctionTestHarness h = + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.TwoTableChangelogModePTF.class) + .withTableArgument( + TableArgument.forName("retracting") + .type(DataTypes.of(VALUE_ROW)) + .changelogMode(ChangelogMode.all()) + .build()) + .withTableArgument( + TableArgument.forName("plain") + .type(DataTypes.of(VALUE_ROW)) + .build()) + .build()) { + h.processElementForTable("retracting", Row.of(1)); + + Row row = h.getOutput().get(0); + assertThat(row.getFieldAs(0)).isEqualTo(ChangelogMode.all().toString()); + // "plain" is unconfigured and lacks SUPPORT_UPDATES, so it reports the default. + assertThat(row.getFieldAs(1)).isEqualTo(ChangelogMode.insertOnly().toString()); + } + } + + @Test + void testConfiguredUpsertKeysReachEval() throws Exception { + try (ProcessTableFunctionTestHarness h = + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.TableSemanticsProbePTF.class) + .withTableArgument( + inputArg("ROW") + .partitionBy("a", "b") + .upsertKey("a", "b") + .changelogMode(ChangelogMode.upsert(true)) + .build()) + .build()) { + h.processElement(RowKind.UPDATE_AFTER, 1, 2, 99); + + // Output layout is [partition_a, partition_b, mode, upsertKeys]. Upsert-key column + // names must arrive as resolved field indices, grouped as a single key. + assertThat(h.getOutput().get(0).getFieldAs(3)).isEqualTo("[[0, 1]]"); + } + } + + /** + * Each {@code upsertKey(...)} call declares one more candidate identity key, mirroring the + * planner's {@code TableSemantics#upsertKeyColumns()} which reports a list of candidates. A + * retracting input is used so the candidate set is observed for its own sake, independent of + * the partition-vs-upsert-key check that applies only to upsert-style input. + */ + @Test + void testMultipleUpsertKeyCallsDeclareSeparateCandidates() throws Exception { + try (ProcessTableFunctionTestHarness h = + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.TableSemanticsProbePTF.class) + .withTableArgument( + inputArg("ROW") + .partitionBy("a") + .upsertKey("a") + .upsertKey("b") + .changelogMode(ChangelogMode.all()) + .build()) + .build()) { + h.processElement(RowKind.UPDATE_AFTER, 1, 2, 99); + + // Output layout is [partition_a, mode, upsertKeys]. Both candidates must reach eval() + // as resolved field indices: {a} -> [0] and {b} -> [1]. + assertThat(h.getOutput().get(0).getFieldAs(2)).isEqualTo("[[0], [1]]"); + } + } + + /** + * The planner delivers an upsert (no UPDATE_BEFORE) stream whenever the partition columns equal + * ANY of the input's candidate identity keys, not only the one declared last. So partitioning + * by "key" is accepted even though a further candidate (value) is declared. + */ + @Test + void testUpsertInputAcceptsPartitionMatchingAnyCandidate() throws Exception { + try (ProcessTableFunctionTestHarness h = + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.TableSemanticsProbePTF.class) + .withTableArgument( + inputArg(KEY_VALUE_ROW) + .partitionBy("key") + .upsertKey("key") + .upsertKey("value") + .changelogMode(ChangelogMode.upsert(true)) + .build()) + .build()) { + h.processElement(RowKind.UPDATE_AFTER, 1, 100); + + // Output layout is [partition_key, mode, upsertKeys]; both candidates are surfaced. + assertThat(h.getOutput().get(0).getFieldAs(2)).isEqualTo("[[0], [1]]"); + } + } + + @Test + void testProcessElementRejectsRowKindNotInConfiguredInputMode() throws Exception { + // insertOnly() is trait-consistent for SUPPORT_UPDATES, so the rejection happens at + // processElement() time rather than at build() time. + try (ProcessTableFunctionTestHarness h = + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.PlainUpdatingPassthroughPTF.class) + .withTableArgument( + inputArg(VALUE_ROW) + .changelogMode(ChangelogMode.insertOnly()) + .build()) + .build()) { + assertThatThrownBy(() -> h.processElement(RowKind.UPDATE_AFTER, 10)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("UPDATE_AFTER"); + } + } + + @Test + void testUpsertDeleteNullsNonKeyFieldsWithoutMutatingCaller() throws Exception { + try (ProcessTableFunctionTestHarness h = + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.UpdatingTwoColumnPassthroughPTF.class) + .withTableArgument( + inputArg(KEY_VALUE_ROW) + .partitionBy("key") + .upsertKey("key") + .changelogMode(ChangelogMode.upsert(true)) + .build()) + .build()) { + h.processElement(Row.of(1, 100)); + Row deleteRow = Row.ofKind(RowKind.DELETE, 1, 999); + h.processElement(deleteRow); + + // Output layout is [partition_key, key, value]; non-key fields of a DELETE are nulled + // before the PTF sees them. + assertThat(h.getOutput()).hasSize(2); + Row deleteOutput = h.getOutput().get(1); + assertThat(deleteOutput.getKind()).isEqualTo(RowKind.DELETE); + assertThat(deleteOutput.getFieldAs(2)).isNull(); + + // Nulling must not write through to the caller's Row. + assertThat(deleteRow.getFieldAs(0)).isEqualTo(1); + assertThat(deleteRow.getFieldAs(1)).isEqualTo(999); + } + } + + @Test + void testChangelogModeEmptyDuringTypeInference() throws Exception { + // TableSemantics#changelogMode() must return Optional.empty() during type inference, + // matching the contract of the real Flink planner's CallBindingCallContext. + PtfTestFunctions.TypeInferenceChangelogModeProbePTF.capturedChangelogMode = null; + try (ProcessTableFunctionTestHarness h = + onInput(PtfTestFunctions.TypeInferenceChangelogModeProbePTF.class, VALUE_ROW) + .build()) { + assertThat(PtfTestFunctions.TypeInferenceChangelogModeProbePTF.capturedChangelogMode) + .isEqualTo(Optional.empty()); + h.processElement(Row.of(1)); + } + } + + @Test + void testProcessElementPreservesRowKindOnUpdatingArgument() throws Exception { + // RowKind is preserved through processing on an argument with SUPPORT_UPDATES. + try (ProcessTableFunctionTestHarness h = + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.UpdatingPassthroughPTF.class) + .withTableArgument( + inputArg(VALUE_ROW).changelogMode(ChangelogMode.all()).build()) + .build()) { + h.processElement(RowKind.INSERT, 10); + h.processElement(RowKind.UPDATE_BEFORE, 15); + h.processElement(RowKind.UPDATE_AFTER, 20); + h.processElement(RowKind.DELETE, 30); + + List output = h.getOutput(); + assertThat(output).hasSize(4); + assertThat(output.get(0).getKind()).isEqualTo(RowKind.INSERT); + assertThat(output.get(0).getField("value")).isEqualTo(10); + assertThat(output.get(1).getKind()).isEqualTo(RowKind.UPDATE_BEFORE); + assertThat(output.get(1).getField("value")).isEqualTo(15); + assertThat(output.get(2).getKind()).isEqualTo(RowKind.UPDATE_AFTER); + assertThat(output.get(2).getField("value")).isEqualTo(20); + assertThat(output.get(3).getKind()).isEqualTo(RowKind.DELETE); + assertThat(output.get(3).getField("value")).isEqualTo(30); + } + } + + @Test + void testProcessElementRejectsUpdateOnNonUpdatingArgument() throws Exception { + try (ProcessTableFunctionTestHarness h = + ProcessTableFunctionTestHarness.ofClass(PtfTestFunctions.PassthroughPTF.class) + .withTableArgument(inputArg(VALUE_ROW).build()) + .build()) { + assertThatThrownBy(() -> h.processElement(RowKind.UPDATE_AFTER, 42)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("SUPPORT_UPDATES"); + } + } + + // ------------------------------------------------------------------------- + // Output changelog mode + // ------------------------------------------------------------------------- + + @Test + void testDerivedOutputChangelogModeTracksInputMode() throws Exception { + // With no explicit output mode, build() must run the resolver. This PTF's + // getChangelogMode() echoes its input mode, so the echoed value proves both that the + // resolver ran and that configured input modes reached it. + try (ProcessTableFunctionTestHarness h = + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.InputDrivenChangelogModePTF.class) + .withTableArgument( + inputArg(KEY_VALUE_ROW) + .partitionBy("key") + .upsertKey("key") + .changelogMode(ChangelogMode.upsert(true)) + .build()) + .build()) { + h.processElement(RowKind.UPDATE_AFTER, 1, 42); + assertThat(h.getOutput().get(0).getFieldAs(1)) + .isEqualTo(ChangelogMode.upsert(true).toString()); + } + } + + @Test + void testUpsertOutputMapsNullValueRowToDelete() throws Exception { + try (ProcessTableFunctionTestHarness h = + ProcessTableFunctionTestHarness.ofClass(PtfTestFunctions.UpsertSetSemanticPTF.class) + .withTableArgument(inputArg(KEY_VALUE_ROW).partitionBy("key").build()) + .build()) { + h.processElement(Row.of(1, 100)); + h.processElement(Row.of(1, null)); + + assertThat(h.getOutput()).hasSize(2); + assertThat(h.getOutput().get(0).getKind()).isEqualTo(RowKind.UPDATE_AFTER); + assertThat(h.getOutput().get(1).getKind()).isEqualTo(RowKind.DELETE); + } + } + + @Test + void testCollectRejectsRowKindNotInDefaultOutputMode() throws Exception { + // A PTF that is not a ChangelogFunction falls back to insertOnly(), and the rejection must + // name the kinds that were permitted. + try (ProcessTableFunctionTestHarness h = + onInput(PtfTestFunctions.InvalidRowKindPTF.class, VALUE_ROW).build()) { + assertThatThrownBy(() -> h.processElement(Row.of(1))) + .isInstanceOf(TableRuntimeException.class) + .hasMessageContainingAll("Invalid row kind received", "DELETE", "[INSERT]"); + } + } + + @Test + void testOnTimerRejectsRowKindNotInConfiguredOutputMode() throws Exception { + try (ProcessTableFunctionTestHarness h = + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.TimerEmitsInvalidRowKindPTF.class) + .withTableArgument(inputArg(TIMED_ROW).partitionBy("partition").build()) + .withOnTimeColumn("ts") + .build()) { + h.processElement(Row.of("P1", LocalDateTime.of(2025, 1, 1, 0, 0, 1))); + assertThatThrownBy(() -> h.setWatermark(LocalDateTime.of(2025, 1, 1, 0, 0, 2))) + .isInstanceOf(TableRuntimeException.class) + .hasMessageContainingAll("Invalid row kind received", "DELETE"); + } + } +} diff --git a/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarnessTest.java b/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarnessTest.java index ef124d2efca69..2f24000a35631 100644 --- a/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarnessTest.java +++ b/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarnessTest.java @@ -32,7 +32,6 @@ import org.apache.flink.table.functions.TableSemantics; import org.apache.flink.table.runtime.functions.ProcessTableFunctionTestHarness.TableArgument; import org.apache.flink.types.Row; -import org.apache.flink.types.RowKind; import org.junit.jupiter.api.Test; @@ -51,13 +50,6 @@ class ProcessTableFunctionTestHarnessTest { - @DataTypeHint("ROW") - public static class PassthroughPTF extends ProcessTableFunction { - public void eval(@ArgumentHint(ArgumentTrait.ROW_SEMANTIC_TABLE) Row input) { - collect(input); - } - } - /** Passthrough PTF for testing field ordering. */ @DataTypeHint("ROW") public static class UserValuePassthroughPTF extends ProcessTableFunction { @@ -717,35 +709,6 @@ void testTableProcessingWithScalarArgumentWrongType() { // Argument Trait Tests // ------------------------------------------------------------------------- - @Test - void testProcessElementWithRowKind() throws Exception { - // Verify RowKind is preserved through processing (ROW_SEMANTIC_TABLE) - try (ProcessTableFunctionTestHarness harness = - ProcessTableFunctionTestHarness.ofClass(PassthroughPTF.class) - .withTableArgument( - TableArgument.forName("input") - .type(DataTypes.of("ROW")) - .build()) - .build()) { - - harness.processElement(RowKind.INSERT, 10); - harness.processElement(RowKind.UPDATE_BEFORE, 15); - harness.processElement(RowKind.UPDATE_AFTER, 20); - harness.processElement(RowKind.DELETE, 30); - - List output = harness.getOutput(); - assertThat(output).hasSize(4); - assertThat(output.get(0).getKind()).isEqualTo(RowKind.INSERT); - assertThat(output.get(0).getField("value")).isEqualTo(10); - assertThat(output.get(1).getKind()).isEqualTo(RowKind.UPDATE_BEFORE); - assertThat(output.get(1).getField("value")).isEqualTo(15); - assertThat(output.get(2).getKind()).isEqualTo(RowKind.UPDATE_AFTER); - assertThat(output.get(2).getField("value")).isEqualTo(20); - assertThat(output.get(3).getKind()).isEqualTo(RowKind.DELETE); - assertThat(output.get(3).getField("value")).isEqualTo(30); - } - } - @Test void testPassColumnsThroughTrait() throws Exception { // Verify PASS_COLUMNS_THROUGH prepends ALL input columns (not just partition keys) @@ -1312,7 +1275,7 @@ void testProcessElementOnMultiTableThrows() throws Exception { @Test void testClearOutput() throws Exception { try (ProcessTableFunctionTestHarness harness = - ProcessTableFunctionTestHarness.ofClass(PassthroughPTF.class) + ProcessTableFunctionTestHarness.ofClass(PtfTestFunctions.PassthroughPTF.class) .withTableArgument( TableArgument.forName("input") .type(DataTypes.of("ROW")) @@ -1381,7 +1344,7 @@ void testFunctionOutputExcludesPrependedPartitionKey() throws Exception { @Test void testProcessElementForTableWithInvalidName() throws Exception { try (ProcessTableFunctionTestHarness harness = - ProcessTableFunctionTestHarness.ofClass(PassthroughPTF.class) + ProcessTableFunctionTestHarness.ofClass(PtfTestFunctions.PassthroughPTF.class) .withTableArgument( TableArgument.forName("input") .type(DataTypes.of("ROW")) @@ -2489,10 +2452,10 @@ void testContextTableSemanticsAndChangelogMode() throws Exception { assertThat(harness.getOutput()).hasSize(1); Row evalResult = harness.getOutput().get(0); - assertThat(evalResult.getFieldAs(0).toString()).isEqualTo("P1"); - assertThat(evalResult.getFieldAs(1).toString()).isEqualTo("[0]"); + assertThat(evalResult.getFieldAs(0)).isEqualTo("P1"); + assertThat(evalResult.getFieldAs(1)).isEqualTo("[0]"); assertThat((int) evalResult.getFieldAs(2)).isEqualTo(1); - assertThat(evalResult.getFieldAs(3).toString()) + assertThat(evalResult.getFieldAs(3)) .isEqualTo(ChangelogMode.insertOnly().toString()); harness.clearOutput(); @@ -2500,10 +2463,10 @@ void testContextTableSemanticsAndChangelogMode() throws Exception { assertThat(harness.getOutput()).hasSize(1); Row timerResult = harness.getOutput().get(0); - assertThat(timerResult.getFieldAs(0).toString()).isEqualTo("P1"); - assertThat(timerResult.getFieldAs(1).toString()).isEqualTo("[0]"); + assertThat(timerResult.getFieldAs(0)).isEqualTo("P1"); + assertThat(timerResult.getFieldAs(1)).isEqualTo("[0]"); assertThat((int) timerResult.getFieldAs(2)).isEqualTo(1); - assertThat(timerResult.getFieldAs(3).toString()) + assertThat(timerResult.getFieldAs(3)) .isEqualTo(ChangelogMode.insertOnly().toString()); } } @@ -2582,7 +2545,7 @@ void testWatermarkAdvancesWithoutTimers() throws Exception { @Test void testWatermarkAdvancesWithoutOnTimeColumn() throws Exception { try (ProcessTableFunctionTestHarness harness = - ProcessTableFunctionTestHarness.ofClass(PassthroughPTF.class) + ProcessTableFunctionTestHarness.ofClass(PtfTestFunctions.PassthroughPTF.class) .withTableArgument( TableArgument.forName("input") .type(DataTypes.of("ROW")) diff --git a/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/PtfChangelogModeResolverTest.java b/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/PtfChangelogModeResolverTest.java new file mode 100644 index 0000000000000..57a42c907068c --- /dev/null +++ b/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/PtfChangelogModeResolverTest.java @@ -0,0 +1,401 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.runtime.functions; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.catalog.DataTypeFactory; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.functions.ChangelogFunction; +import org.apache.flink.table.functions.FunctionKind; +import org.apache.flink.table.functions.TableSemantics; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.inference.StaticArgumentTrait; +import org.apache.flink.table.types.inference.TypeInference; +import org.apache.flink.types.RowKind; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import javax.annotation.Nullable; + +import java.util.Arrays; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.function.Function; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Unit tests for {@link PtfChangelogModeResolver}, exercising it directly against mock {@link + * ChangelogFunction} implementations rather than through {@link + * ProcessTableFunctionTestHarness.Builder}. + * + *

The resolver probes the function up to three times with different required-mode hints to + * settle whether the output carries UPDATE_BEFORE and whether deletes are key-only. These tests + * assert the resolved mode and the {@link ChangelogFunction.ChangelogContext} the resolver exposes. + */ +class PtfChangelogModeResolverTest { + + /** A {@link ChangelogFunction} whose answer is driven by a supplied behavior function. */ + private static class StubChangelogFunction implements ChangelogFunction { + private final Function behavior; + + StubChangelogFunction(Function behavior) { + this.behavior = behavior; + } + + /** + * Always returns the same {@link ChangelogMode}, regardless of context — a valid pattern + * per {@link ChangelogFunction}'s own Javadoc. + */ + static StubChangelogFunction fixed(ChangelogMode mode) { + return new StubChangelogFunction(ctx -> mode); + } + + /** Answers based on the call index (1-based), for probe-dependent behavior. */ + static StubChangelogFunction perCall(Function byCallIndex) { + int[] calls = {0}; + return new StubChangelogFunction(ctx -> byCallIndex.apply(++calls[0])); + } + + @Override + public ChangelogMode getChangelogMode(ChangelogContext ctx) { + return behavior.apply(ctx); + } + + @Override + public TypeInference getTypeInference(DataTypeFactory typeFactory) { + throw new UnsupportedOperationException("Not used by these tests."); + } + + @Override + public FunctionKind getKind() { + throw new UnsupportedOperationException("Not used by these tests."); + } + } + + // ------------------------------------------------------------------------- + // Resolved mode for fixed answers + // ------------------------------------------------------------------------- + + @Test + void testInsertOnlyAnswerResolvesToInsertOnly() { + StubChangelogFunction fn = StubChangelogFunction.fixed(ChangelogMode.insertOnly()); + + assertThat(resolve(fn)).isEqualTo(ChangelogMode.insertOnly()); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("upsertAnswerCases") + void testUpsertAnswerResolvesToItself(String name, ChangelogMode fixedAnswer) { + // A fixed upsert answer is preserved as-is. The upsert(false) case additionally proves that + // a function ignoring the key-only-deletes hint keeps its full deletes, since a full delete + // is a strictly compatible superset of a key-only delete. + StubChangelogFunction fn = StubChangelogFunction.fixed(fixedAnswer); + + assertThat(resolve(fn)).isEqualTo(fixedAnswer); + } + + private static Stream upsertAnswerCases() { + return Stream.of( + Arguments.of("key-only deletes preserved", ChangelogMode.upsert(true)), + Arguments.of( + "full deletes preserved when key-only hint is ignored", + ChangelogMode.upsert(false))); + } + + @Test + void testRetractAnswerResolvesToItself() { + // A retract answer keeps UPDATE_BEFORE (a retract stream always carries full deletes). + StubChangelogFunction fn = StubChangelogFunction.fixed(ChangelogMode.all()); + + assertThat(resolve(fn)).isEqualTo(ChangelogMode.all()); + } + + @Test + void testUpdateModeWithoutDeleteHasNoKeyOnlyDeletes() { + // A mode without DELETE has no key-only distinction to settle. + StubChangelogFunction fn = + StubChangelogFunction.fixed( + ChangelogMode.newBuilder() + .addContainedKind(RowKind.INSERT) + .addContainedKind(RowKind.UPDATE_AFTER) + .build()); + + ChangelogMode result = resolve(fn); + + assertThat(result.contains(RowKind.DELETE)).isFalse(); + assertThat(result.keyOnlyDeletes()).isFalse(); + } + + // ------------------------------------------------------------------------- + // Assembling the resolved mode from hint-dependent answers + // ------------------------------------------------------------------------- + + @Test + void testUpdateBeforeAddedWhenOnlyTheUpdateBeforeProbeReportsIt() { + // The emitted-kinds probe answers upsert(true) (no UPDATE_BEFORE); the update-before probe + // answers all(). The resolved mode must gain UPDATE_BEFORE from that later answer. + StubChangelogFunction fn = + StubChangelogFunction.perCall( + call -> call == 1 ? ChangelogMode.upsert(true) : ChangelogMode.all()); + + assertThat(resolve(fn)).isEqualTo(ChangelogMode.all()); + } + + @Test + void testUpdateBeforeNotLeakedFromHintWhenFunctionDoesNotAskForIt() { + // The reverse direction: the function returns all() only when UPDATE_BEFORE is requested, + // and the derived hints never request it. Without proper filtering, UPDATE_BEFORE would + // leak into the result. + StubChangelogFunction fn = + new StubChangelogFunction( + ctx -> + ctx.getRequiredChangelogMode().contains(RowKind.UPDATE_BEFORE) + ? ChangelogMode.all() + : ChangelogMode.upsert(true)); + + ChangelogMode result = resolve(fn); + + assertThat(result).isEqualTo(ChangelogMode.upsert(true)); + assertThat(result.contains(RowKind.UPDATE_BEFORE)).isFalse(); + } + + @Test + void testUpdateBeforeNotAddedWhenEmittedKindsLackUpdateAfter() { + // UPDATE_BEFORE is only valid alongside UPDATE_AFTER, so an update-before-probe answer of + // all() must not introduce it when the emitted kinds were [INSERT, DELETE]. + StubChangelogFunction fn = + StubChangelogFunction.perCall( + call -> + call == 1 + ? ChangelogMode.newBuilder() + .addContainedKind(RowKind.INSERT) + .addContainedKind(RowKind.DELETE) + .build() + : ChangelogMode.all()); + + ChangelogMode result = resolve(fn); + + assertThat(result.contains(RowKind.UPDATE_AFTER)).isFalse(); + assertThat(result.contains(RowKind.UPDATE_BEFORE)).isFalse(); + assertThat(result.contains(RowKind.DELETE)).isTrue(); + } + + @Test + void testDeleteShapeProbeOnlyChangesKeyOnlyDeletesNotEmittedKinds() { + // The delete-shape probe may only flip the keyOnlyDeletes bit, never the kind set from the + // emitted-kinds probe. Here it degrades to insertOnly() when asked for key-only deletes; + // the + // upsert kind set must survive with keyOnlyDeletes=false. + StubChangelogFunction fn = + new StubChangelogFunction( + ctx -> { + ChangelogMode hint = ctx.getRequiredChangelogMode(); + if (hint.contains(RowKind.UPDATE_BEFORE) || !hint.keyOnlyDeletes()) { + return ChangelogMode.upsert(false); + } + return ChangelogMode.insertOnly(); + }); + + assertThat(resolve(fn)).isEqualTo(ChangelogMode.upsert(false)); + } + + @Test + void testInsertOnlyShortCircuitStripsStrayKeyOnlyDeletesFlag() { + // The emitted-kinds probe answers [INSERT] with keyOnlyDeletes=true (an impossible + // combination constructable programmatically but never produced in practice). The + // insert-only short-circuit must still route through the assembler to strip the stray flag. + StubChangelogFunction fn = + StubChangelogFunction.fixed( + ChangelogMode.newBuilder() + .addContainedKind(RowKind.INSERT) + .keyOnlyDeletes(true) + .build()); + + ChangelogMode result = resolve(fn); + + assertThat(result.contains(RowKind.INSERT)).isTrue(); + assertThat(result.keyOnlyDeletes()).isFalse(); + } + + // ------------------------------------------------------------------------- + // ChangelogContext exposed to the function + // ------------------------------------------------------------------------- + + @Test + void testContextTableChangelogModesReflectConfiguredInputModes() { + // A multi-argument PTF: table "t1" (configured to all()), a scalar in between (must report + // null per ChangelogContext#getTableChangelogMode's own Javadoc), and table "t2" (left at + // its insertOnly() default). Position 0 also confirms that the SQL-operand-ordered list the + // resolver receives maps positionally onto configured modes. + List arguments = + Arrays.asList( + tableArgWithChangelogMode( + "t1", + DataTypes.ROW(DataTypes.FIELD("v", DataTypes.INT())), + ChangelogMode.all(), + StaticArgumentTrait.ROW_SEMANTIC_TABLE, + StaticArgumentTrait.SUPPORT_UPDATES), + new ProcessTableFunctionTestHarness.ScalarArgumentInfo( + "s", DataTypes.INT(), 1), + tableArg( + "t2", + DataTypes.ROW(DataTypes.FIELD("n", DataTypes.STRING())), + null, + StaticArgumentTrait.ROW_SEMANTIC_TABLE)); + + Map observed = new HashMap<>(); + StubChangelogFunction fn = + new StubChangelogFunction( + ctx -> { + for (int pos = 0; pos < arguments.size(); pos++) { + observed.put(pos, ctx.getTableChangelogMode(pos)); + } + return ChangelogMode.insertOnly(); + }); + + resolve(fn, arguments); + + assertThat(observed.get(0)).isEqualTo(ChangelogMode.all()); + assertThat(observed.get(1)).isNull(); + assertThat(observed.get(2)).isEqualTo(ChangelogMode.insertOnly()); + } + + @Test + void testContextTableSemanticsReflectConfiguredPartitionAndUpsertKeys() { + // The recorded TableSemantics must expose both the configured partition key and the upsert + // key as resolved field indices, mirroring what production TableSemantics provides. + List arguments = + Collections.singletonList( + tableArgWithUpsertKey( + "input", + DataTypes.ROW( + DataTypes.FIELD("k", DataTypes.STRING()), + DataTypes.FIELD("v", DataTypes.INT())), + new String[] {"k"}, + new String[] {"k"}, + StaticArgumentTrait.SET_SEMANTIC_TABLE)); + + Map observed = new HashMap<>(); + StubChangelogFunction fn = + new StubChangelogFunction( + ctx -> { + ctx.getTableSemantics(0).ifPresent(sem -> observed.put(0, sem)); + return ChangelogMode.upsert(true); + }); + + resolve(fn, arguments); + + TableSemantics semantics = observed.get(0); + assertThat(semantics).isNotNull(); + assertThat(semantics.partitionByColumns()).containsExactly(0); + assertThat(semantics.upsertKeyColumns()).hasSize(1); + assertThat(semantics.upsertKeyColumns().get(0)).containsExactly(0); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("argumentValueCases") + void testContextArgumentValue( + String name, + @Nullable Object configuredValue, + Class requestedClass, + @Nullable Object expected) { + List arguments = + Collections.singletonList( + new ProcessTableFunctionTestHarness.ScalarArgumentInfo( + "arg", DataTypes.INT(), configuredValue)); + + Object[] captured = new Object[1]; + StubChangelogFunction fn = + new StubChangelogFunction( + ctx -> { + captured[0] = ctx.getArgumentValue(0, requestedClass).orElse(null); + return ChangelogMode.insertOnly(); + }); + + resolve(fn, arguments); + + assertThat(captured[0]).isEqualTo(expected); + } + + private static Stream argumentValueCases() { + return Stream.of( + Arguments.of("configured value is returned", 42, Integer.class, 42), + Arguments.of("null value yields empty", null, Integer.class, null), + Arguments.of("incompatible class yields empty", 42, String.class, null), + Arguments.of("primitive class matches boxed value", 5, int.class, 5)); + } + + // ------------------------------------------------------------------------- + // Helpers + // ------------------------------------------------------------------------- + + private static ChangelogMode resolve(ChangelogFunction fn) { + return resolve(fn, Collections.emptyList()); + } + + private static ChangelogMode resolve( + ChangelogFunction fn, List arguments) { + return new PtfChangelogModeResolver(fn, arguments).resolve(); + } + + private static ProcessTableFunctionTestHarness.TableArgumentInfo tableArg( + String name, + DataType dataType, + @Nullable String[] partitionColumnNames, + StaticArgumentTrait... traits) { + return new ProcessTableFunctionTestHarness.TableArgumentInfo( + name, + dataType, + EnumSet.copyOf(Arrays.asList(traits)), + partitionColumnNames, + null, + null); + } + + private static ProcessTableFunctionTestHarness.TableArgumentInfo tableArgWithChangelogMode( + String name, + DataType dataType, + ChangelogMode changelogMode, + StaticArgumentTrait... traits) { + return new ProcessTableFunctionTestHarness.TableArgumentInfo( + name, dataType, EnumSet.copyOf(Arrays.asList(traits)), null, changelogMode, null); + } + + private static ProcessTableFunctionTestHarness.TableArgumentInfo tableArgWithUpsertKey( + String name, + DataType dataType, + @Nullable String[] partitionColumnNames, + String[] upsertKey, + StaticArgumentTrait... traits) { + return new ProcessTableFunctionTestHarness.TableArgumentInfo( + name, + dataType, + EnumSet.copyOf(Arrays.asList(traits)), + partitionColumnNames, + null, + Collections.singletonList(upsertKey)); + } +} diff --git a/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/PtfChangelogModeValidatorTest.java b/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/PtfChangelogModeValidatorTest.java new file mode 100644 index 0000000000000..d833fd80e17c3 --- /dev/null +++ b/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/PtfChangelogModeValidatorTest.java @@ -0,0 +1,286 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.runtime.functions; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.runtime.functions.ProcessTableFunctionTestHarness.TableArgument; +import org.apache.flink.types.Row; +import org.apache.flink.types.RowKind; + +import org.assertj.core.api.ThrowableAssert.ThrowingCallable; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import java.util.Arrays; +import java.util.stream.Stream; + +import static org.apache.flink.table.runtime.functions.PtfTestFunctions.KEY_VALUE_ROW; +import static org.apache.flink.table.runtime.functions.PtfTestFunctions.TIMED_ROW; +import static org.apache.flink.table.runtime.functions.PtfTestFunctions.VALUE_ROW; +import static org.apache.flink.table.runtime.functions.PtfTestFunctions.inputArg; +import static org.apache.flink.table.runtime.functions.PtfTestFunctions.onInput; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Tests for the changelog-mode rules enforced by {@link PtfChangelogModeValidator}, driven through + * {@link ProcessTableFunctionTestHarness.Builder#build()} so that the wiring between builder and + * validator is covered too. The trait combinations each case relies on are declared by the PTFs in + * {@link PtfTestFunctions}. + */ +class PtfChangelogModeValidatorTest { + + @ParameterizedTest(name = "{0}") + @MethodSource("validConfigurations") + void testValidConfigurationBuildsAndAcceptsInsert( + String name, ProcessTableFunctionTestHarness.Builder builder, Row input) + throws Exception { + try (ProcessTableFunctionTestHarness h = builder.build()) { + h.processElement(input); + assertThat(h.getOutput()).hasSize(1); + assertThat(h.getOutput().get(0).getKind()).isEqualTo(RowKind.INSERT); + } + } + + private static Stream validConfigurations() { + return Stream.of( + Arguments.of( + // Upsert key comparison is set-based: partition (a, b) matches key (b, a). + "upsert key ordering independent of partition column order", + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.UpdatingTwoColumnPassthroughPTF.class) + .withTableArgument( + inputArg("ROW") + .partitionBy("a", "b") + .upsertKey("b", "a") + .changelogMode(ChangelogMode.upsert(true)) + .build()), + Row.of(1, 2)), + Arguments.of( + // Insert-only mode has no updates, so it trivially satisfies the trait. + "REQUIRE_UPDATE_BEFORE with insertOnly()", + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.UpdatingPassthroughPTF.class) + .withTableArgument( + inputArg(VALUE_ROW) + .changelogMode(ChangelogMode.insertOnly()) + .build()), + Row.of(1)), + Arguments.of( + // upsert(false) has keyOnlyDeletes=false, satisfying REQUIRE_FULL_DELETE. + "REQUIRE_FULL_DELETE with upsert(false)", + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.RequireFullDeletePassthroughPTF.class) + .withTableArgument( + inputArg(VALUE_ROW) + .partitionBy("value") + .upsertKey("value") + .changelogMode(ChangelogMode.upsert(false)) + .build()), + Row.of(1)), + Arguments.of( + // Name validation is deferred to build(), so order must not matter. + "upsert key configured before the table argument", + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.UpdatingTwoColumnPassthroughPTF.class) + .withTableArgument( + TableArgument.forName("input") + .upsertKey("key") + .type(DataTypes.of(KEY_VALUE_ROW)) + .partitionBy("key") + .changelogMode(ChangelogMode.upsert(true)) + .build()), + Row.of(1, 42))); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("invalidConfigurations") + void testInvalidConfigurationRejectedAtBuild( + String name, + ThrowingCallable build, + Class expectedType, + String expectedMessage) { + assertThatThrownBy(build).isInstanceOf(expectedType).hasMessageContaining(expectedMessage); + } + + private static Stream invalidConfigurations() { + return Stream.of( + invalid( + "REQUIRE_UPDATE_BEFORE with a mode lacking UPDATE_BEFORE", + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.UpdatingPassthroughPTF.class) + .withTableArgument( + inputArg(VALUE_ROW) + .changelogMode(ChangelogMode.upsert(true)) + .build()), + IllegalStateException.class, + "declares REQUIRE_UPDATE_BEFORE"), + invalid( + "REQUIRE_FULL_DELETE with keyOnlyDeletes=true", + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.RequireFullDeletePassthroughPTF.class) + .withTableArgument( + inputArg(VALUE_ROW) + .changelogMode(ChangelogMode.upsert(true)) + .build()), + IllegalStateException.class, + "declares REQUIRE_FULL_DELETE"), + invalid( + "updating mode on an argument without SUPPORT_UPDATES", + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.PassthroughPTF.class) + .withTableArgument( + inputArg(VALUE_ROW) + .changelogMode(ChangelogMode.all()) + .build()), + IllegalStateException.class, + "does not declare SUPPORT_UPDATES"), + invalid( + "upsert input mode on a non-partitioned argument", + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.PlainUpdatingPassthroughPTF.class) + .withTableArgument( + inputArg(VALUE_ROW) + .upsertKey("value") + .changelogMode(ChangelogMode.upsert(true)) + .build()), + IllegalStateException.class, + "only possible for SET_SEMANTIC_TABLE arguments"), + invalid( + // {INSERT, DELETE} with keyOnlyDeletes(true) has no UPDATE_BEFORE, so it + // is upsert-style and must still require partitioning + a matching upsert + // key. + "upsert-style mode without UPDATE_AFTER on a non-partitioned argument", + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.PlainUpdatingPassthroughPTF.class) + .withTableArgument( + inputArg(VALUE_ROW) + .changelogMode( + keyOnlyDeleteMode( + RowKind.INSERT, RowKind.DELETE)) + .build()), + IllegalStateException.class, + "only possible for SET_SEMANTIC_TABLE arguments"), + invalid( + "upsert input mode without a configured upsert key", + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.UpdatingTwoColumnPassthroughPTF.class) + .withTableArgument( + inputArg(KEY_VALUE_ROW) + .partitionBy("key") + .changelogMode(ChangelogMode.upsert(true)) + .build()), + IllegalStateException.class, + "no upsert key was configured"), + invalid( + "upsert key columns not matching partition columns", + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.UpdatingTwoColumnPassthroughPTF.class) + .withTableArgument( + inputArg(KEY_VALUE_ROW) + .partitionBy("key") + .upsertKey("value") + .changelogMode(ChangelogMode.upsert(true)) + .build()), + IllegalStateException.class, + "do not match any configured upsert-key candidate"), + invalid( + // Upsert-key column validation runs unconditionally, so a non-upsert mode + // must not skip it. + "upsert key naming a column absent from the schema", + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.UpdatingTwoColumnPassthroughPTF.class) + .withTableArgument( + inputArg(KEY_VALUE_ROW) + .partitionBy("key") + .upsertKey("nonexistentColumn") + .changelogMode(ChangelogMode.all()) + .build()), + IllegalArgumentException.class, + "Upsert key column 'nonexistentColumn' not found"), + invalid( + "SUPPORT_UPDATES declared without a configured changelog mode", + onInput(PtfTestFunctions.PlainUpdatingPassthroughPTF.class, VALUE_ROW), + IllegalStateException.class, + "declares SUPPORT_UPDATES but no changelog mode"), + invalid( + "upsert output with row-semantic table arguments", + onInput(PtfTestFunctions.UpdatingOutputPTF.class, VALUE_ROW), + IllegalStateException.class, + "don't support upsert output"), + invalid( + "on_time with updating output", + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.OnTimeUpdatingOutputPTF.class) + .withTableArgument( + inputArg(TIMED_ROW).partitionBy("partition").build()) + .withOnTimeColumn("ts"), + IllegalStateException.class, + "not supported for PTFs that consume or produce updates"), + invalid( + // The updating input alone triggers the rejection; the PTF declares + // insert-only output so the violation is unambiguously input-side. + "on_time with updating input", + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.TimerWithUpdatingInputPTF.class) + .withTableArgument( + inputArg(TIMED_ROW) + .partitionBy("partition") + .upsertKey("partition") + .changelogMode(ChangelogMode.upsert(true)) + .build()) + .withOnTimeColumn("ts"), + IllegalStateException.class, + "not supported for PTFs that consume or produce updates"), + invalid( + "on_time column absent from every table argument", + ProcessTableFunctionTestHarness.ofClass( + PtfTestFunctions.TimerEmitsInvalidRowKindPTF.class) + .withTableArgument( + inputArg(TIMED_ROW).partitionBy("partition").build()) + .withOnTimeColumn("missing"), + IllegalArgumentException.class, + "does not exist in any"), + invalid( + "pass-through columns with updating output", + onInput(PtfTestFunctions.PassThroughUpdatingOutputPTF.class, VALUE_ROW), + IllegalStateException.class, + "Pass-through columns are not supported")); + } + + private static Arguments invalid( + String name, + ProcessTableFunctionTestHarness.Builder builder, + Class expectedType, + String expectedMessage) { + return Arguments.of(name, (ThrowingCallable) builder::build, expectedType, expectedMessage); + } + + private static ChangelogMode keyOnlyDeleteMode(RowKind... kinds) { + return modeBuilder(kinds).keyOnlyDeletes(true).build(); + } + + private static ChangelogMode.Builder modeBuilder(RowKind... kinds) { + ChangelogMode.Builder builder = ChangelogMode.newBuilder(); + Arrays.stream(kinds).forEach(builder::addContainedKind); + return builder; + } +} diff --git a/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/PtfTestFunctions.java b/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/PtfTestFunctions.java new file mode 100644 index 0000000000000..89ad6090f5dd8 --- /dev/null +++ b/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/PtfTestFunctions.java @@ -0,0 +1,424 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.runtime.functions; + +import org.apache.flink.table.annotation.ArgumentHint; +import org.apache.flink.table.annotation.ArgumentTrait; +import org.apache.flink.table.annotation.DataTypeHint; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.catalog.DataTypeFactory; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.functions.ChangelogFunction; +import org.apache.flink.table.functions.ProcessTableFunction; +import org.apache.flink.table.functions.TableSemantics; +import org.apache.flink.table.types.inference.TypeInference; +import org.apache.flink.types.Row; +import org.apache.flink.types.RowKind; + +import java.time.Duration; +import java.time.LocalDateTime; +import java.util.Arrays; +import java.util.Optional; + +/** + * Shared PTF test fixtures referenced by {@link ProcessTableFunctionTestHarnessTest}, {@link + * ProcessTableFunctionTestHarnessChangelogModeTest} and {@link PtfChangelogModeValidatorTest}. + * + *

Mirrors the role of {@code ProcessTableFunctionTestUtils} in the upstream planner tests. + * + *

Recurring trait combinations

+ * + *

Two declaration constraints explain why several fixtures below look more elaborate than the + * behavior they exercise: + * + *

    + *
  • {@code REQUIRE_UPDATE_BEFORE} and {@code REQUIRE_FULL_DELETE} are documented as valid only + * on {@code SET_SEMANTIC_TABLE} arguments, so fixtures carrying them must use set semantics + * to describe a reachable PTF configuration. + *
  • Multiple table arguments require all of them to use set semantics ({@code + * SystemTypeInference#checkMultipleTableArgs}). + *
+ * + *

{@code OPTIONAL_PARTITION_BY} accompanies {@code SET_SEMANTIC_TABLE} wherever a test does not + * care about partitioning, so that no {@code .partitionBy(...)} call is needed. + */ +class PtfTestFunctions { + + static final String VALUE_ROW = "ROW"; + static final String KEY_VALUE_ROW = "ROW"; + static final String TIMED_ROW = "ROW"; + + /** + * Registers the sole table argument under the name {@code input}, as every single-argument + * fixture here declares it. + */ + static ProcessTableFunctionTestHarness.Builder onInput( + Class> functionClass, String inputType) { + return ProcessTableFunctionTestHarness.ofClass(functionClass) + .withTableArgument(inputArg(inputType).build()); + } + + /** + * Returns a builder for the sole table argument named {@code input}, for callers that need to + * attach further per-argument configuration (e.g. {@code changelogMode(...)}) before building. + */ + static ProcessTableFunctionTestHarness.TableArgument.Builder inputArg(String inputType) { + return ProcessTableFunctionTestHarness.TableArgument.forName("input") + .type(DataTypes.of(inputType)); + } + + /** + * Mirrors a passthrough PTF's declared output mode onto whatever changelog mode its (single) + * table argument carries — the honest declaration for a function that forwards its input + * unchanged, and the shape the planner would infer for it. + */ + private static ChangelogMode echoInputMode(ChangelogFunction.ChangelogContext ctx) { + ChangelogMode inputMode = ctx.getTableChangelogMode(0); + return inputMode == null ? ChangelogMode.insertOnly() : inputMode; + } + + // ------------------------------------------------------------------------- + // Passthrough inputs + // ------------------------------------------------------------------------- + + @DataTypeHint("ROW") + public static class PassthroughPTF extends ProcessTableFunction { + public void eval(@ArgumentHint(ArgumentTrait.ROW_SEMANTIC_TABLE) Row input) { + collect(input); + } + } + + /** Passthrough PTF with {@code SUPPORT_UPDATES} and {@code REQUIRE_UPDATE_BEFORE}. */ + @DataTypeHint("ROW") + public static class UpdatingPassthroughPTF extends ProcessTableFunction + implements ChangelogFunction { + public void eval( + @ArgumentHint({ + ArgumentTrait.SET_SEMANTIC_TABLE, + ArgumentTrait.OPTIONAL_PARTITION_BY, + ArgumentTrait.SUPPORT_UPDATES, + ArgumentTrait.REQUIRE_UPDATE_BEFORE + }) + Row input) { + collect(input); + } + + @Override + public ChangelogMode getChangelogMode(ChangelogContext changelogContext) { + return echoInputMode(changelogContext); + } + } + + /** + * Like {@link UpdatingPassthroughPTF} but without {@code REQUIRE_UPDATE_BEFORE}/{@code + * REQUIRE_FULL_DELETE}, so any updating {@link ChangelogMode} is trait-consistent and only the + * configured mode itself constrains which {@link RowKind}s are accepted. + */ + @DataTypeHint("ROW") + public static class PlainUpdatingPassthroughPTF extends ProcessTableFunction { + public void eval( + @ArgumentHint({ArgumentTrait.ROW_SEMANTIC_TABLE, ArgumentTrait.SUPPORT_UPDATES}) + Row input) { + collect(input); + } + } + + /** Two-column set-semantic variant, for upsert-key and key-only-delete scenarios. */ + @DataTypeHint("ROW") + public static class UpdatingTwoColumnPassthroughPTF extends ProcessTableFunction + implements ChangelogFunction { + public void eval( + @ArgumentHint({ + ArgumentTrait.SET_SEMANTIC_TABLE, + ArgumentTrait.OPTIONAL_PARTITION_BY, + ArgumentTrait.SUPPORT_UPDATES + }) + Row input) { + collect(input); + } + + @Override + public ChangelogMode getChangelogMode(ChangelogContext changelogContext) { + return echoInputMode(changelogContext); + } + } + + @DataTypeHint("ROW") + public static class RequireFullDeletePassthroughPTF extends ProcessTableFunction { + public void eval( + @ArgumentHint({ + ArgumentTrait.SET_SEMANTIC_TABLE, + ArgumentTrait.OPTIONAL_PARTITION_BY, + ArgumentTrait.SUPPORT_UPDATES, + ArgumentTrait.REQUIRE_FULL_DELETE + }) + Row input) { + collect(input); + } + } + + // ------------------------------------------------------------------------- + // TableSemantics probes + // ------------------------------------------------------------------------- + + /** + * Renders what {@link TableSemantics} reports to {@code eval()} for its single table argument: + * the changelog mode, and the upsert key columns as resolved field indices. + */ + @DataTypeHint("ROW") + public static class TableSemanticsProbePTF extends ProcessTableFunction { + public void eval( + Context ctx, + @ArgumentHint({ + ArgumentTrait.SET_SEMANTIC_TABLE, + ArgumentTrait.OPTIONAL_PARTITION_BY, + ArgumentTrait.SUPPORT_UPDATES + }) + Row input) { + TableSemantics semantics = ctx.tableSemanticsFor("input"); + collect( + Row.of( + semantics.changelogMode().map(Object::toString).orElse("empty"), + Arrays.deepToString(semantics.upsertKeyColumns().toArray()))); + } + } + + /** + * Reports the changelog mode each of its two table arguments sees: one that declares {@code + * SUPPORT_UPDATES} and one that does not. + */ + @DataTypeHint("ROW") + public static class TwoTableChangelogModePTF extends ProcessTableFunction { + public void eval( + Context ctx, + @ArgumentHint({ + ArgumentTrait.SET_SEMANTIC_TABLE, + ArgumentTrait.OPTIONAL_PARTITION_BY, + ArgumentTrait.SUPPORT_UPDATES + }) + Row retracting, + @ArgumentHint({ + ArgumentTrait.SET_SEMANTIC_TABLE, + ArgumentTrait.OPTIONAL_PARTITION_BY + }) + Row plain) { + collect(Row.of(modeOf(ctx, "retracting"), modeOf(ctx, "plain"))); + } + + private static String modeOf(Context ctx, String argName) { + return ctx.tableSemanticsFor(argName) + .changelogMode() + .map(Object::toString) + .orElse("empty"); + } + } + + /** + * A PTF whose custom output {@link org.apache.flink.table.types.inference.TypeStrategy} reads + * {@link TableSemantics#changelogMode()} during type inference and captures the result in a + * static field for assertion. + */ + @DataTypeHint("ROW") + public static class TypeInferenceChangelogModeProbePTF extends ProcessTableFunction { + static volatile Optional capturedChangelogMode = null; + + public void eval(@ArgumentHint(ArgumentTrait.ROW_SEMANTIC_TABLE) Row input) { + collect(input); + } + + @Override + public TypeInference getTypeInference(DataTypeFactory typeFactory) { + TypeInference base = super.getTypeInference(typeFactory); + return TypeInference.newBuilder() + .staticArguments(base.getStaticArguments().orElse(null)) + .outputTypeStrategy( + ctx -> { + capturedChangelogMode = + ctx.getTableSemantics(0) + .map(TableSemantics::changelogMode) + .orElse(null); + return base.getOutputTypeStrategy().inferType(ctx); + }) + .build(); + } + } + + // ------------------------------------------------------------------------- + // Output changelog mode + // ------------------------------------------------------------------------- + + /** + * Emits a DELETE without implementing {@link ChangelogFunction}, so the harness must fall back + * to its insert-only default output mode and reject the row. + */ + @DataTypeHint("ROW") + public static class InvalidRowKindPTF extends ProcessTableFunction { + public void eval(@ArgumentHint(ArgumentTrait.ROW_SEMANTIC_TABLE) Row input) { + collect(Row.ofKind(RowKind.DELETE, 1)); + } + } + + /** + * Collects the resolver-derived output mode into its first field so a test can observe it. Its + * {@code getChangelogMode()} ties the output mode to the input mode. + */ + @DataTypeHint("ROW") + public static class InputDrivenChangelogModePTF extends ProcessTableFunction + implements ChangelogFunction { + public void eval( + Context ctx, + @ArgumentHint({ + ArgumentTrait.SET_SEMANTIC_TABLE, + ArgumentTrait.SUPPORT_UPDATES, + ArgumentTrait.OPTIONAL_PARTITION_BY + }) + Row input) { + collect(Row.of(ctx.getChangelogMode().toString())); + } + + @Override + public ChangelogMode getChangelogMode(ChangelogContext changelogContext) { + return echoInputMode(changelogContext); + } + } + + /** Declares upsert output on a row-semantic argument, which the planner rejects. */ + @DataTypeHint("ROW") + public static class UpdatingOutputPTF extends ProcessTableFunction + implements ChangelogFunction { + public void eval(Context ctx, @ArgumentHint(ArgumentTrait.ROW_SEMANTIC_TABLE) Row input) { + int v = input.getFieldAs("value"); + collect(Row.ofKind(RowKind.UPDATE_AFTER, v)); + } + + @Override + public ChangelogMode getChangelogMode(ChangelogContext changelogContext) { + return ChangelogMode.upsert(true); + } + } + + /** Emits upsert-style output, which set semantics allow but row semantics do not. */ + @DataTypeHint("ROW") + public static class UpsertSetSemanticPTF extends ProcessTableFunction + implements ChangelogFunction { + public void eval( + Context ctx, + @ArgumentHint({ + ArgumentTrait.SET_SEMANTIC_TABLE, + ArgumentTrait.OPTIONAL_PARTITION_BY + }) + Row input) { + int key = input.getFieldAs("key"); + Integer value = input.getFieldAs("value"); + if (value == null) { + collect(Row.ofKind(RowKind.DELETE, key, null)); + } else { + collect(Row.ofKind(RowKind.UPDATE_AFTER, key, value)); + } + } + + @Override + public ChangelogMode getChangelogMode(ChangelogContext changelogContext) { + return ChangelogMode.upsert(true); + } + } + + // ------------------------------------------------------------------------- + // Timers + // ------------------------------------------------------------------------- + + /** Used only in tests that assert a build() failure — the body is unreachable. */ + @DataTypeHint("ROW") + public static class TimerWithUpdatingInputPTF extends ProcessTableFunction + implements ChangelogFunction { + public void eval( + Context ctx, + @ArgumentHint({ + ArgumentTrait.SET_SEMANTIC_TABLE, + ArgumentTrait.SUPPORT_UPDATES, + ArgumentTrait.REQUIRE_ON_TIME + }) + Row input) { + throw new UnsupportedOperationException("Never invoked by this test"); + } + + // Insert-only output keeps the on-time rejection attributable to the updating input. + @Override + public ChangelogMode getChangelogMode(ChangelogContext changelogContext) { + return ChangelogMode.insertOnly(); + } + } + + /** + * Insert-only (not a {@link ChangelogFunction}); its timer emits a DELETE the harness rejects. + */ + @DataTypeHint("ROW") + public static class TimerEmitsInvalidRowKindPTF extends ProcessTableFunction { + public void eval( + Context ctx, + @ArgumentHint({ArgumentTrait.SET_SEMANTIC_TABLE, ArgumentTrait.REQUIRE_ON_TIME}) + Row input) { + TimeContext t = ctx.timeContext(LocalDateTime.class); + t.registerOnTime("test", t.time().plus(Duration.ofSeconds(1))); + } + + public void onTimer(OnTimerContext ctx) { + collect(Row.ofKind(RowKind.DELETE, 1)); + } + } + + /** Declares updating output alongside {@code REQUIRE_ON_TIME}, which the harness rejects. */ + @DataTypeHint("ROW") + public static class OnTimeUpdatingOutputPTF extends ProcessTableFunction + implements ChangelogFunction { + public void eval( + Context ctx, + @ArgumentHint({ArgumentTrait.SET_SEMANTIC_TABLE, ArgumentTrait.REQUIRE_ON_TIME}) + Row input) { + throw new UnsupportedOperationException("Never invoked by this test"); + } + + @Override + public ChangelogMode getChangelogMode(ChangelogContext changelogContext) { + return ChangelogMode.all(); + } + } + + /** + * Declares pass-through columns alongside updating output, which the harness rejects to mirror + * production's {@code verifyPassThroughColumnsForUpdates}. + */ + @DataTypeHint("ROW") + public static class PassThroughUpdatingOutputPTF extends ProcessTableFunction + implements ChangelogFunction { + public void eval( + Context ctx, + @ArgumentHint({ + ArgumentTrait.ROW_SEMANTIC_TABLE, + ArgumentTrait.PASS_COLUMNS_THROUGH + }) + Row input) { + throw new UnsupportedOperationException("Never invoked by this test"); + } + + @Override + public ChangelogMode getChangelogMode(ChangelogContext changelogContext) { + return ChangelogMode.all(); + } + } +}