[FLINK-29406][table] Expose finish method for TableFunction #20899
Conversation
| s""" | ||
| |$fieldTerm.finish(); | ||
| """.stripMargin | ||
| reusableFinishStatements.add(finishFunction) |
There was a problem hiding this comment.
nit: just s"$fieldTerm.finish();"?
| @Override | ||
| public void finish() throws Exception { | ||
| ${ctx.reuseFinishCode()} | ||
| super.finish(); |
There was a problem hiding this comment.
NIT: why reuseFinishCode first, super.finish second.
There was a problem hiding this comment.
similar to the close sequence, it is relatively more safer (not always strictly) to handle the actions of the child class first and call the parent public logic last. Btw, I changed the current close call at first but reverted it for focusing on this pr itself.
| @@ -25,7 +25,7 @@ LogicalJoin(condition=[=($3, $1)], joinType=[inner]) | |||
| : +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{}]) | |||
| : :- LogicalProject(o_rowtime=[AS($0, _UTF-16LE'o_rowtime')], o_comment=[AS($1, _UTF-16LE'o_comment')], o_amount=[AS($2, _UTF-16LE'o_amount')], o_currency=[AS($3, _UTF-16LE'o_currency')], o_secondary_key=[AS($4, _UTF-16LE'o_secondary_key')]) | |||
| : : +- LogicalTableScan(table=[[default_catalog, default_database, Orders]]) | |||
| : +- LogicalTableFunctionScan(invocation=[*org.apache.flink.table.functions.TemporalTableFunctionImpl$bbf912e58a3fb2d083552961c0f87dbe*($0)], rowType=[RecordType(TIMESTAMP(3) *ROWTIME* rowtime, VARCHAR(2147483647) comment, VARCHAR(2147483647) currency, INTEGER rate, INTEGER secondary_key)]) | |||
There was a problem hiding this comment.
NIT: Can have serialVersionUID in TableFunction to be compatible with previous version?
There was a problem hiding this comment.
I did try to add the uid, but it doesn't work in the current TemporalTableFunction implementation, it was wrapped into a BridgingSqlFunction (a subclass of calcite's SqlFunction which is not serializable) during the conversion, so give up the change for now.
There was a problem hiding this comment.
Thanks for review this @JingsongLi, I've updated the pr according to your comments.
| @@ -25,7 +25,7 @@ LogicalJoin(condition=[=($3, $1)], joinType=[inner]) | |||
| : +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{}]) | |||
| : :- LogicalProject(o_rowtime=[AS($0, _UTF-16LE'o_rowtime')], o_comment=[AS($1, _UTF-16LE'o_comment')], o_amount=[AS($2, _UTF-16LE'o_amount')], o_currency=[AS($3, _UTF-16LE'o_currency')], o_secondary_key=[AS($4, _UTF-16LE'o_secondary_key')]) | |||
| : : +- LogicalTableScan(table=[[default_catalog, default_database, Orders]]) | |||
| : +- LogicalTableFunctionScan(invocation=[*org.apache.flink.table.functions.TemporalTableFunctionImpl$bbf912e58a3fb2d083552961c0f87dbe*($0)], rowType=[RecordType(TIMESTAMP(3) *ROWTIME* rowtime, VARCHAR(2147483647) comment, VARCHAR(2147483647) currency, INTEGER rate, INTEGER secondary_key)]) | |||
There was a problem hiding this comment.
I did try to add the uid, but it doesn't work in the current TemporalTableFunction implementation, it was wrapped into a BridgingSqlFunction (a subclass of calcite's SqlFunction which is not serializable) during the conversion, so give up the change for now.
| @Override | ||
| public void finish() throws Exception { | ||
| ${ctx.reuseFinishCode()} | ||
| super.finish(); |
There was a problem hiding this comment.
similar to the close sequence, it is relatively more safer (not always strictly) to handle the actions of the child class first and call the parent public logic last. Btw, I changed the current close call at first but reverted it for focusing on this pr itself.
What is the purpose of the change
The task lifecycle was changed in FLINK-22972: a new finish() phase was introduced (extracted the ‘finish’ part out of the ‘close’) and the behavior changed in
closemethod. This was some kind of 'break change' for TableFunction users who did some flush work in previousclosemethod(< 1.14). This pr aims to provide a possible migration path to support it again. If necessary, we can also consider backporting it to the corresponding older versions, including 1.14~1.16.However, as a method that is not recommended for most users, I tend not to add it to the user documentation (just as we didn't actively prompt users to do flush data in the close method before), so it is only been described in the method comments of the TableFunction, users who used this 'advanced' usage in previous versions 'should' be also able to get the usage of the new method.
Brief change log
Verifying this change
CorrelateITCase#testTableFunctionWithFinishMethod
Does this pull request potentially affect one of the following parts:
Documentation