fhueske commented on code in PR #28657:
URL: https://github.com/apache/flink/pull/28657#discussion_r3578305747
##########
docs/content.zh/docs/dev/table/functions/ptfs.md:
##########
@@ -2700,6 +2700,54 @@ void testPOJO() throws Exception {
{{< top >}}
+### Testing Pure Function Output with getFunctionOutput()
+
+`getOutput()` returns the full row the Flink runtime emits: partition keys and
pass-through
+columns are prepended, the `rowtime` column is appended when `on_time` is
configured, and
+non-`Row` output — an atomic value or a POJO — is wrapped or flattened into a
`Row`. Use
+`getFunctionOutput()` to inspect exactly what the PTF itself collected,
without those generated
Review Comment:
```suggestion
`getFunctionOutput()` to inspect exactly what the PTF itself collected,
without those additional
```
or `forwarded`?
"generated" sounds as if the framework would produce the values itself but
it is just "routing" the original values around the function.
##########
flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java:
##########
@@ -137,6 +140,11 @@ private static class TableArgumentConverters {
private final Map<String, TableArgumentConverters> argumentConverters;
private final DataStructureConverter<Object, Object>
harnessOutputConverter;
+ private final boolean outputIsAtomic;
+ private final boolean outputIsRow;
Review Comment:
aren't these mutually exclusive? define an enum and use it here?
##########
flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java:
##########
@@ -815,19 +857,23 @@ private Row extractPartitionKey(TableArgumentInfo
tableArg, Row row) {
private class HarnessCollector implements Collector<OUT> {
@Override
+ @SuppressWarnings("unchecked")
public void collect(OUT record) {
- OUT finalRecord;
+ Row ptfRow = toPtfOutputRow(record);
+ functionOutput.add(outputIsRow ? (OUT) ptfRow : record);
Review Comment:
Why do we need the condition?
Couldn't we just always put `record` into `functionOutput` since it should
capture exactly what the function produces?
##########
flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java:
##########
@@ -163,9 +172,28 @@ private ProcessTableFunctionTestHarness(
this.arguments = arguments;
this.argumentConverters = argumentConverters;
this.harnessOutputConverter = harnessOutputConverter;
+ this.outputIsAtomic =
!LogicalTypeChecks.isCompositeType(ptfOutputType.getLogicalType());
+ this.outputIsRow = ptfOutputType.getLogicalType() instanceof RowType;
+ DataStructureConverter<Object, Object> compositeOutputToInternal =
null;
+ DataStructureConverter<Object, Object> compositeOutputToRow = null;
Review Comment:
Do we need these local vars?
Can't we use the member vars directly?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
