Realtime Compute for Apache Flink:General invocation

Content type parameters

• For single-column input, use content-type to specify the content type. Supported values: text, image_url.

• For multi-column input, use content-types to specify the content type for each column. The supported values and their mapping to Flink SQL column types are listed below:

• We recommend that you specify content-types in the parameters. If you set both content-type and content-types, the framework reports an error. Note: the content-type and content-types settings in model parameters and per-call settings affect each other:

Column-level parameters

Bundle processing

• Processing: Bundle processing first aggregates multiple input records into a batch and then calls the AI model for the batch. This reduces the number of model calls but increases the model inference time for each request. When records can be effectively aggregated, bundle processing can significantly improve operator throughput.

• Parameter description: bundle-size specifies the maximum number of records accumulated in a batch. bundle-allow-latency specifies the maximum wait time when a batch has not reached the bundle-size value. Processing starts when either the batch-size or wait-time condition is met.

• Requirements: Bundle processing takes effect based on the input and output schema of the model.

  • Text generation: For models whose task is chat/completions, the input contains exactly one STRING column of text, and the output contains exactly one STRING or VARIANT column. Multi-column inputs and multimodal inputs such as images, videos, and audio are not supported.

  • Text embedding: For models whose task is embeddings, the input contains exactly one STRING column, and the output contains exactly one ARRAY<FLOAT> column.

• Supported functions: ML_PREDICT, text classification (AI_CLASSIFY), sentiment analysis (AI_SENTIMENT), information extraction (AI_EXTRACT), text summarization (AI_SUMMARIZE), text translation (AI_TRANSLATE), data masking (AI_MASK), and text embedding (AI_EMBED). Among them, ML_PREDICT supports bundle processing only when the registered model meets the preceding requirements.

• Global configuration: This configuration applies to the ML_PREDICT calls and the calls to the listed domain-specific AI functions in the job that meet the preceding requirements. Use the following global parameters to enable bundle processing:

  -- Bundle processing is supported only in VVR 11.9 and later.
  SET 'table.exec.ml-predict.bundle-size' = '10';
  SET 'table.exec.ml-predict.bundle-allow-latency' = '5s';

• Per-call configuration: These parameters apply only to the function call in which they are specified. ML_PREDICT and the listed domain-specific AI functions use the same configuration method. For parameters and examples, see Per-call configuration.

Limits

• Supported only in VVR 11.9 and later.

• You cannot enable error handling when model calls run in synchronous mode.

• Error handling cannot be used together with Vectorization result cache (CACHE_TABLE) at this time.

Enable error handling

Add the FAIL_ON_ERROR => FALSE parameter when you call an AI function.

Prerequisite: Keep the default retry settings of the model: error-handling-strategy='retry' and retry-fallback-strategy='failover'. For more information about model parameters, see Model settings.

Output changes

After you enable error handling, the system appends an error column err_msg of the nullable VARIANT type to the end of the original output fields. This column is generated by the system. You do not need to declare it in the input table or in the OUTPUT clause of CREATE MODEL.

• Call succeeds: err_msg is NULL.

• Call ultimately fails: all model result fields are NULL, and err_msg contains the error information.

If you use SELECT *, the result includes the new error column. If you select fields explicitly, you must specify the column manually. The default column name is err_msg. If the name conflicts with an existing output column, the system appends a numeric suffix, such as err_msg0. Use the actual column name to reference the column.

Examples

CREATE TEMPORARY VIEW input_data AS
SELECT *
FROM (VALUES
  (1, CAST('xxxx (a description of violent conflict that violates humanity)' AS STRING), 'negative'),
  (2, CAST('The customer service of this store has always been among the best' AS STRING), 'positive')
) AS t(id, input_text, expected_label);

CREATE TEMPORARY VIEW ai_results AS
SELECT i.id, i.input_text, i.expected_label,
       r.score, r.label, r.confidence, r.err_msg
FROM input_data AS i,
LATERAL TABLE(AI_SENTIMENT(
  MODEL => MODEL ai_sentiment_model,
  INPUT => i.input_text,
  FAIL_ON_ERROR => FALSE,
  CONFIG => MAP['async', 'true']
)) AS r(score, label, confidence, err_msg);

SELECT * FROM ai_results;

message: 
java.util.concurrent.TimeoutException: AsyncPredictFunction did not complete within the configured timeout. Increase the ML_PREDICT timeout or check external service connectivity.

stack_trace:
java.util.concurrent.TimeoutException: AsyncPredictFunction did not complete within the configured timeout. Increase the ML_PREDICT timeout or check external service connectivity.
at org.apache.flink.table.runtime.operators.predict.AsyncMLPredictRunner.timeout(AsyncMLPredictRunner.java:153)
at org.apache.flink.table.runtime.operators.predict.AsyncMLPredictRunner.timeout(AsyncMLPredictRunner.java:51)
at org.apache.flink.table.runtime.operators.TableKeyedAsyncWaitOperator$KeyedResultHandler.timerTriggered(TableKeyedAsyncWaitOperator.java:462)
at org.apache.flink.table.runtime.operators.TableKeyedAsyncWaitOperator$KeyedResultHandler.lambda$registerTimeout$1(TableKeyedAsyncWaitOperator.java:457)
at org.apache.flink.table.runtime.operators.TableKeyedAsyncWaitOperator.lambda$registerTimer$4(TableKeyedAsyncWaitOperator.java:486)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invokeProcessingTimeCallback(StreamTask.java:2285)
at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$deferCallbackToMailbox$28(StreamTask.java:2276)
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:115)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMail(MailboxProcessor.java:415)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:384)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:369)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:229)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:1273)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:1217)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:1133)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:1110)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:914)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:716)
at java.base/java.lang.Thread.run(Thread.java:991)

