Skip to content

[FLINK-40572][runtime] Support dynamic AI model selection - #4525

Merged
yuxiqian merged 2 commits into
apache:masterfrom
haruki-830:FLINK-40572
Sep 9, 2026
Merged

[FLINK-40572][runtime] Support dynamic AI model selection#4525
yuxiqian merged 2 commits into
apache:masterfrom
haruki-830:FLINK-40572

Conversation

@haruki-830

@haruki-830 haruki-830 commented Sep 7, 2026

Copy link
Copy Markdown
Contributor

What is the purpose of this pull request?

This PR fixes FLINK-40572 and follows up on the remaining work discussed in #4520.

Previously, the model argument of an AI function had to be a string literal. The parser extracted the model name before execution, and JaninoCompiler rewrote it to a pre-bound model client variable. As a result, expressions such as columns, IF, and CASE could not be used to select an AI model dynamically.

This PR evaluates the model argument for each record and resolves the selected model at runtime. It allows a pipeline to choose among the models declared in pipeline.model based on record contents.

Brief change log

  • Pass the declared model client map to generated expressions for runtime model lookup.
  • Evaluate the model argument as a regular STRING expression for every record.
  • Support dynamic model selection through columns, IF, and CASE expressions.
  • Remove the static model extraction and capability validation from the parser and composer.
  • Remove the special model-literal rewriting logic from JaninoCompiler.
  • Validate null, undeclared, and capability-incompatible models when the AI function is invoked.
  • Preserve null-input short-circuit behavior without resolving the model.
  • Preserve the precedence of user-defined functions with the same name.
  • Update the English and Chinese AI model documentation.

Verifying this change

This change added and updated tests and can be verified as follows:

  • Updated AiFunctionsTest to cover runtime model resolution, missing models, null model names, incompatible capabilities, and null-input short-circuiting.
  • Updated AiFunctionParserTest to cover dynamic model expressions and user-defined function precedence.
  • Updated FlinkPipelineAiFunctionITCase to verify dynamic model selection in a running pipeline.
  • Verified the runtime and composer tests with Flink 1.20.
  • Verified the runtime and composer tests with Flink 2.x using the flink2 profile.
  • Verified code formatting with mvn spotless:check.

Documentation

  • Does this pull request introduce a new feature? yes
  • If yes, how is the feature documented? English and Chinese documentation

Was generative AI tooling used to co-author this PR?
  • Yes (OpenAI Codex)

Generated-by: OpenAI Codex (GPT-5)

@github-actions github-actions Bot added docs Improvements or additions to documentation composer runtime labels Sep 7, 2026
@haruki-830
haruki-830 marked this pull request as ready for review September 8, 2026 02:16
@leonardBang
leonardBang requested a review from yuxiqian September 9, 2026 02:39

@yuxiqian yuxiqian left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks Haruki for the cleaning-up work, just left some minor comments.

@yuxiqian yuxiqian left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM, just left a follow up note.

argumentNames.add(paramName);
argumentClasses.add(AiModelClient.class);
}
argumentNames.add(JaninoCompiler.DEFAULT_AI_MODEL_CLIENTS);

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not relevant to this PR, but the previous argument declaration & binding implementation is rather weird:

  • Projection column/time parameters are declared and bound in ProjectionColumnProcessor.
  • Filter column/time parameters are declared and bound in TransformFilterProcessor.
  • UDF and AI model parameters are declared in TransformExpressionCompiler.
  • Their values are bound separately in ProjectionColumnProcessor and TransformFilterProcessor.

The order is consistent, but the logic is scattered. Shall we build the complete parameter layout in one shared place and let TransformExpressionCompiler only consume it? This could be done later in a following ticket.

@yuxiqian
yuxiqian merged commit b11d6fd into apache:master Sep 9, 2026
21 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

approved composer docs Improvements or additions to documentation reviewed runtime

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants