gustavodemorais commented on code in PR #28120:
URL: https://github.com/apache/flink/pull/28120#discussion_r3196472251
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ToChangelogTypeStrategy.java:
##########
@@ -96,10 +97,14 @@ public List<Signature> getExpectedSignatures(final
FunctionDefinition definition
final String opColumnName = resolveOpColumnName(callContext);
final List<Field> inputFields =
DataType.getFields(semantics.dataType());
+ // Excludes partition keys when set semantics; the framework
prepends them so
+ // including them again here would duplicate the columns.
Review Comment:
removed
##########
flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.xml:
##########
@@ -35,42 +35,42 @@ ProcessTableFunction(invocation=[TO_CHANGELOG(TABLE(#0),
DEFAULT(), DEFAULT(), D
]]>
</Resource>
</TestCase>
- <TestCase name="testRetractSource">
+ <TestCase name="testRetractPartitionBy">
<Resource name="sql">
- <![CDATA[SELECT * FROM TO_CHANGELOG(input => TABLE retract_source)]]>
+ <![CDATA[SELECT * FROM TO_CHANGELOG(input => TABLE retract_source
PARTITION BY id)]]>
</Resource>
<Resource name="ast">
<![CDATA[
-LogicalProject(op=[$0], id=[$1], name=[$2])
-+- LogicalTableFunctionScan(invocation=[TO_CHANGELOG(TABLE(#0), DEFAULT(),
DEFAULT(), DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) op,
INTEGER id, VARCHAR(2147483647) name)])
+LogicalProject(id=[$0], op=[$1], name=[$2])
Review Comment:
yeah, that checks out for me
--
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]
