[FLINK-40416][runtime] Add built-in text AI functions

docs/content.zh/docs/core-concept/ai-model.md

@@ -24,11 +24,30 @@ under the License.
 
 # AI 模型
 
-AI 模型可用于 transform 表达式中的文本补全和 embedding。
+AI 模型可用于 transform 表达式中的文本生成、文本分析和 embedding。
+
+## AI Functions
+
+模型名称必须是字符串常量，并引用 `pipeline.model` 中声明的模型。文本函数要求模型客户端支持文本生成，`AI_EMBED` 要求模型客户端支持 embedding；Pipeline 会在执行前校验引用模型的 capability 是否匹配。
+
+所有文本函数都会将模型返回的 JSON 解析为 `VARIANT`。
+
+| 函数 | 说明 | JSON 字段 |
+|------|------|-----------|
+| `AI_COMPLETE(model, input, system_prompt)` | 使用调用方提供的 system prompt 补全文本。 | `result` |
+| `AI_CLASSIFY(model, input, labels)` | 将输入分类到给定标签之一。 | `category`、`confidence` |
+| `AI_TRANSLATE(model, input, source_lang, target_lang)` | 翻译输入；`source_lang` 可传 `auto` 自动识别。 | `translated_text`、`detected_language` |
+| `AI_SUMMARIZE(model, input, max_length)` | 在指定字符数内生成摘要。 | `summary` |
+| `AI_SENTIMENT(model, input)` | 分析文本情感。 | `score`、`label`、`confidence` |
+| `AI_EXTRACT(model, input, schema)` | 按 schema 字符串描述提取字段。 | `extracted_json` |
+| `AI_MASK(model, input, entities)` | 对指定实体类型进行脱敏。 | `masked_text`、`detected_entities` |
+| `AI_EMBED(model, input)` | 生成 embedding 向量。 | 不返回 JSON，而是返回 `ARRAY<FLOAT>`。 |
+
+六个专用文本函数使用内置英文 prompt 模板，但输入文本可以是任意语言。输入为 `NULL` 时直接返回 `NULL`，且不会调用模型；模型返回 `NULL` 时也返回 `NULL`。非空文本响应必须是语法合法的 JSON，否则当前记录处理失败，错误信息会标明具体 AI 函数。运行时只校验 JSON 语法，不校验响应字段是否存在或字段类型是否匹配。
 
 ## OpenAI-compatible 模型客户端
 
-AI 模型客户端可供 transform 中的 `AI_COMPLETE` 和 `AI_EMBED` 函数引用。使用时，需要通过 `--jar` 将模型实现 JAR（例如 `flink-cdc-pipeline-model-openai-compatible`）添加到 Pipeline 命令中。
+AI 模型客户端可供上述 AI Functions 引用。使用时，需要通过 `--jar` 将模型实现 JAR（例如 `flink-cdc-pipeline-model-openai-compatible`）添加到 Pipeline 命令中。
 
 OpenAI-compatible 客户端支持调用实现 OpenAI Chat Completions 和 Embeddings REST API 的服务。
 
@@ -40,6 +59,7 @@ transform:
     projection: >-
       *,
       AI_COMPLETE('completion_model', content, '总结输入内容') AS summary,
+      AI_SENTIMENT('completion_model', content) AS sentiment,
       AI_EMBED('embedding_model', content) AS embedding
 
 pipeline:
@@ -71,11 +91,11 @@ pipeline:
 | `model` | 是 | 发送给服务端的模型名称；`model-name` 作为废弃别名仍可使用。 |
 | `endpoint` | 是 | OpenAI-compatible 服务的 Base URL。 |
 | `api-key` | 是 | 请求认证使用的 Bearer Token。 |
-| `system-prompt` | 否 | 添加在 `AI_COMPLETE` 生成的 system prompt 之前。 |
+| `system-prompt` | 否 | 添加在所有文本 AI Function 的 prompt 之前。 |
 | `user-prompt` | 否 | 在输入之后追加一条 user message。 |
 | `temperature`、`top-p`、`stop`、`max-tokens` | 否 | 常用文本生成参数。 |
 | `presence-penalty`、`frequency-penalty`、`n`、`seed` | 否 | 其他文本生成参数。 |