How each error type is handled

The following behavior applies when error handling is enabled and the default retry settings are kept. An error row is a row that retains the input, returns empty model results, and records the failure reason in err_msg.

Text

The following example registers and uses a Flink AI Service built-in model to classify input text by sentiment. Example 1 uses MODEL-level parameters. Example 2 overrides user-prompt at call time.

-- Register a built-in model. When content-type is not specified, it defaults to text
CREATE MODEL sentiment_model
INPUT (prompt STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.6-flash',
  'system-prompt' = 'You are a sentiment classifier. Output one label: negative, positive, or neutral'
);

-- Create a source table: simulated product review data
CREATE TEMPORARY VIEW input_table(id, content)
AS VALUES
  (1, 'Great quality, soft fabric, fits perfectly'),
  (2, 'Had loose threads on arrival, faded badly after one wash'),
  (3, 'Received the item, looks as pictured'),
  (4, 'Started pilling after two weeks, customer service refused returns'),
  (5, 'Flattering fit, color is even better than the photo, already ordered a third one');

-- Create a result table
CREATE TEMPORARY TABLE output_table (
  id INT,
  content STRING,
  sentiment STRING
) WITH (
  'connector' = 'print'
);

-- Use ML_PREDICT for real-time inference
-- Example 1: Use MODEL-level parameters
INSERT INTO output_table
SELECT
  id,
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content));

-- Example 2: Specify per-call parameters
INSERT INTO output_table
SELECT
  id,
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content),
  MAP['user-prompt', 'Reply in Chinese']);

Example 1 output:

Example 2 output:

Image

The following example registers a multimodal model and classifies input images.

The image input column (when content-type is image_url) is of the STRING type and supports the following formats:

If images are stored in Blob columns of a Paimon Catalog, you can use the Paimon Blob presigned URL feature to generate a public URL as input. For more information, see Apache Paimon.

-- Register a built-in model with content-type set to image_url
CREATE TEMPORARY MODEL sentiment_model
INPUT (prompt STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.6-flash',
  'content-type' = 'image_url'
);

-- Use ML_PREDICT for real-time inference
INSERT INTO output_table
SELECT
  id,
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content));

Video

The following example registers a multimodal model to understand an input video.

CREATE TEMPORARY MODEL sentiment_model
INPUT (video_url STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.6-plus',
  'content-types' = 'video_url',
  'system-prompt' = 'Briefly describe the video content'
);

CREATE TEMPORARY VIEW input_table(content)
AS VALUES ('https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20251111/ckgcxt/%E5%8D%95%E4%BA%BA%E8%AF%B4%E8%AF%9D-1.mp4');

SELECT
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content));

Text + single image

The following example registers a multimodal model for chat-based inference on text and image inputs.

CREATE MODEL vl_model
INPUT (text_input STRING, image_input STRING)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.5-plus',
  'task' = 'chat/completions',
  'content-types' = 'text;image_url'
);

INSERT INTO result_sink
SELECT content FROM TABLE(ML_PREDICT(
  TABLE image_source,
  MODEL vl_model,
  DESCRIPTOR(text_input, image_input)
));         

Text + multiple images (multiple columns)

Pass each image through a separate INPUT column. Specify image_url for each image column in content-types.

CREATE MODEL vl_model_multi
INPUT (prompt STRING, img1 STRING, img2 STRING)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.5-plus',
  'task' = 'chat/completions',
  'content-types' = 'text;image_url;image_url'
);

SELECT content FROM TABLE(ML_PREDICT(
  TABLE my_source,
  MODEL vl_model_multi,
  DESCRIPTOR(prompt, img1, img2)
));          

Text + multiple images (array)

Pass multiple images in an ARRAY<STRING> column. Use multi_image_urls in content-types.

CREATE MODEL vl_model_array
INPUT (prompt STRING, images ARRAY<STRING>)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.5-plus',
  'task' = 'chat/completions',
  'content-types' = 'text;multi_image_urls'
);

SELECT content FROM TABLE(ML_PREDICT(
  TABLE my_source,
  MODEL vl_model_array,
  DESCRIPTOR(prompt, images)
)); 

Text + video

CREATE TEMPORARY MODEL sentiment_model
INPUT (video_url STRING, question STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.6-plus',
  'content-types' = 'video_url;text'
);

CREATE TEMPORARY VIEW input_table(content, question)
AS VALUES ('https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20251111/ckgcxt/%E5%8D%95%E4%BA%BA%E8%AF%B4%E8%AF%9D-1.mp4', 'How many people are there?');

SELECT
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content, question));