[
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)