-| `response-format` | 否 | 支持 `json_object`；AI completion 的结果必须是合法 JSON。 |
+| `response-format` | 否 | 支持 `json_object`；文本 AI Function 的结果必须是合法 JSON。 |
 | `content-type` | 否 | `text`（默认）或 `image_url`。 |
 | `dimension` | 否 | 请求的 embedding 维度。 |
 | `extra-header`、`extra-body` | 否 | JSON 对象格式的厂商自定义请求头或请求体字段。 |

docs/content/docs/core-concept/ai-model.md

@@ -24,11 +24,30 @@ under the License.
 
 # AI Model
 
-AI models can be used in transform expressions for text completion and embedding.
+AI models can be used in transform expressions for text generation, text analysis, and embedding.
+
+## AI Functions
+
+The model name must be a string constant that refers to a model declared in `pipeline.model`. Text functions require a model client that implements text generation, while `AI_EMBED` requires embedding support. The pipeline validates the referenced model capability before execution.
+
+All text functions return `VARIANT` values parsed from the model's JSON response.
+
+| Function | Description | JSON fields |
+|----------|-------------|-------------|
+| `AI_COMPLETE(model, input, system_prompt)` | Completes the input using a caller-provided system prompt. | `result` |
+| `AI_CLASSIFY(model, input, labels)` | Classifies the input into one of the provided labels. | `category`, `confidence` |
+| `AI_TRANSLATE(model, input, source_lang, target_lang)` | Translates the input. Use `auto` to detect the source language. | `translated_text`, `detected_language` |
+| `AI_SUMMARIZE(model, input, max_length)` | Summarizes the input within the requested character limit. | `summary` |
+| `AI_SENTIMENT(model, input)` | Analyzes sentiment. | `score`, `label`, `confidence` |
+| `AI_EXTRACT(model, input, schema)` | Extracts fields described by the schema string. | `extracted_json` |
+| `AI_MASK(model, input, entities)` | Masks the requested entity types. | `masked_text`, `detected_entities` |
+| `AI_EMBED(model, input)` | Creates an embedding vector. | Returns `ARRAY<FLOAT>` instead of JSON. |
+
+The specialized text functions use built-in English prompt templates, but their input may be in any language. If the input is `NULL`, the function returns `NULL` without invoking the model. A `NULL` model response also produces `NULL`. A non-null text response must be syntactically valid JSON; otherwise, record processing fails with an error that identifies the AI function. The runtime validates JSON syntax but does not validate the presence or types of individual response fields.
 
 ## OpenAI-compatible Model Client
 
-AI model clients can be referenced by the `AI_COMPLETE` and `AI_EMBED` transform functions. Add the model implementation JAR, such as `flink-cdc-pipeline-model-openai-compatible`, to the pipeline command with `--jar`.
+AI model clients can be referenced by the AI functions above. Add the model implementation JAR, such as `flink-cdc-pipeline-model-openai-compatible`, to the pipeline command with `--jar`.
 
 The OpenAI-compatible client supports chat completions and text embeddings against endpoints that implement the corresponding OpenAI REST APIs.
 
@@ -40,6 +59,7 @@ transform:
     projection: >-
       *,
       AI_COMPLETE('completion_model', content, 'Summarize the input') AS summary,
+      AI_SENTIMENT('completion_model', content) AS sentiment,
       AI_EMBED('embedding_model', content) AS embedding
 
 pipeline:
@@ -71,11 +91,11 @@ Do not store API keys in source control. Supply them through the secret-manageme
 | `model` | Yes | Model name sent to the endpoint. `model-name` is accepted as a deprecated alias. |
 | `endpoint` | Yes | Base URL of the OpenAI-compatible endpoint. |
 | `api-key` | Yes | Bearer token used to authenticate requests. |
-| `system-prompt` | No | Prompt prepended to the system prompt generated by `AI_COMPLETE`. |
+| `system-prompt` | No | Prompt prepended to every text AI function prompt. |
 | `user-prompt` | No | Additional user message appended after the input. |
 | `temperature`, `top-p`, `stop`, `max-tokens` | No | Common generation parameters. |
 | `presence-penalty`, `frequency-penalty`, `n`, `seed` | No | Additional generation parameters. |
