[jira] [Updated] (FLINK-40513) Introduce multimodal AI functions for image understanding

["Mao Jiayi (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-40513?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Mao Jiayi updated FLINK-40513:
------------------------------
    Description: 
This issue adds multimodal AI functions to the transform framework, so users can process image columns directly in transform expressions.

Two {{@Experimental}} capability interfaces are introduced in {{{}flink-cdc-common{}}}:
 * {{SupportsImageTextGeneration}} — generates text from image bytes and a prompt.
 * {{SupportsImageEmbedding}} — converts image bytes to a float vector.

Two built-in functions are registered:
 * {{AI_IMAGE_COMPLETE(model, image, prompt)}} returns {{{}STRING{}}}. The OpenAI-compatible provider implements this via standard vision chat, encoding images as Base64 data URLs.
 * {{AI_IMAGE_EMBED(model, image)}} returns {{{}ARRAY<FLOAT>{}}}. This function is provided as a framework extension point only. The default OpenAI-compatible provider does not implement image embedding, and the community distribution does not yet include a production-ready provider. Users who need image embedding must wait for a follow-up PR.

Image MIME type is detected from magic bytes (PNG, JPEG, GIF, WebP). A NULL image returns NULL without calling the provider. Empty byte arrays and unrecognized formats are rejected before the HTTP request.

  was:
h3. Background

FLINK-40331 / PR #4498 introduces an opt-in {{FORWARD}} routing strategy for Fluss-to-Fluss synchronization. It preserves the upstream subtask distribution and avoids repartitioning records before the Fluss sink.

However, FORWARD routing is beneficial and safe only when the source and target tables have compatible bucket layouts. The Fluss sink currently does not validate this assumption or explicitly expose bucket-affinity behavior.
h3. Proposed Changes

Enhance the Fluss sink when {{{}sink.partitioning.strategy=FORWARD{}}}:
 # Validate that source and target tables have compatible distribution settings, including bucket count, bucket keys, partition keys, table type and bucketing algorithm.
 # Validate at runtime that each record is routed to the expected target bucket before writing.
 # Fail fast when bucket affinity cannot be guaranteed, such as log tables without deterministic bucket keys.
 # Cache table metadata and bucket calculators in {{{}FlussSinkWriter{}}}.
 # Reuse {{MultiTableWriter}} and its per-table/bucket batching so that each sink subtask writes only its affinity bucket subset.
 # Add metrics and logs for validated records, validation failures and active buckets.

h3. Acceptance Criteria
 * Compatible Fluss-to-Fluss FORWARD jobs write data correctly.
 * Incompatible bucket layouts fail with a clear error instead of silently degrading.
 * Log tables without deterministic bucket keys are rejected in FORWARD mode.
 * DEFAULT routing behavior remains unchanged.
 * Unit and E2E tests cover successful and failed validation scenarios.
 * The implementation is verified with both Flink 1.20 and Flink 2.x.

     Issue Type: New Feature  (was: Improvement)
        Summary: Introduce multimodal AI functions for image understanding  (was: Validate forward routing and optimize Fluss sink writer with bucket affinity)

> Introduce multimodal AI functions for image understanding
> ---------------------------------------------------------
>
>                 Key: FLINK-40513
>                 URL: https://issues.apache.org/jira/browse/FLINK-40513
>             Project: Flink
>          Issue Type: New Feature
>          Components: Flink CDC
>            Reporter: Mao Jiayi
>            Priority: Major
>              Labels: pull-request-available
>
> This issue adds multimodal AI functions to the transform framework, so users can process image columns directly in transform expressions.
> Two {{@Experimental}} capability interfaces are introduced in {{{}flink-cdc-common{}}}:
>  * {{SupportsImageTextGeneration}} — generates text from image bytes and a prompt.
>  * {{SupportsImageEmbedding}} — converts image bytes to a float vector.
> Two built-in functions are registered:
>  * {{AI_IMAGE_COMPLETE(model, image, prompt)}} returns {{{}STRING{}}}. The OpenAI-compatible provider implements this via standard vision chat, encoding images as Base64 data URLs.
>  * {{AI_IMAGE_EMBED(model, image)}} returns {{{}ARRAY<FLOAT>{}}}. This function is provided as a framework extension point only. The default OpenAI-compatible provider does not implement image embedding, and the community distribution does not yet include a production-ready provider. Users who need image embedding must wait for a follow-up PR.
> Image MIME type is detected from magic bytes (PNG, JPEG, GIF, WebP). A NULL image returns NULL without calling the provider. Empty byte arrays and unrecognized formats are rejected before the HTTP request.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)