[FLINK-39515][table] Fix compiled plan restore for built-in PTFs with default args#27996
Merged
twalthr merged 2 commits intoapache:masterfrom Apr 23, 2026
Merged
[FLINK-39515][table] Fix compiled plan restore for built-in PTFs with default args#27996twalthr merged 2 commits intoapache:masterfrom
twalthr merged 2 commits intoapache:masterfrom
Conversation
Collaborator
twalthr
reviewed
Apr 22, 2026
Contributor
twalthr
left a comment
twalthr
left a comment
There was a problem hiding this comment.
Good catch @gustavodemorais.
Comment on lines
+378
to
+381
| // SqlDefaultArgOperator is constructed per-call site by FlinkSqlCallBinding and not | ||
| // registered in any operator table, so the lookup above falls through to Calcite's stock | ||
| // SqlStdOperatorTable.DEFAULT_OPERATOR. Rebuild the typed Flink instance here so codegen's | ||
| // `op instanceof SqlDefaultArgOperator` check (StringCallGen) recognizes it on restore. |
Contributor
There was a problem hiding this comment.
Suggested change
| // SqlDefaultArgOperator is constructed per-call site by FlinkSqlCallBinding and not | |
| // registered in any operator table, so the lookup above falls through to Calcite's stock | |
| // SqlStdOperatorTable.DEFAULT_OPERATOR. Rebuild the typed Flink instance here so codegen's | |
| // `op instanceof SqlDefaultArgOperator` check (StringCallGen) recognizes it on restore. | |
| // SqlDefaultArgOperator is constructed per-call site by FlinkSqlCallBinding and not | |
| // registered in any operator table. Rebuild the typed Flink instance here. |
Comment on lines
+27
to
+29
| * Restore tests for the built-in FROM_CHANGELOG PTF. Verifies that {@link | ||
| * org.apache.flink.table.planner.functions.sql.SqlDefaultArgOperator} placeholders for unprovided | ||
| * optional args round-trip through compiled-plan serialization. |
Contributor
There was a problem hiding this comment.
Nobody cares about this specific bug in the future. So comments can be simplified.
Suggested change
| * Restore tests for the built-in FROM_CHANGELOG PTF. Verifies that {@link | |
| * org.apache.flink.table.planner.functions.sql.SqlDefaultArgOperator} placeholders for unprovided | |
| * optional args round-trip through compiled-plan serialization. | |
| * Restore tests for the built-in {@link BuiltInFunctionDefinitsions.FROM_CHANGELOG} PTF. |
Comment on lines
+192
to
+194
| * savepoint restore. The unprovided {@code op} and {@code op_mapping} args become {@code | ||
| * DEFAULT()} placeholders that survive {@code toUdfCall} and reach codegen on the restore path | ||
| * - exercising the {@link org.apache.flink.table.planner.functions.sql.SqlDefaultArgOperator} |
Contributor
There was a problem hiding this comment.
Suggested change
| * savepoint restore. The unprovided {@code op} and {@code op_mapping} args become {@code | |
| * DEFAULT()} placeholders that survive {@code toUdfCall} and reach codegen on the restore path | |
| * - exercising the {@link org.apache.flink.table.planner.functions.sql.SqlDefaultArgOperator} | |
| * savepoint restore. The unprovided {@code op} and {@code op_mapping} args become {@code | |
| * DEFAULT()} placeholders. |
Contributor
Author
| callType = serdeContext.getRexBuilder().deriveReturnType(operator, rexOperands); | ||
| } | ||
| return serdeContext.getRexBuilder().makeCall(callType, operator, rexOperands); | ||
| // SqlDefaultArgOperator is constructed per-call site by FlinkSqlCallBinding and not |
Contributor
There was a problem hiding this comment.
Update RexNodeJsonSerdeTest
Contributor
Author
There was a problem hiding this comment.
Added a test
Contributor
Author
|
Thanks for the review, @twalthr! Simplified comments and added test |
twalthr
reviewed
Apr 22, 2026
| } | ||
|
|
||
| @Test | ||
| public void testDefaultArgOperatorRoundTrip() throws IOException { |
Contributor
There was a problem hiding this comment.
Is there a reason why you added a dedicated test, and not just a new list entry in testRexNodeSerde?
Contributor
Author
There was a problem hiding this comment.
No, there isn't. Thanks for the pointer
…ArgOperator serde test
airlock-confluentinc
Bot
force-pushed
the
FLINK-39515
branch
from
5044eb6 to
6e34eb5
Compare
airlock-confluentinc Bot
pushed a commit
to confluentinc/flink-public
that referenced
this pull request
Apr 23, 2026
…TFs with default args This closes apache#27996.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What is the purpose of the change
When restoring a compiled plan whose query calls a built-in PTF with unprovided optional args (e.g. TO_CHANGELOG(input => TABLE t) without op / op_mapping), code generation fails: Unsupported call: DEFAULT()
Cause: Unprovided slots are filled by SqlDefaultArgOperator, a per-call-site instance that carries the expected return type. It's not registered in any operator table, so on restore RexNodeJsonDeserializer falls through to Calcite's stock SqlStdOperatorTable.DEFAULT_OPERATOR. Codegen's op instanceof SqlDefaultArgOperator check (StringCallGen.scala:242) misses, and the call hits the generic unsupported branch.
The same bug affects any built-in PTF that exposes optional user-level args (those survive toUdfCall and reach codegen). Existing user-defined PTF restore tests don't catch it because their DEFAULT() operands are system args (uid, on_time) that toUdfCall strips.
Fix: In RexNodeJsonDeserializer.deserializeCall, when the deserialized operator's kind is SqlKind.DEFAULT, rebuild it as new SqlDefaultArgOperator(callType) so the typed Flink instance reaches codegen.
Brief change log
Verifying this change
Repro / coverage: New restore tests ToChangelogRestoreTest and FromChangelogRestoreTest, each running a RETRACT_RESTORE program. Both fail with Unsupported call: DEFAULT() without the fix and pass with it.
Does this pull request potentially affect one of the following parts:
@Public(Evolving): (no)Documentation
Was generative AI tooling used to co-author this PR?