-| `response-format` | No | `json_object` is supported. AI completion results must be valid JSON. |
+| `response-format` | No | `json_object` is supported. Text AI function results must be valid JSON. |
 | `content-type` | No | `text` (default) or `image_url`. |
 | `dimension` | No | Requested embedding dimension. |
 | `extra-header`, `extra-body` | No | Provider-specific headers or body fields encoded as JSON objects. |

flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java

@@ -41,6 +41,7 @@
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -66,7 +67,8 @@ public DataStream<Event> translatePreTransform(
         if (transforms.isEmpty()) {
             return input;
         }
-        validateModelReferences(transforms, models);
+        validateModelReferences(
+                transforms, models, getUserDefinedFunctionNames(udfFunctions, models));
         return input.transform(
                 "Transform:Schema",
                 new EventTypeInfo(),
@@ -144,7 +146,10 @@ public DataStream<Event> translatePostTransform(
                         .filter(ModelDef::isLegacy)
                         .map(this::modelToUDFTuple)
                         .collect(Collectors.toList()));
-        postTransformFunctionBuilder.addModelClients(loadModelClients(models, env));
+        Map<String, AiModelClient> modelClients = loadModelClients(models, env);
+        validateModelCapabilities(
+                transforms, modelClients, getUserDefinedFunctionNames(udfFunctions, models));
+        postTransformFunctionBuilder.addModelClients(modelClients);
         return input.transform(
                         "Transform:Data", new EventTypeInfo(), postTransformFunctionBuilder.build())
                 .uid(operatorUidGenerator.generateUid("post-transform"));
@@ -186,18 +191,48 @@ private Map<String, AiModelClient> loadModelClients(
         return clients;
     }
 
-    private void validateModelReferences(List<TransformDef> transforms, List<ModelDef> models) {
+    private void validateModelReferences(
+            List<TransformDef> transforms,
+            List<ModelDef> models,
+            Set<String> userDefinedFunctionNames) {
         Set<String> clientModelNames =
                 models.stream()
                         .filter(model -> !model.isLegacy())
                         .map(ModelDef::getName)
                         .collect(Collectors.toSet());
         for (TransformDef transform : transforms) {
             TransformParser.validateAiModelReferences(
-                    transform.getProjection(), transform.getFilter(), clientModelNames);
+                    transform.getProjection(),
+                    transform.getFilter(),
+                    clientModelNames,
+                    userDefinedFunctionNames);
         }
     }
 
+    private void validateModelCapabilities(
+            List<TransformDef> transforms,
+            Map<String, AiModelClient> modelClients,
+            Set<String> userDefinedFunctionNames) {
+        for (TransformDef transform : transforms) {
+            TransformParser.validateAiModelCapabilities(
+                    transform.getProjection(),
+                    transform.getFilter(),
+                    modelClients,
+                    userDefinedFunctionNames);
+        }
+    }
+
+    private Set<String> getUserDefinedFunctionNames(
+            List<UdfDef> udfFunctions, List<ModelDef> models) {
+        Set<String> functionNames = new HashSet<>();
+        udfFunctions.stream().map(UdfDef::getName).forEach(functionNames::add);
+        models.stream()
+                .filter(ModelDef::isLegacy)
+                .map(ModelDef::getName)
+                .forEach(functionNames::add);
+        return functionNames;
+    }
+
     private Tuple3<String, String, Map<String, String>> udfDefToUDFTuple(UdfDef udf) {
         return Tuple3.of(udf.getName(), udf.getClasspath(), udf.getOptions());
     }

flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineAiFunctionITCase.java

@@ -34,6 +34,7 @@
 import org.apache.flink.cdc.composer.definition.SinkDef;
 import org.apache.flink.cdc.composer.definition.SourceDef;
 import org.apache.flink.cdc.composer.definition.TransformDef;
+import org.apache.flink.cdc.composer.definition.UdfDef;
 import org.apache.flink.cdc.connectors.values.ValuesDatabase;
 import org.apache.flink.cdc.connectors.values.factory.ValuesDataFactory;
 import org.apache.flink.cdc.connectors.values.sink.ValuesDataSinkOptions;
@@ -127,6 +128,42 @@ void testAiCompleteInProjection() throws Exception {
                         "Dummy model closed.");
     }
 
+    @Test
+    void testSpecializedTextAiFunctionsInProjection() throws Exception {
+        String[] output =
+                runAiFunctionTest(
+                        "id, "
+                                + "AI_CLASSIFY('testModel', content, 'positive,negative') AS classified, "
+                                + "AI_TRANSLATE('testModel', content, 'auto', 'en') AS translated, "
+                                + "AI_SUMMARIZE('testModel', content, 100) AS summarized, "
+                                + "AI_SENTIMENT('testModel', content) AS sentiment, "
+                                + "AI_EXTRACT('testModel', content, 'name:string') AS extracted, "
+                                + "AI_MASK('testModel', content, 'name') AS masked",
+                        List.of(ModelDef.of("testModel", "dummy", Collections.emptyMap())));
+
+        assertThat(output)
+                .containsExactly(
+                        "CreateTableEvent{tableId=default_namespace.default_schema.mytable1, schema=columns={`id` INT NOT NULL,`classified` VARIANT,`translated` VARIANT,`summarized` VARIANT,`sentiment` VARIANT,`extracted` VARIANT,`masked` VARIANT}, primaryKeys=id, options=()}",
+                        "DataChangeEvent{tableId=default_namespace.default_schema.mytable1, before=[], after=[1, {\"category\":\"dummy\",\"confidence\":1}, {\"detected_language\":\"en\",\"translated_text\":\"dummy translation\"}, {\"summary\":\"dummy summary\"}, {\"confidence\":1,\"label\":\"neutral\",\"score\":0}, {\"extracted_json\":{\"name\":\"dummy\"}}, {\"detected_entities\":\"name\",\"masked_text\":\"d***y\"}], op=INSERT, meta=()}");
+    }
+
+    @Test
+    void testSameNamedUdfTakesPrecedenceOverAiFunction() throws Exception {
+        String[] output =
+                runAiFunctionTest(
+                        "id, AI_SENTIMENT(id) AS sentiment",
+                        List.of(ModelDef.of("unusedModel", "dummy", Collections.emptyMap())),
+                        List.of(
+                                new UdfDef(
+                                        "ai_sentiment",
+                                        "org.apache.flink.cdc.udf.examples.java.AddOneFunctionClass")));
+
+        assertThat(output)
+                .containsExactly(
+                        "CreateTableEvent{tableId=default_namespace.default_schema.mytable1, schema=columns={`id` INT NOT NULL,`sentiment` STRING}, primaryKeys=id, options=()}",
+                        "DataChangeEvent{tableId=default_namespace.default_schema.mytable1, before=[], after=[1, 2], op=INSERT, meta=()}");
+    }
+
     @Test
     void testAiEmbedInProjection() throws Exception {
         String[] output =
@@ -140,18 +177,24 @@ void testAiEmbedInProjection() throws Exception {
     }
 
     private String[] runAiFunctionTest(String projection, List<ModelDef> models) throws Exception {
+        return runAiFunctionTest(projection, models, Collections.emptyList());
+    }
+
+    private String[] runAiFunctionTest(
+            String projection, List<ModelDef> models, List<UdfDef> udfFunctions) throws Exception {
         URL modelJar = createDummyModelJar().toUri().toURL();
         ClassLoader originalClassLoader = Thread.currentThread().getContextClassLoader();
         try (URLClassLoader modelClassLoader =
                 new DummyModelClassLoader(modelJar, originalClassLoader)) {
             Thread.currentThread().setContextClassLoader(modelClassLoader);
-            return runAiFunctionTest(projection, models, modelJar);
+            return runAiFunctionTest(projection, models, udfFunctions, modelJar);
         } finally {
             Thread.currentThread().setContextClassLoader(originalClassLoader);
         }
     }
 
-    private String[] runAiFunctionTest(String projection, List<ModelDef> models, URL modelJar)
+    private String[] runAiFunctionTest(
+            String projection, List<ModelDef> models, List<UdfDef> udfFunctions, URL modelJar)
             throws Exception {
         FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
 
@@ -212,7 +255,7 @@ private String[] runAiFunctionTest(String projection, List<ModelDef> models, URL
                         sinkDef,
                         Collections.emptyList(),
                         Collections.singletonList(transformDef),
-                        Collections.emptyList(),
+                        udfFunctions,
                         models,
                         pipelineConfig);