Skip to content

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

Open
haruki-830 wants to merge 1 commit into
apache:masterfrom
haruki-830:FLINK-40572
Open

[FLINK-40572][runtime] Support dynamic AI model selection#4525
haruki-830 wants to merge 1 commit into
apache:masterfrom
haruki-830:FLINK-40572

Conversation

@haruki-830

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

  • Add an internal AiModelClientResolver for resolving declared model clients at runtime.
  • 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.


/** Resolves AI model clients by their logical names during expression evaluation. */
@Internal
public class AiModelClientResolver {

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.

Seems this resolver is just wrapping the Map internally. Can we keep it as-is for simplicity?

Comment on lines +197 to +202
private static <T> T resolveModel(
AiModelClientResolver modelClientResolver,
String modelName,
String functionName,
Class<T> requiredCapability,
String capabilityName) {

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.

Stating the required interface explicitly could be useful:

Suggested change
private static <T> T resolveModel(
AiModelClientResolver modelClientResolver,
String modelName,
String functionName,
Class<T> requiredCapability,
String capabilityName) {
private static <T> T resolveModel(
AiModelClientResolver modelClientResolver,
String modelName,
String functionName,
Class<T> requiredCapability) {

and

throw new UnsupportedOperationException(
    "Model '"
            + modelName
            + "' could not be used in "
            + functionName
            + " because it does not implement "
            + requiredCapability.getSimpleName()
            + " interface.");

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

composer docs Improvements or additions to documentation runtime

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants