This is an automated email from the ASF dual-hosted git repository.
yuxiqian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new e3f0caf44 [FLINK-39568][runtime] Introduce AI model client function
API (#4504)
e3f0caf44 is described below
commit e3f0caf4492756016a36a0c8c302898da458d2b1
Author: haruki <[email protected]>
AuthorDate: Thu Aug 13 21:45:54 2026 +0800
[FLINK-39568][runtime] Introduce AI model client function API (#4504)
---
.../cli/parser/YamlPipelineDefinitionParser.java | 96 ++++++++--
.../parser/YamlPipelineDefinitionParserTest.java | 109 ++++++++++++
.../flink/cdc/common/model/AiModelClient.java | 42 +++++
.../cdc/common/model/AiModelClientFactory.java | 35 ++++
.../flink/cdc/common/model/ModelContext.java | 36 ++++
.../common/model/abilities/SupportsEmbedding.java | 32 ++++
.../model/abilities/SupportsTextGeneration.java | 35 ++++
flink-cdc-composer/pom.xml | 8 +-
.../flink/cdc/composer/definition/ModelDef.java | 99 ++++++++---
.../flink/translator/TransformTranslator.java | 85 ++++++++-
.../flink/FlinkPipelineAiFunctionITCase.java | 197 +++++++++++++++++++++
.../flink-cdc-pipeline-e2e-tests/pom.xml | 10 ++
.../cdc/pipeline/tests/AiFunctionE2eITCase.java | 123 +++++++++++++
.../src/test/resources/rules/malformed.yaml | 17 --
.../{ => flink-cdc-pipeline-model-dummy}/pom.xml | 42 +----
.../flink/cdc/models/dummy/DummyModelClient.java | 61 +++++++
.../cdc/models/dummy/DummyModelClientFactory.java | 56 ++++++
.../org.apache.flink.cdc.common.factories.Factory | 16 ++
.../models/dummy/DummyModelClientFactoryTest.java | 97 ++++++++++
.../{ => flink-cdc-pipeline-model-legacy}/pom.xml | 7 +-
.../flink/cdc/runtime/model/ModelOptions.java | 0
.../flink/cdc/runtime/model/OpenAIChatModel.java | 0
.../cdc/runtime/model/OpenAIEmbeddingModel.java | 0
.../cdc/runtime/model/TestOpenAIChatModel.java | 0
.../runtime/model/TestOpenAIEmbeddingModel.java | 0
flink-cdc-pipeline-model/pom.xml | 65 +------
.../cdc/runtime/ai/AiEmbeddingFunctionDef.java | 50 ++++++
.../flink/cdc/runtime/ai/AiTextFunctionDef.java | 61 +++++++
.../cdc/runtime/functions/impl/AiFunctions.java | 85 +++++++++
.../operators/transform/PostTransformOperator.java | 47 ++++-
.../transform/PostTransformOperatorBuilder.java | 10 +-
.../transform/ProjectionColumnProcessor.java | 33 +++-
.../transform/TransformExpressionCompiler.java | 22 +++
.../transform/TransformFilterProcessor.java | 19 +-
.../transform/TransformProjectionProcessor.java | 9 +-
.../flink/cdc/runtime/parser/JaninoCompiler.java | 33 +++-
.../flink/cdc/runtime/parser/TransformParser.java | 56 +++++-
.../metadata/AiFunctionSqlOperatorTable.java | 110 ++++++++++++
.../runtime/functions/impl/AiFunctionsTest.java | 91 ++++++++++
.../cdc/runtime/parser/AiFunctionParserTest.java | 111 ++++++++++++
40 files changed, 1833 insertions(+), 172 deletions(-)
diff --git
a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java
b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java
index 831bb5e65..b5519afd5 100644
---
a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java
+++
b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java
@@ -46,6 +46,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -96,9 +97,11 @@ public class YamlPipelineDefinitionParser implements
PipelineDefinitionParser {
private static final String UDF_OPTIONS_KEY = "options";
// Model related keys
- private static final String MODEL_NAME_KEY = "model-name";
-
- private static final String MODEL_CLASS_NAME_KEY = "class-name";
+ private static final String MODEL_NAME_KEY = "name";
+ private static final String MODEL_TYPE_KEY = "type";
+ private static final String MODEL_OPTIONS_KEY = "options";
+ private static final String LEGACY_MODEL_NAME_KEY = "model-name";
+ private static final String LEGACY_MODEL_CLASS_NAME_KEY = "class-name";
public static final String TRANSFORM_PRIMARY_KEY_KEY = "primary-keys";
@@ -145,7 +148,6 @@ public class YamlPipelineDefinitionParser implements
PipelineDefinitionParser {
Optional.ofNullable(
((ObjectNode)
pipelineDefJsonNode.get(PIPELINE_KEY)).remove(MODEL_KEY))
- .map(node -> validateArray("model", node))
.ifPresent(node -> modelDefs.addAll(parseModels(node)));
}
@@ -428,24 +430,98 @@ public class YamlPipelineDefinitionParser implements
PipelineDefinitionParser {
} else {
modelDefs.add(convertJsonNodeToModelDef(modelsNode));
}
+ Set<String> seenNames = new HashSet<>();
+ for (ModelDef model : modelDefs) {
+ if (!seenNames.add(model.getName())) {
+ throw new IllegalArgumentException(
+ "Duplicate model name '" + model.getName() + "' in
pipeline definition.");
+ }
+ }
return modelDefs;
}
private ModelDef convertJsonNodeToModelDef(JsonNode modelNode) {
+ Preconditions.checkArgument(
+ modelNode instanceof ObjectNode,
+ "`model` in `pipeline` should be an object, but got %s",
+ modelNode);
+ ObjectNode node = ((ObjectNode) modelNode).deepCopy();
+ boolean usesNewFormat = node.has(MODEL_NAME_KEY) ||
node.has(MODEL_TYPE_KEY);
+ boolean usesLegacyFormat =
+ node.has(LEGACY_MODEL_NAME_KEY) &&
node.has(LEGACY_MODEL_CLASS_NAME_KEY);
+ Preconditions.checkArgument(
+ !(usesNewFormat && usesLegacyFormat),
+ "Model definition must use either name/type/options or
model-name/class-name, but not both: %s",
+ modelNode);
+
+ if (usesLegacyFormat) {
+ String modelName =
+ checkNotNull(
+ node.get(LEGACY_MODEL_NAME_KEY),
+ "Missing required field \"%s\" in `model`",
+ LEGACY_MODEL_NAME_KEY)
+ .asText();
+ validateModelName(modelName);
+ String className =
+ checkNotNull(
+ node.get(LEGACY_MODEL_CLASS_NAME_KEY),
+ "Missing required field \"%s\" in `model`",
+ LEGACY_MODEL_CLASS_NAME_KEY)
+ .asText();
+ Map<String, String> parameters =
+ mapper.convertValue(node, new TypeReference<Map<String,
String>>() {});
+ return new ModelDef(modelName, className, parameters);
+ }
+
String name =
checkNotNull(
- modelNode.get(MODEL_NAME_KEY),
+ node.remove(MODEL_NAME_KEY),
"Missing required field \"%s\" in `model`",
MODEL_NAME_KEY)
.asText();
- String model =
+ validateModelName(name);
+ String type =
checkNotNull(
- modelNode.get(MODEL_CLASS_NAME_KEY),
+ node.remove(MODEL_TYPE_KEY),
"Missing required field \"%s\" in `model`",
- MODEL_CLASS_NAME_KEY)
+ MODEL_TYPE_KEY)
.asText();
- Map<String, String> properties = mapper.convertValue(modelNode,
Map.class);
- return new ModelDef(name, model, properties);
+ Preconditions.checkArgument(
+ !StringUtils.isNullOrWhitespaceOnly(type),
+ "Model type must not be empty for model '%s'.",
+ name);
+
+ Map<String, String> options = new LinkedHashMap<>();
+ JsonNode optionsNode = node.remove(MODEL_OPTIONS_KEY);
+ if (optionsNode != null) {
+ Preconditions.checkArgument(
+ optionsNode instanceof ObjectNode,
+ "Model options must be an object, but got %s",
+ optionsNode);
+ options.putAll(
+ mapper.convertValue(optionsNode, new
TypeReference<Map<String, String>>() {}));
+ }
+ node.fields()
+ .forEachRemaining(
+ entry -> {
+ Preconditions.checkArgument(
+ !options.containsKey(entry.getKey()),
+ "Duplicate model option '%s' for model
'%s'.",
+ entry.getKey(),
+ name);
+ options.put(entry.getKey(),
entry.getValue().asText());
+ });
+ return ModelDef.of(name, type, options);
+ }
+
+ private void validateModelName(String name) {
+ Preconditions.checkArgument(
+ name.matches("[a-zA-Z_][a-zA-Z0-9_]*") &&
!name.startsWith("__"),
+ "Model name \"%s\" is not a valid identifier. "
+ + "It must start with a letter or underscore, "
+ + "contain only letters, digits, or underscores, "
+ + "and must not start with double underscores.",
+ name);
}
private void validateJsonNodeKeys(
diff --git
a/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParserTest.java
b/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParserTest.java
index 4ecc72d14..834831b6c 100644
---
a/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParserTest.java
+++
b/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParserTest.java
@@ -364,6 +364,115 @@ class YamlPipelineDefinitionParserTest {
.build())));
}
+ @Test
+ void testParsingFactoryBasedModel() throws Exception {
+ String yaml =
+ "source:\n"
+ + " type: values\n"
+ + "sink:\n"
+ + " type: values\n"
+ + "pipeline:\n"
+ + " model:\n"
+ + " name: completion_model\n"
+ + " type: dummy\n"
+ + " options:\n"
+ + " debug: true\n"
+ + " retries: 3\n";
+
+ PipelineDef pipelineDef =
+ new YamlPipelineDefinitionParser().parse(yaml, new
Configuration());
+
+ assertThat(pipelineDef.getModels())
+ .containsExactly(
+ ModelDef.of(
+ "completion_model",
+ "dummy",
+ ImmutableMap.of("debug", "true", "retries",
"3")));
+ }
+
+ @Test
+ void testParsingFactoryBasedModelWithFlatModelNameOption() throws
Exception {
+ String yaml =
+ "source:\n"
+ + " type: values\n"
+ + "sink:\n"
+ + " type: values\n"
+ + "pipeline:\n"
+ + " model:\n"
+ + " name: completion_model\n"
+ + " type: openai-compatible\n"
+ + " model-name: model-v1\n";
+
+ PipelineDef pipelineDef =
+ new YamlPipelineDefinitionParser().parse(yaml, new
Configuration());
+
+ assertThat(pipelineDef.getModels())
+ .containsExactly(
+ ModelDef.of(
+ "completion_model",
+ "openai-compatible",
+ Collections.singletonMap("model-name",
"model-v1")));
+ }
+
+ @Test
+ void testParsingLegacyAndFactoryBasedModelsTogether() throws Exception {
+ String yaml =
+ "source:\n"
+ + " type: values\n"
+ + "sink:\n"
+ + " type: values\n"
+ + "pipeline:\n"
+ + " model:\n"
+ + " - model-name: LEGACY_CHAT\n"
+ + " class-name: OpenAIChatModel\n"
+ + " openai.model: legacy-model\n"
+ + " - name: completion_model\n"
+ + " type: dummy\n"
+ + " debug: false\n";
+
+ PipelineDef pipelineDef =
+ new YamlPipelineDefinitionParser().parse(yaml, new
Configuration());
+
+ assertThat(pipelineDef.getModels()).hasSize(2);
+ assertThat(pipelineDef.getModels().get(0).isLegacy()).isTrue();
+ assertThat(pipelineDef.getModels().get(1))
+ .isEqualTo(
+ ModelDef.of(
+ "completion_model",
+ "dummy",
+ Collections.singletonMap("debug", "false")));
+ }
+
+ @Test
+ void testDuplicateAndInvalidModelNames() {
+ String duplicate =
+ "source:\n"
+ + " type: values\n"
+ + "sink:\n"
+ + " type: values\n"
+ + "pipeline:\n"
+ + " model:\n"
+ + " - name: duplicated\n"
+ + " type: dummy\n"
+ + " - model-name: duplicated\n"
+ + " class-name: OpenAIChatModel\n";
+ String invalid =
+ "source:\n"
+ + " type: values\n"
+ + "sink:\n"
+ + " type: values\n"
+ + "pipeline:\n"
+ + " model:\n"
+ + " name: invalid-name\n"
+ + " type: dummy\n";
+
+ YamlPipelineDefinitionParser parser = new
YamlPipelineDefinitionParser();
+ assertThatThrownBy(() -> parser.parse(duplicate, new Configuration()))
+ .hasMessage("Duplicate model name 'duplicated' in pipeline
definition.");
+ assertThatThrownBy(() -> parser.parse(invalid, new Configuration()))
+ .hasMessageContaining("is not a valid identifier");
+ }
+
private final PipelineDef fullDef =
new PipelineDef(
new SourceDef(
diff --git
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/AiModelClient.java
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/AiModelClient.java
new file mode 100644
index 000000000..0e46508bb
--- /dev/null
+++
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/AiModelClient.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.common.model;
+
+import org.apache.flink.cdc.common.annotation.Experimental;
+
+import java.io.Serializable;
+
+/**
+ * Marker interface for a runtime AI model client. Concrete capabilities are
declared via ability
+ * interfaces in {@code org.apache.flink.cdc.common.model.abilities}.
+ *
+ * <p>Implementations must be {@link Serializable} so that they can be
distributed across Flink task
+ * managers together with the operator that holds them.
+ */
+@Experimental
+public interface AiModelClient extends Serializable, AutoCloseable {
+
+ default void open() throws Exception {
+ // Do nothing
+ }
+
+ @Override
+ default void close() throws Exception {
+ // Do nothing
+ }
+}
diff --git
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/AiModelClientFactory.java
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/AiModelClientFactory.java
new file mode 100644
index 000000000..3f50c5295
--- /dev/null
+++
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/AiModelClientFactory.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.common.model;
+
+import org.apache.flink.cdc.common.annotation.Experimental;
+import org.apache.flink.cdc.common.factories.Factory;
+
+/**
+ * A {@link Factory} that creates {@link AiModelClient} instances. The factory
identifier maps to
+ * the {@code type} field of a pipeline model definition.
+ */
+@Experimental
+public interface AiModelClientFactory extends Factory {
+
+ /**
+ * Creates a new {@link AiModelClient} from the given context. The
returned client is serialized
+ * with the transform operator and opened on task managers.
+ */
+ AiModelClient createClient(ModelContext context);
+}
diff --git
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/ModelContext.java
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/ModelContext.java
new file mode 100644
index 000000000..a6b30250a
--- /dev/null
+++
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/ModelContext.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.common.model;
+
+import org.apache.flink.cdc.common.annotation.Experimental;
+
+import java.util.Map;
+
+/** Context passed to {@link AiModelClientFactory#createClient} at pipeline
assembly time. */
+@Experimental
+public interface ModelContext {
+
+ /** The logical name of this model as declared in the pipeline YAML. */
+ String getModelName();
+
+ /** Raw key/value options from the pipeline YAML {@code model.options}
block. */
+ Map<String, String> getOptions();
+
+ /** Class loader to use when loading implementation classes. */
+ ClassLoader getClassLoader();
+}
diff --git
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/abilities/SupportsEmbedding.java
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/abilities/SupportsEmbedding.java
new file mode 100644
index 000000000..d6c8fc7d1
--- /dev/null
+++
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/abilities/SupportsEmbedding.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.common.model.abilities;
+
+import org.apache.flink.cdc.common.annotation.Experimental;
+import org.apache.flink.cdc.common.model.AiModelClient;
+
+/**
+ * Ability interface for {@link AiModelClient} implementations that can
produce dense vector
+ * embeddings from text input.
+ */
+@Experimental
+public interface SupportsEmbedding {
+
+ /** Converts the given text into a dense float vector. */
+ float[] embed(String text);
+}
diff --git
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/abilities/SupportsTextGeneration.java
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/abilities/SupportsTextGeneration.java
new file mode 100644
index 000000000..eb1ea7362
--- /dev/null
+++
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/abilities/SupportsTextGeneration.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.common.model.abilities;
+
+import org.apache.flink.cdc.common.annotation.Experimental;
+import org.apache.flink.cdc.common.model.AiModelClient;
+
+/**
+ * Ability interface for {@link AiModelClient} implementations that can
perform chat-style text
+ * generation given a system prompt and a user input.
+ */
+@Experimental
+public interface SupportsTextGeneration {
+
+ /**
+ * Generates text based on a system-level prompt and a user-provided input
message. Returns a
+ * JSON string conforming to the output schema declared by the calling AI
function.
+ */
+ String generate(String systemPrompt, String userInput);
+}
diff --git a/flink-cdc-composer/pom.xml b/flink-cdc-composer/pom.xml
index 4161491ba..693d93e1d 100644
--- a/flink-cdc-composer/pom.xml
+++ b/flink-cdc-composer/pom.xml
@@ -86,6 +86,12 @@ limitations under the License.
<version>${project.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-pipeline-model-dummy</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
<!-- This is for testing Scala UDF.-->
<dependency>
@@ -139,4 +145,4 @@ limitations under the License.
</profile>
</profiles>
-</project>
\ No newline at end of file
+</project>
diff --git
a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/definition/ModelDef.java
b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/definition/ModelDef.java
index 21cc6befa..1439e329a 100644
---
a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/definition/ModelDef.java
+++
b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/definition/ModelDef.java
@@ -17,44 +17,89 @@
package org.apache.flink.cdc.composer.definition;
+import java.util.Collections;
import java.util.Map;
import java.util.Objects;
-/**
- * Common properties of model.
- *
- * <p>A transformation definition contains:
- *
- * <ul>
- * <li>modelName: The name of function.
- * <li>className: The model to transform data.
- * <li>parameters: The parameters that used to configure the model.
- * </ul>
- */
+/** Definition of an AI model declared in a pipeline. */
public class ModelDef {
- private final String modelName;
+ private final String name;
+
+ private final String type;
private final String className;
- private final Map<String, String> parameters;
+ private final Map<String, String> options;
+
+ private final boolean legacy;
+ /**
+ * Creates a legacy model definition backed by a model UDF class.
+ *
+ * @deprecated Use {@link #of(String, String, Map)} for factory-based AI
model clients.
+ */
+ @Deprecated
public ModelDef(String modelName, String className, Map<String, String>
parameters) {
- this.modelName = modelName;
+ this(modelName, null, className, parameters, true);
+ }
+
+ private ModelDef(
+ String name,
+ String type,
+ String className,
+ Map<String, String> options,
+ boolean legacy) {
+ this.name = name;
+ this.type = type;
this.className = className;
- this.parameters = parameters;
+ this.options = options == null ? Collections.emptyMap() : options;
+ this.legacy = legacy;
+ }
+
+ /** Creates a factory-based AI model client definition. */
+ public static ModelDef of(String name, String type, Map<String, String>
options) {
+ return new ModelDef(name, type, null, options, false);
}
+ public String getName() {
+ return name;
+ }
+
+ public String getType() {
+ return type;
+ }
+
+ public Map<String, String> getOptions() {
+ return options;
+ }
+
+ public boolean isLegacy() {
+ return legacy;
+ }
+
+ /**
+ * @deprecated Use {@link #getName()}.
+ */
+ @Deprecated
public String getModelName() {
- return modelName;
+ return name;
}
+ /**
+ * @deprecated Factory-based models use {@link #getType()}.
+ */
+ @Deprecated
public String getClassName() {
return className;
}
+ /**
+ * @deprecated Use {@link #getOptions()}.
+ */
+ @Deprecated
public Map<String, String> getParameters() {
- return parameters;
+ return options;
}
@Override
@@ -66,27 +111,29 @@ public class ModelDef {
return false;
}
ModelDef modelDef = (ModelDef) o;
- return Objects.equals(modelName, modelDef.modelName)
+ return Objects.equals(name, modelDef.name)
+ && Objects.equals(type, modelDef.type)
&& Objects.equals(className, modelDef.className)
- && Objects.equals(parameters, modelDef.parameters);
+ && Objects.equals(options, modelDef.options)
+ && legacy == modelDef.legacy;
}
@Override
public int hashCode() {
- return Objects.hash(modelName, className, parameters);
+ return Objects.hash(name, type, className, options, legacy);
}
@Override
public String toString() {
return "ModelDef{"
+ "name='"
- + modelName
- + '\''
- + ", model='"
- + className
+ + name
+ '\''
- + ", parameters="
- + parameters
+ + (legacy ? ", className='" + className + '\'' : ", type='" +
type + '\'')
+ + ", options="
+ + options
+ + ", legacy="
+ + legacy
+ '}';
}
}
diff --git
a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
index 6e296db4d..0ccb3afd8 100644
---
a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
+++
b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
@@ -18,20 +18,30 @@
package org.apache.flink.cdc.composer.flink.translator;
import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.cdc.common.configuration.Configuration;
import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.factories.FactoryHelper;
+import org.apache.flink.cdc.common.model.AiModelClient;
+import org.apache.flink.cdc.common.model.AiModelClientFactory;
+import org.apache.flink.cdc.common.model.ModelContext;
import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.composer.definition.ModelDef;
import org.apache.flink.cdc.composer.definition.TransformDef;
import org.apache.flink.cdc.composer.definition.UdfDef;
+import org.apache.flink.cdc.composer.utils.FactoryDiscoveryUtils;
import org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator;
import
org.apache.flink.cdc.runtime.operators.transform.PostTransformOperatorBuilder;
import org.apache.flink.cdc.runtime.operators.transform.PreTransformOperator;
import
org.apache.flink.cdc.runtime.operators.transform.PreTransformOperatorBuilder;
+import org.apache.flink.cdc.runtime.parser.TransformParser;
import org.apache.flink.cdc.runtime.typeutils.EventTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
+import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.stream.Collectors;
/**
@@ -53,6 +63,7 @@ public class TransformTranslator {
if (transforms.isEmpty()) {
return input;
}
+ validateModelReferences(transforms, models);
return input.transform(
"Transform:Schema",
new EventTypeInfo(),
@@ -85,7 +96,10 @@ public class TransformTranslator {
.map(this::udfDefToUDFTuple)
.collect(Collectors.toList()))
.addUdfFunctions(
-
models.stream().map(this::modelToUDFTuple).collect(Collectors.toList()));
+ models.stream()
+ .filter(ModelDef::isLegacy)
+ .map(this::modelToUDFTuple)
+ .collect(Collectors.toList()));
return preTransformFunctionBuilder.build();
}
@@ -120,7 +134,11 @@ public class TransformTranslator {
postTransformFunctionBuilder.addUdfFunctions(
udfFunctions.stream().map(this::udfDefToUDFTuple).collect(Collectors.toList()));
postTransformFunctionBuilder.addUdfFunctions(
-
models.stream().map(this::modelToUDFTuple).collect(Collectors.toList()));
+ models.stream()
+ .filter(ModelDef::isLegacy)
+ .map(this::modelToUDFTuple)
+ .collect(Collectors.toList()));
+ postTransformFunctionBuilder.addModelClients(loadModelClients(models));
return input.transform(
"Transform:Data", new EventTypeInfo(),
postTransformFunctionBuilder.build())
.uid(operatorUidGenerator.generateUid("post-transform"));
@@ -133,7 +151,70 @@ public class TransformTranslator {
model.getParameters());
}
+ private Map<String, AiModelClient> loadModelClients(List<ModelDef> models)
{
+ List<ModelDef> clientModels =
+ models.stream().filter(model ->
!model.isLegacy()).collect(Collectors.toList());
+ if (clientModels.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ ClassLoader classLoader =
Thread.currentThread().getContextClassLoader();
+ Map<String, AiModelClient> clients = new LinkedHashMap<>();
+ for (ModelDef model : clientModels) {
+ AiModelClientFactory factory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier(
+ model.getType(), AiModelClientFactory.class);
+ FactoryHelper.createFactoryHelper(
+ factory,
+ new FactoryHelper.DefaultContext(
+ Configuration.fromMap(model.getOptions()),
+ new Configuration(),
+ classLoader))
+ .validate();
+ ModelContext context = new DefaultModelContext(model, classLoader);
+ clients.put(model.getName(), factory.createClient(context));
+ }
+ return clients;
+ }
+
+ private void validateModelReferences(List<TransformDef> transforms,
List<ModelDef> models) {
+ 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);
+ }
+ }
+
private Tuple3<String, String, Map<String, String>>
udfDefToUDFTuple(UdfDef udf) {
return Tuple3.of(udf.getName(), udf.getClasspath(), udf.getOptions());
}
+
+ private static final class DefaultModelContext implements ModelContext {
+ private final ModelDef modelDef;
+ private final ClassLoader classLoader;
+
+ private DefaultModelContext(ModelDef modelDef, ClassLoader
classLoader) {
+ this.modelDef = modelDef;
+ this.classLoader = classLoader;
+ }
+
+ @Override
+ public String getModelName() {
+ return modelDef.getName();
+ }
+
+ @Override
+ public Map<String, String> getOptions() {
+ return modelDef.getOptions();
+ }
+
+ @Override
+ public ClassLoader getClassLoader() {
+ return classLoader;
+ }
+ }
}
diff --git
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineAiFunctionITCase.java
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineAiFunctionITCase.java
new file mode 100644
index 000000000..8aabfaffb
--- /dev/null
+++
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineAiFunctionITCase.java
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.composer.flink;
+
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.pipeline.PipelineOptions;
+import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.composer.PipelineExecution;
+import org.apache.flink.cdc.composer.definition.ModelDef;
+import org.apache.flink.cdc.composer.definition.PipelineDef;
+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.connectors.values.ValuesDatabase;
+import org.apache.flink.cdc.connectors.values.factory.ValuesDataFactory;
+import org.apache.flink.cdc.connectors.values.sink.ValuesDataSinkOptions;
+import org.apache.flink.cdc.connectors.values.source.ValuesDataSourceHelper;
+import org.apache.flink.cdc.connectors.values.source.ValuesDataSourceOptions;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static
org.apache.flink.configuration.CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Integration test for AI functions in the Flink pipeline. */
+class FlinkPipelineAiFunctionITCase {
+
+ private static final int MAX_PARALLELISM = 4;
+
+ private static final org.apache.flink.configuration.Configuration
MINI_CLUSTER_CONFIG =
+ new org.apache.flink.configuration.Configuration();
+
+ static {
+ MINI_CLUSTER_CONFIG.set(
+ ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL,
+ Collections.singletonList("org.apache.flink.cdc"));
+ }
+
+ @RegisterExtension
+ static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
+ new MiniClusterExtension(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(MAX_PARALLELISM)
+ .setConfiguration(MINI_CLUSTER_CONFIG)
+ .build());
+
+ private final PrintStream standardOut = System.out;
+ private final ByteArrayOutputStream outCaptor = new
ByteArrayOutputStream();
+
+ @BeforeEach
+ void init() {
+ System.setOut(new PrintStream(outCaptor));
+ ValuesDatabase.clear();
+ }
+
+ @AfterEach
+ void cleanup() {
+ System.setOut(standardOut);
+ }
+
+ @Test
+ void testAiCompleteInProjection() throws Exception {
+ String[] output =
+ runAiFunctionTest(
+ "id, content, AI_COMPLETE('testModel', content,
'Complete the text') AS completed",
+ List.of(
+ ModelDef.of(
+ "testModel",
+ "dummy",
+ Collections.singletonMap("debug",
"true"))));
+ assertThat(output)
+ .contains(
+ "Dummy model opened.",
+
"CreateTableEvent{tableId=default_namespace.default_schema.mytable1,
schema=columns={`id` INT NOT NULL,`content` STRING,`completed` VARIANT},
primaryKeys=id, options=()}",
+
"DataChangeEvent{tableId=default_namespace.default_schema.mytable1, before=[],
after=[1, I love this product, {\"result\":\"dummy response\"}], op=INSERT,
meta=()}",
+ "Dummy model closed.");
+ }
+
+ @Test
+ void testAiEmbedInProjection() throws Exception {
+ String[] output =
+ runAiFunctionTest(
+ "id, AI_EMBED('embedModel', content) AS embedding",
+ List.of(ModelDef.of("embedModel", "dummy",
Collections.emptyMap())));
+ assertThat(output)
+ .containsExactly(
+
"CreateTableEvent{tableId=default_namespace.default_schema.mytable1,
schema=columns={`id` INT NOT NULL,`embedding` ARRAY<FLOAT>}, primaryKeys=id,
options=()}",
+
"DataChangeEvent{tableId=default_namespace.default_schema.mytable1, before=[],
after=[1, [3.0, 1.0, 4.0, 1.0, 5.0, 9.0, 2.0, 6.0]], op=INSERT, meta=()}");
+ }
+
+ private String[] runAiFunctionTest(String projection, List<ModelDef>
models) throws Exception {
+ FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
+
+ // Source: one table with a single row
+ TableId tableId = TableId.tableId("default_namespace",
"default_schema", "mytable1");
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT())
+ .physicalColumn("content", DataTypes.STRING())
+ .primaryKey("id")
+ .build();
+ BinaryRecordDataGenerator generator =
+ new
BinaryRecordDataGenerator(schema.getColumnDataTypes().toArray(new DataType[0]));
+
+ List<Event> events = new ArrayList<>();
+ events.add(new CreateTableEvent(tableId, schema));
+ events.add(
+ DataChangeEvent.insertEvent(
+ tableId,
+ generator.generate(
+ new Object[] {
+ 1, BinaryStringData.fromString("I love
this product")
+ })));
+
ValuesDataSourceHelper.setSourceEvents(Collections.singletonList(events));
+
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ ValuesDataSourceOptions.EVENT_SET_ID,
+ ValuesDataSourceHelper.EventSetId.CUSTOM_SOURCE_EVENTS);
+ SourceDef sourceDef =
+ new SourceDef(ValuesDataFactory.IDENTIFIER, "Value Source",
sourceConfig);
+
+ // Sink
+ Configuration sinkConfig = new Configuration();
+ sinkConfig.set(ValuesDataSinkOptions.MATERIALIZED_IN_MEMORY, true);
+ SinkDef sinkDef = new SinkDef(ValuesDataFactory.IDENTIFIER, "Value
Sink", sinkConfig);
+
+ // Transform
+ TransformDef transformDef =
+ new TransformDef(
+ "default_namespace.default_schema.mytable1",
+ projection,
+ null,
+ "id",
+ null,
+ null,
+ null,
+ null);
+
+ // Pipeline
+ Configuration pipelineConfig = new Configuration();
+ pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR,
SchemaChangeBehavior.EVOLVE);
+ PipelineDef pipelineDef =
+ new PipelineDef(
+ sourceDef,
+ sinkDef,
+ Collections.emptyList(),
+ Collections.singletonList(transformDef),
+ Collections.emptyList(),
+ models,
+ pipelineConfig);
+
+ // Execute & capture output
+ PipelineExecution execution = composer.compose(pipelineDef);
+ execution.execute();
+
+ return outCaptor.toString().trim().split("\n");
+ }
+}
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/pom.xml
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/pom.xml
index 029360af4..52e3dc6aa 100644
--- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/pom.xml
+++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/pom.xml
@@ -564,6 +564,16 @@ limitations under the License.
</outputDirectory>
</artifactItem>
+ <artifactItem>
+ <groupId>org.apache.flink</groupId>
+
<artifactId>flink-cdc-pipeline-model-dummy</artifactId>
+ <version>${project.version}</version>
+ <destFileName>dummy-model.jar</destFileName>
+ <type>jar</type>
+
<outputDirectory>${project.build.directory}/dependencies
+ </outputDirectory>
+ </artifactItem>
+
<artifactItem>
<groupId>org.apache.flink</groupId>
<artifactId>flink-cdc-pipeline-connector-mysql</artifactId>
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/AiFunctionE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/AiFunctionE2eITCase.java
new file mode 100644
index 000000000..8bdb7cb9a
--- /dev/null
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/AiFunctionE2eITCase.java
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.pipeline.tests;
+
+import org.apache.flink.cdc.common.test.utils.TestUtils;
+import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
+
+import org.junit.jupiter.api.Test;
+
+import java.nio.file.Path;
+import java.time.Duration;
+
+/** E2e tests for AI functions with the dummy model SPI. */
+class AiFunctionE2eITCase extends PipelineTestEnvironment {
+
+ private static final String TABLE_1 =
"default_namespace.default_schema.table1";
+ private static final String TABLE_2 =
"default_namespace.default_schema.table2";
+ private static final String DUMMY_JSON = "{\"result\":\"dummy response\"}";
+ private static final String DUMMY_EMBEDDING = "[3.0, 1.0, 4.0, 1.0, 5.0,
9.0, 2.0, 6.0]";
+
+ @Test
+ void testAiFunctionsWithDummyModel() throws Exception {
+ String pipelineJob =
+ "source:\n"
+ + " type: values\n"
+ + " event-set.id: SINGLE_SPLIT_MULTI_TABLES\n"
+ + "\n"
+ + "sink:\n"
+ + " type: values\n"
+ + "\n"
+ + "transform:\n"
+ + " - source-table: "
+ + TABLE_1
+ + "\n"
+ + " projection: col1, AI_COMPLETE('myModel', col1,
'Complete it') AS completed\n"
+ + " - source-table: "
+ + TABLE_2
+ + "\n"
+ + " projection: col1, AI_EMBED('myModel', col1) AS
embedding\n"
+ + "\n"
+ + "pipeline:\n"
+ + " parallelism: 1\n"
+ + " schema.change.behavior: evolve\n"
+ + " model:\n"
+ + " name: myModel\n"
+ + " type: dummy\n"
+ + " options:\n"
+ + " debug: true\n";
+
+ Path dummyModelJar = TestUtils.getResource("dummy-model.jar");
+ submitPipelineJob(pipelineJob, dummyModelJar);
+ waitUntilJobFinished(Duration.ofMinutes(3));
+
+ validateResult("Successfully opened AI model client 'myModel'.");
+ validateResult(
+ "CreateTableEvent{tableId="
+ + TABLE_1
+ + ", schema=columns={`col1` STRING NOT
NULL,`completed` VARIANT}, primaryKeys=col1, options=()}",
+ "DataChangeEvent{tableId="
+ + TABLE_1
+ + ", before=[], after=[1, "
+ + DUMMY_JSON
+ + "], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId="
+ + TABLE_1
+ + ", before=[], after=[2, "
+ + DUMMY_JSON
+ + "], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId="
+ + TABLE_1
+ + ", before=[], after=[3, "
+ + DUMMY_JSON
+ + "], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId="
+ + TABLE_1
+ + ", before=[1, "
+ + DUMMY_JSON
+ + "], after=[], op=DELETE, meta=()}",
+ "DataChangeEvent{tableId="
+ + TABLE_1
+ + ", before=[2, "
+ + DUMMY_JSON
+ + "], after=[2, "
+ + DUMMY_JSON
+ + "], op=UPDATE, meta=()}");
+
+ validateResult(
+ "CreateTableEvent{tableId="
+ + TABLE_2
+ + ", schema=columns={`col1` STRING NOT
NULL,`embedding` ARRAY<FLOAT>}, primaryKeys=col1, options=()}",
+ "DataChangeEvent{tableId="
+ + TABLE_2
+ + ", before=[], after=[1, "
+ + DUMMY_EMBEDDING
+ + "], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId="
+ + TABLE_2
+ + ", before=[], after=[2, "
+ + DUMMY_EMBEDDING
+ + "], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId="
+ + TABLE_2
+ + ", before=[], after=[3, "
+ + DUMMY_EMBEDDING
+ + "], op=INSERT, meta=()}");
+ validateResult("Successfully closed AI model client 'myModel'.");
+ }
+}
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/resources/rules/malformed.yaml
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/resources/rules/malformed.yaml
index ffed7ad9b..6fe4a61e1 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/resources/rules/malformed.yaml
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/resources/rules/malformed.yaml
@@ -55,20 +55,3 @@ steps:
error: |
YAML UDF block is expecting an array children, but got an OBJECT
({"name":"addone","classpath":"org.apache.flink.cdc.udf.examples.%s.AddOneFunctionClass"}).
Perhaps you missed a dash prefix `-`?
- # Models not an array
- - type: submit
- yaml: |
- source:
- type: values
- sink:
- type: values
- pipeline:
- model:
- model-name: GET_EMBEDDING
- class-name: OpenAIEmbeddingModel
- openai.model: text-embedding-3-small
- openai.host: https://xxxx
- openai.apikey: abcd1234
- error: |
- YAML model block is expecting an array children, but got an OBJECT
({"model-name":"GET_EMBEDDING","class-name":"OpenAIEmbeddingModel","openai.model":"text-embedding-3-small","openai.host":"https://xxxx","openai.apikey":"abcd1234"}).
- Perhaps you missed a dash prefix `-`?
diff --git a/flink-cdc-pipeline-model/pom.xml
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-dummy/pom.xml
similarity index 59%
copy from flink-cdc-pipeline-model/pom.xml
copy to flink-cdc-pipeline-model/flink-cdc-pipeline-model-dummy/pom.xml
index e7dba7f6f..d72d93e9a 100644
--- a/flink-cdc-pipeline-model/pom.xml
+++ b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-dummy/pom.xml
@@ -19,16 +19,13 @@ limitations under the License.
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
<parent>
- <artifactId>flink-cdc-parent</artifactId>
+ <artifactId>flink-cdc-pipeline-model-parent</artifactId>
<groupId>org.apache.flink</groupId>
<version>${revision}</version>
</parent>
<modelVersion>4.0.0</modelVersion>
- <artifactId>flink-cdc-pipeline-model</artifactId>
- <properties>
- <langchain4j.version>0.23.0</langchain4j.version>
- </properties>
+ <artifactId>flink-cdc-pipeline-model-dummy</artifactId>
<dependencies>
<dependency>
@@ -49,39 +46,6 @@ limitations under the License.
</exclusion>
</exclusions>
</dependency>
- <dependency>
- <groupId>dev.langchain4j</groupId>
- <artifactId>langchain4j</artifactId>
- <version>${langchain4j.version}</version>
- </dependency>
- <dependency>
- <groupId>dev.langchain4j</groupId>
- <artifactId>langchain4j-open-ai</artifactId>
- <version>${langchain4j.version}</version>
- </dependency>
- <dependency>
- <groupId>com.theokanning.openai-gpt3-java</groupId>
- <artifactId>service</artifactId>
- <version>0.12.0</version>
- </dependency>
</dependencies>
-
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-jar-plugin</artifactId>
- <executions>
- <execution>
- <id>test-jar</id>
- <goals>
- <goal>test-jar</goal>
- </goals>
- </execution>
- </executions>
- </plugin>
- </plugins>
- </build>
-
-</project>
\ No newline at end of file
+</project>
diff --git
a/flink-cdc-pipeline-model/flink-cdc-pipeline-model-dummy/src/main/java/org/apache/flink/cdc/models/dummy/DummyModelClient.java
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-dummy/src/main/java/org/apache/flink/cdc/models/dummy/DummyModelClient.java
new file mode 100644
index 000000000..f4c2b7444
--- /dev/null
+++
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-dummy/src/main/java/org/apache/flink/cdc/models/dummy/DummyModelClient.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.models.dummy;
+
+import org.apache.flink.cdc.common.model.AiModelClient;
+import org.apache.flink.cdc.common.model.abilities.SupportsEmbedding;
+import org.apache.flink.cdc.common.model.abilities.SupportsTextGeneration;
+
+/** Deterministic AI model client used by tests. */
+public class DummyModelClient implements AiModelClient,
SupportsTextGeneration, SupportsEmbedding {
+
+ private static final long serialVersionUID = 1L;
+
+ private final boolean debug;
+
+ public DummyModelClient(boolean debug) {
+ this.debug = debug;
+ }
+
+ @Override
+ public String generate(String systemPrompt, String userInput) {
+ if (debug) {
+ System.out.printf("Received prompt: %s%nUser input: %s%n",
systemPrompt, userInput);
+ }
+ return "{\"result\":\"dummy response\"}";
+ }
+
+ @Override
+ public float[] embed(String text) {
+ return new float[] {3f, 1f, 4f, 1f, 5f, 9f, 2f, 6f};
+ }
+
+ @Override
+ public void open() {
+ if (debug) {
+ System.out.println("Dummy model opened.");
+ }
+ }
+
+ @Override
+ public void close() {
+ if (debug) {
+ System.out.println("Dummy model closed.");
+ }
+ }
+}
diff --git
a/flink-cdc-pipeline-model/flink-cdc-pipeline-model-dummy/src/main/java/org/apache/flink/cdc/models/dummy/DummyModelClientFactory.java
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-dummy/src/main/java/org/apache/flink/cdc/models/dummy/DummyModelClientFactory.java
new file mode 100644
index 000000000..4bf89de02
--- /dev/null
+++
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-dummy/src/main/java/org/apache/flink/cdc/models/dummy/DummyModelClientFactory.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.models.dummy;
+
+import org.apache.flink.cdc.common.configuration.ConfigOption;
+import org.apache.flink.cdc.common.configuration.ConfigOptions;
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.model.AiModelClient;
+import org.apache.flink.cdc.common.model.AiModelClientFactory;
+import org.apache.flink.cdc.common.model.ModelContext;
+
+import java.util.Collections;
+import java.util.Set;
+
+/** Factory for the test-only {@link DummyModelClient}. */
+public class DummyModelClientFactory implements AiModelClientFactory {
+
+ private static final ConfigOption<Boolean> DEBUG =
+ ConfigOptions.key("debug").booleanType().defaultValue(false);
+
+ @Override
+ public String identifier() {
+ return "dummy";
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return Set.of(DEBUG);
+ }
+
+ @Override
+ public AiModelClient createClient(ModelContext context) {
+ boolean debug = Configuration.fromMap(context.getOptions()).get(DEBUG);
+ return new DummyModelClient(debug);
+ }
+}
diff --git
a/flink-cdc-pipeline-model/flink-cdc-pipeline-model-dummy/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-dummy/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
new file mode 100644
index 000000000..c1ed9c43f
--- /dev/null
+++
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-dummy/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+org.apache.flink.cdc.models.dummy.DummyModelClientFactory
diff --git
a/flink-cdc-pipeline-model/flink-cdc-pipeline-model-dummy/src/test/java/org/apache/flink/cdc/models/dummy/DummyModelClientFactoryTest.java
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-dummy/src/test/java/org/apache/flink/cdc/models/dummy/DummyModelClientFactoryTest.java
new file mode 100644
index 000000000..eaee34182
--- /dev/null
+++
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-dummy/src/test/java/org/apache/flink/cdc/models/dummy/DummyModelClientFactoryTest.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.models.dummy;
+
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.factories.FactoryHelper;
+import org.apache.flink.cdc.common.model.AiModelClient;
+import org.apache.flink.cdc.common.model.ModelContext;
+import org.apache.flink.table.api.ValidationException;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link DummyModelClientFactory}. */
+class DummyModelClientFactoryTest {
+
+ @Test
+ void testFactoryHelperValidationAndClientCreation() {
+ DummyModelClientFactory factory = new DummyModelClientFactory();
+ ClassLoader classLoader =
Thread.currentThread().getContextClassLoader();
+ Configuration options =
Configuration.fromMap(Collections.singletonMap("debug", "true"));
+
+ FactoryHelper.createFactoryHelper(
+ factory,
+ new FactoryHelper.DefaultContext(options, new
Configuration(), classLoader))
+ .validate();
+ AiModelClient client =
+ factory.createClient(new TestingModelContext(options.toMap(),
classLoader));
+
+ assertThat(client).isInstanceOf(DummyModelClient.class);
+ }
+
+ @Test
+ void testUnknownOptionIsRejected() {
+ DummyModelClientFactory factory = new DummyModelClientFactory();
+ Configuration options =
+
Configuration.fromMap(Collections.singletonMap("unknown-option", "value"));
+
+ assertThatThrownBy(
+ () ->
+ FactoryHelper.createFactoryHelper(
+ factory,
+ new
FactoryHelper.DefaultContext(
+ options,
+ new Configuration(),
+ Thread.currentThread()
+
.getContextClassLoader()))
+ .validate())
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Unsupported options");
+ }
+
+ private static class TestingModelContext implements ModelContext {
+ private final Map<String, String> options;
+ private final ClassLoader classLoader;
+
+ private TestingModelContext(Map<String, String> options, ClassLoader
classLoader) {
+ this.options = options;
+ this.classLoader = classLoader;
+ }
+
+ @Override
+ public String getModelName() {
+ return "dummy-model";
+ }
+
+ @Override
+ public Map<String, String> getOptions() {
+ return options;
+ }
+
+ @Override
+ public ClassLoader getClassLoader() {
+ return classLoader;
+ }
+ }
+}
diff --git a/flink-cdc-pipeline-model/pom.xml
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-legacy/pom.xml
similarity index 95%
copy from flink-cdc-pipeline-model/pom.xml
copy to flink-cdc-pipeline-model/flink-cdc-pipeline-model-legacy/pom.xml
index e7dba7f6f..e0b0c2fd9 100644
--- a/flink-cdc-pipeline-model/pom.xml
+++ b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-legacy/pom.xml
@@ -19,13 +19,15 @@ limitations under the License.
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
<parent>
- <artifactId>flink-cdc-parent</artifactId>
+ <artifactId>flink-cdc-pipeline-model-parent</artifactId>
<groupId>org.apache.flink</groupId>
<version>${revision}</version>
</parent>
<modelVersion>4.0.0</modelVersion>
+ <!-- Keep the original artifact coordinates for backward compatibility. -->
<artifactId>flink-cdc-pipeline-model</artifactId>
+
<properties>
<langchain4j.version>0.23.0</langchain4j.version>
</properties>
@@ -66,7 +68,6 @@ limitations under the License.
</dependency>
</dependencies>
-
<build>
<plugins>
<plugin>
@@ -84,4 +85,4 @@ limitations under the License.
</plugins>
</build>
-</project>
\ No newline at end of file
+</project>
diff --git
a/flink-cdc-pipeline-model/src/main/java/org/apache/flink/cdc/runtime/model/ModelOptions.java
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-legacy/src/main/java/org/apache/flink/cdc/runtime/model/ModelOptions.java
similarity index 100%
rename from
flink-cdc-pipeline-model/src/main/java/org/apache/flink/cdc/runtime/model/ModelOptions.java
rename to
flink-cdc-pipeline-model/flink-cdc-pipeline-model-legacy/src/main/java/org/apache/flink/cdc/runtime/model/ModelOptions.java
diff --git
a/flink-cdc-pipeline-model/src/main/java/org/apache/flink/cdc/runtime/model/OpenAIChatModel.java
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-legacy/src/main/java/org/apache/flink/cdc/runtime/model/OpenAIChatModel.java
similarity index 100%
rename from
flink-cdc-pipeline-model/src/main/java/org/apache/flink/cdc/runtime/model/OpenAIChatModel.java
rename to
flink-cdc-pipeline-model/flink-cdc-pipeline-model-legacy/src/main/java/org/apache/flink/cdc/runtime/model/OpenAIChatModel.java
diff --git
a/flink-cdc-pipeline-model/src/main/java/org/apache/flink/cdc/runtime/model/OpenAIEmbeddingModel.java
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-legacy/src/main/java/org/apache/flink/cdc/runtime/model/OpenAIEmbeddingModel.java
similarity index 100%
rename from
flink-cdc-pipeline-model/src/main/java/org/apache/flink/cdc/runtime/model/OpenAIEmbeddingModel.java
rename to
flink-cdc-pipeline-model/flink-cdc-pipeline-model-legacy/src/main/java/org/apache/flink/cdc/runtime/model/OpenAIEmbeddingModel.java
diff --git
a/flink-cdc-pipeline-model/src/test/java/org/apache/flink/cdc/runtime/model/TestOpenAIChatModel.java
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-legacy/src/test/java/org/apache/flink/cdc/runtime/model/TestOpenAIChatModel.java
similarity index 100%
rename from
flink-cdc-pipeline-model/src/test/java/org/apache/flink/cdc/runtime/model/TestOpenAIChatModel.java
rename to
flink-cdc-pipeline-model/flink-cdc-pipeline-model-legacy/src/test/java/org/apache/flink/cdc/runtime/model/TestOpenAIChatModel.java
diff --git
a/flink-cdc-pipeline-model/src/test/java/org/apache/flink/cdc/runtime/model/TestOpenAIEmbeddingModel.java
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-legacy/src/test/java/org/apache/flink/cdc/runtime/model/TestOpenAIEmbeddingModel.java
similarity index 100%
rename from
flink-cdc-pipeline-model/src/test/java/org/apache/flink/cdc/runtime/model/TestOpenAIEmbeddingModel.java
rename to
flink-cdc-pipeline-model/flink-cdc-pipeline-model-legacy/src/test/java/org/apache/flink/cdc/runtime/model/TestOpenAIEmbeddingModel.java
diff --git a/flink-cdc-pipeline-model/pom.xml b/flink-cdc-pipeline-model/pom.xml
index e7dba7f6f..1a3e398b7 100644
--- a/flink-cdc-pipeline-model/pom.xml
+++ b/flink-cdc-pipeline-model/pom.xml
@@ -25,63 +25,12 @@ limitations under the License.
</parent>
<modelVersion>4.0.0</modelVersion>
- <artifactId>flink-cdc-pipeline-model</artifactId>
- <properties>
- <langchain4j.version>0.23.0</langchain4j.version>
- </properties>
+ <artifactId>flink-cdc-pipeline-model-parent</artifactId>
+ <packaging>pom</packaging>
- <dependencies>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-cdc-common</artifactId>
- <version>${project.version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-test-utils-junit</artifactId>
- <version>${flink.version}</version>
- <scope>test</scope>
- <exclusions>
- <exclusion>
- <groupId>org.testcontainers</groupId>
- <artifactId>testcontainers</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>dev.langchain4j</groupId>
- <artifactId>langchain4j</artifactId>
- <version>${langchain4j.version}</version>
- </dependency>
- <dependency>
- <groupId>dev.langchain4j</groupId>
- <artifactId>langchain4j-open-ai</artifactId>
- <version>${langchain4j.version}</version>
- </dependency>
- <dependency>
- <groupId>com.theokanning.openai-gpt3-java</groupId>
- <artifactId>service</artifactId>
- <version>0.12.0</version>
- </dependency>
- </dependencies>
+ <modules>
+ <module>flink-cdc-pipeline-model-legacy</module>
+ <module>flink-cdc-pipeline-model-dummy</module>
+ </modules>
-
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-jar-plugin</artifactId>
- <executions>
- <execution>
- <id>test-jar</id>
- <goals>
- <goal>test-jar</goal>
- </goals>
- </execution>
- </executions>
- </plugin>
- </plugins>
- </build>
-
-</project>
\ No newline at end of file
+</project>
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/ai/AiEmbeddingFunctionDef.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/ai/AiEmbeddingFunctionDef.java
new file mode 100644
index 000000000..3554f604c
--- /dev/null
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/ai/AiEmbeddingFunctionDef.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.runtime.ai;
+
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+
+/** Built-in AI embedding function definitions with configurable input and
output types. */
+public enum AiEmbeddingFunctionDef {
+ AI_EMBED("AI_EMBED", DataTypes.STRING(),
DataTypes.ARRAY(DataTypes.FLOAT()));
+
+ private final String functionName;
+ private final DataType inputType;
+ private final DataType outputType;
+
+ AiEmbeddingFunctionDef(String functionName, DataType inputType, DataType
outputType) {
+ this.functionName = functionName;
+ this.inputType = inputType;
+ this.outputType = outputType;
+ }
+
+ public String getFunctionName() {
+ return functionName;
+ }
+
+ /** The type of the input value (e.g. STRING for text embedding). */
+ public DataType getInputType() {
+ return inputType;
+ }
+
+ /** The type of the output value (e.g. ARRAY<FLOAT> for vector
embedding). */
+ public DataType getOutputType() {
+ return outputType;
+ }
+}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/ai/AiTextFunctionDef.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/ai/AiTextFunctionDef.java
new file mode 100644
index 000000000..74687617a
--- /dev/null
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/ai/AiTextFunctionDef.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.runtime.ai;
+
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.common.types.RowType;
+
+/** Built-in AI text generation function definitions. */
+public enum AiTextFunctionDef {
+ AI_COMPLETE(
+ "AI_COMPLETE",
+ RowType.of(new DataType[] {DataTypes.STRING()}, new String[]
{"systemPrompt"}),
+ RowType.of(new DataType[] {DataTypes.STRING()}, new String[]
{"result"}),
+ "%s\n");
+
+ private final String functionName;
+ private final RowType inputType;
+ private final RowType outputType;
+ private final String promptTemplate;
+
+ AiTextFunctionDef(
+ String functionName, RowType inputType, RowType outputType, String
promptTemplate) {
+ this.functionName = functionName;
+ this.inputType = inputType;
+ this.outputType = outputType;
+ this.promptTemplate = promptTemplate;
+ }
+
+ public String getFunctionName() {
+ return functionName;
+ }
+
+ /** Returns the parameter types after the model and input arguments. */
+ public RowType getInputType() {
+ return inputType;
+ }
+
+ public RowType getOutputType() {
+ return outputType;
+ }
+
+ public String buildPrompt(Object... args) {
+ return String.format(promptTemplate, args);
+ }
+}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/AiFunctions.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/AiFunctions.java
new file mode 100644
index 000000000..7d17c86c4
--- /dev/null
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/AiFunctions.java
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.runtime.functions.impl;
+
+import org.apache.flink.cdc.common.model.AiModelClient;
+import org.apache.flink.cdc.common.model.abilities.SupportsEmbedding;
+import org.apache.flink.cdc.common.model.abilities.SupportsTextGeneration;
+import org.apache.flink.cdc.common.types.RowType;
+import org.apache.flink.cdc.common.types.variant.BinaryVariant;
+import org.apache.flink.cdc.common.types.variant.BinaryVariantInternalBuilder;
+import org.apache.flink.cdc.runtime.ai.AiTextFunctionDef;
+
+import org.apache.flink.shaded.guava31.com.google.common.primitives.Floats;
+
+import java.io.IOException;
+import java.util.List;
+
+/** Built-in AI functions available to transform expressions. */
+public class AiFunctions {
+
+ private AiFunctions() {}
+
+ public static BinaryVariant aiComplete(AiModelClient model, String input,
String systemPrompt) {
+ if (!(model instanceof SupportsTextGeneration)) {
+ throw new UnsupportedOperationException(
+ "Model " + model.getClass().getName() + " does not support
text generation");
+ }
+
+ AiTextFunctionDef function = AiTextFunctionDef.AI_COMPLETE;
+ String prompt =
+ function.buildPrompt(systemPrompt)
+ + "\n"
+ + buildOutputSchemaHint(function.getOutputType());
+ String json = ((SupportsTextGeneration) model).generate(prompt, input);
+ if (json == null) {
+ return null;
+ }
+ try {
+ return BinaryVariantInternalBuilder.parseJson(json, false);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to parse AI response as JSON: "
+ json, e);
+ }
+ }
+
+ public static List<Float> aiEmbed(AiModelClient model, String input) {
+ if (!(model instanceof SupportsEmbedding)) {
+ throw new UnsupportedOperationException(
+ "Model " + model.getClass().getName() + " does not support
embedding");
+ }
+ float[] embedding = ((SupportsEmbedding) model).embed(input);
+ return embedding == null ? null : Floats.asList(embedding);
+ }
+
+ private static String buildOutputSchemaHint(RowType outputType) {
+ StringBuilder builder = new StringBuilder("Return valid JSON with this
shape:\n{\n");
+ List<String> fieldNames = outputType.getFieldNames();
+ for (int i = 0; i < fieldNames.size(); i++) {
+ builder.append(" \"")
+ .append(fieldNames.get(i))
+ .append("\": <")
+ .append(fieldNames.get(i))
+ .append(">");
+ if (i < fieldNames.size() - 1) {
+ builder.append(',');
+ }
+ builder.append('\n');
+ }
+ return builder.append('}').toString();
+ }
+}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperator.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperator.java
index 0a8b63703..9c6552fbe 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperator.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperator.java
@@ -29,6 +29,7 @@ import org.apache.flink.cdc.common.event.DataChangeEvent;
import org.apache.flink.cdc.common.event.Event;
import org.apache.flink.cdc.common.event.SchemaChangeEvent;
import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.model.AiModelClient;
import org.apache.flink.cdc.common.schema.Schema;
import org.apache.flink.cdc.common.schema.Selectors;
import org.apache.flink.cdc.common.udf.UserDefinedFunctionContext;
@@ -41,6 +42,7 @@ import
org.apache.flink.cdc.runtime.typeutils.BinaryInternalObjectConverter;
import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.shaded.guava31.com.google.common.cache.CacheBuilder;
import org.apache.flink.shaded.guava31.com.google.common.cache.CacheLoader;
@@ -48,6 +50,9 @@ import
org.apache.flink.shaded.guava31.com.google.common.cache.LoadingCache;
import
org.apache.flink.shaded.guava31.com.google.common.collect.HashBasedTable;
import org.apache.flink.shaded.guava31.com.google.common.collect.Table;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import javax.annotation.Nullable;
import java.io.Serializable;
@@ -69,6 +74,7 @@ public class PostTransformOperator extends
AbstractStreamOperatorAdapter<Event>
implements OneInputStreamOperator<Event, Event>, Serializable {
private static final long serialVersionUID = 1L;
+ private static final Logger LOG =
LoggerFactory.getLogger(PostTransformOperator.class);
private final String timezone;
private final List<TransformRule> transformRules;
@@ -79,6 +85,9 @@ public class PostTransformOperator extends
AbstractStreamOperatorAdapter<Event>
// Tuple3 items are: function name, class path, and extra options.
private final List<Tuple3<String, String, Map<String, String>>>
udfFunctions;
+ // Serializable AI model clients keyed by model name, e.g. myModel.
+ private final Map<String, AiModelClient> modelClients;
+
private transient List<PostTransformer> transformers;
private transient List<UserDefinedFunctionDescriptor> udfDescriptors;
private transient List<Object> udfFunctionInstances;
@@ -98,13 +107,15 @@ public class PostTransformOperator extends
AbstractStreamOperatorAdapter<Event>
PostTransformOperator(
List<TransformRule> transformRules,
String timezone,
- List<Tuple3<String, String, Map<String, String>>> udfFunctions) {
+ List<Tuple3<String, String, Map<String, String>>> udfFunctions,
+ Map<String, AiModelClient> modelClients) {
this.timezone = timezone;
this.transformRules = transformRules;
this.hasAsteriskMap = new HashMap<>();
this.projectedColumnsMap = new HashMap<>();
this.postTransformInfoMap = new ConcurrentHashMap<>();
this.udfFunctions = udfFunctions;
+ this.modelClients = modelClients;
}
@Override
@@ -115,6 +126,9 @@ public class PostTransformOperator extends
AbstractStreamOperatorAdapter<Event>
this.projectionProcessors = HashBasedTable.create();
this.filterProcessors = HashBasedTable.create();
+ // Initialize AI model clients
+ initializeAiModelClients();
+
// Be sure to initialize UDF related fields before creating
transformers
initializeUdf();
@@ -136,6 +150,7 @@ public class PostTransformOperator extends
AbstractStreamOperatorAdapter<Event>
super.close();
TransformExpressionCompiler.cleanUp();
destroyUdf();
+ destroyAiModelClients();
}
@Override
@@ -447,7 +462,8 @@ public class PostTransformOperator extends
AbstractStreamOperatorAdapter<Event>
timezone,
udfDescriptors,
udfFunctionInstances,
- postTransformer.getSupportedMetadataColumns()));
+ postTransformer.getSupportedMetadataColumns(),
+ modelClients));
}
return projectionProcessors.get(tableId, postTransformer);
}
@@ -472,7 +488,8 @@ public class PostTransformOperator extends
AbstractStreamOperatorAdapter<Event>
timezone,
udfDescriptors,
udfFunctionInstances,
-
postTransformer.getSupportedMetadataColumns()));
+ postTransformer.getSupportedMetadataColumns(),
+ modelClients));
}
}
return filterProcessors.get(tableId, postTransformer);
@@ -558,4 +575,28 @@ public class PostTransformOperator extends
AbstractStreamOperatorAdapter<Event>
udfDescriptors.clear();
udfFunctionInstances.clear();
}
+
+ private void initializeAiModelClients() {
+ for (Map.Entry<String, AiModelClient> entry : modelClients.entrySet())
{
+ try {
+ entry.getValue().open();
+ LOG.info("Successfully opened AI model client '{}'.",
entry.getKey());
+ } catch (Exception e) {
+ LOG.error("Failed to open AI model client '{}'.",
entry.getKey(), e);
+ throw new FlinkRuntimeException(
+ "Failed to initialize AI model: " + entry.getKey(), e);
+ }
+ }
+ }
+
+ private void destroyAiModelClients() {
+ for (Map.Entry<String, AiModelClient> entry : modelClients.entrySet())
{
+ try {
+ entry.getValue().close();
+ LOG.info("Successfully closed AI model client '{}'.",
entry.getKey());
+ } catch (Exception e) {
+ LOG.warn("Failed to close AI model client '{}'.",
entry.getKey(), e);
+ }
+ }
+ }
}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorBuilder.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorBuilder.java
index 380d9343a..be2e8823b 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorBuilder.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorBuilder.java
@@ -18,6 +18,7 @@
package org.apache.flink.cdc.runtime.operators.transform;
import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.cdc.common.model.AiModelClient;
import org.apache.flink.cdc.common.pipeline.PipelineOptions;
import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
@@ -25,6 +26,7 @@ import javax.annotation.Nullable;
import java.time.ZoneId;
import java.util.ArrayList;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -34,6 +36,7 @@ public class PostTransformOperatorBuilder {
private String timezone;
private final List<Tuple3<String, String, Map<String, String>>>
udfFunctions =
new ArrayList<>();
+ private final Map<String, AiModelClient> modelClients = new
LinkedHashMap<>();
public PostTransformOperatorBuilder addTransform(
String tableInclusions,
@@ -111,7 +114,12 @@ public class PostTransformOperatorBuilder {
return this;
}
+ public PostTransformOperatorBuilder addModelClients(Map<String,
AiModelClient> clients) {
+ this.modelClients.putAll(clients);
+ return this;
+ }
+
public PostTransformOperator build() {
- return new PostTransformOperator(transformRules, timezone,
udfFunctions);
+ return new PostTransformOperator(transformRules, timezone,
udfFunctions, modelClients);
}
}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumnProcessor.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumnProcessor.java
index db5c37ca2..2dc86b37b 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumnProcessor.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumnProcessor.java
@@ -18,6 +18,7 @@
package org.apache.flink.cdc.runtime.operators.transform;
import org.apache.flink.cdc.common.converter.JavaClassConverter;
+import org.apache.flink.cdc.common.model.AiModelClient;
import org.apache.flink.cdc.common.schema.Column;
import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.runtime.parser.JaninoCompiler;
@@ -26,6 +27,7 @@ import org.codehaus.janino.ExpressionEvaluator;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
@@ -45,6 +47,7 @@ public class ProjectionColumnProcessor {
private final TransformExpressionKey transformExpressionKey;
private final Map<String, SupportedMetadataColumn>
supportedMetadataColumns;
private final List<Object> udfFunctionInstances;
+ private final Map<String, AiModelClient> modelClients;
private final ExpressionEvaluator expressionEvaluator;
public ProjectionColumnProcessor(
@@ -53,15 +56,17 @@ public class ProjectionColumnProcessor {
String timezone,
List<UserDefinedFunctionDescriptor> udfDescriptors,
final List<Object> udfFunctionInstances,
- Map<String, SupportedMetadataColumn> supportedMetadataColumns) {
+ Map<String, SupportedMetadataColumn> supportedMetadataColumns,
+ Map<String, AiModelClient> modelClients) {
this.tableInfo = tableInfo;
this.projectionColumn = projectionColumn;
this.timezone = timezone;
this.supportedMetadataColumns = supportedMetadataColumns;
+ this.modelClients = modelClients;
this.transformExpressionKey = generateTransformExpressionKey();
this.expressionEvaluator =
TransformExpressionCompiler.compileExpression(
- transformExpressionKey, udfDescriptors);
+ transformExpressionKey, udfDescriptors, modelClients);
this.udfFunctionInstances = udfFunctionInstances;
}
@@ -72,13 +77,32 @@ public class ProjectionColumnProcessor {
List<UserDefinedFunctionDescriptor> udfDescriptors,
List<Object> udfFunctionInstances,
Map<String, SupportedMetadataColumn> supportedMetadataColumns) {
+ return of(
+ tableInfo,
+ projectionColumn,
+ timezone,
+ udfDescriptors,
+ udfFunctionInstances,
+ supportedMetadataColumns,
+ Collections.emptyMap());
+ }
+
+ public static ProjectionColumnProcessor of(
+ PostTransformChangeInfo tableInfo,
+ ProjectionColumn projectionColumn,
+ String timezone,
+ List<UserDefinedFunctionDescriptor> udfDescriptors,
+ List<Object> udfFunctionInstances,
+ Map<String, SupportedMetadataColumn> supportedMetadataColumns,
+ Map<String, AiModelClient> modelClients) {
return new ProjectionColumnProcessor(
tableInfo,
projectionColumn,
timezone,
udfDescriptors,
udfFunctionInstances,
- supportedMetadataColumns);
+ supportedMetadataColumns,
+ modelClients);
}
public Object evaluate(Object[] rowData, TransformContext context) {
@@ -123,6 +147,9 @@ public class ProjectionColumnProcessor {
// 3 - Add UDF function instances
params.addAll(udfFunctionInstances);
+
+ // 4 - Add AI model client instances
+ params.addAll(modelClients.values());
return params.toArray();
}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformExpressionCompiler.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformExpressionCompiler.java
index 1085a7df8..776b4af2d 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformExpressionCompiler.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformExpressionCompiler.java
@@ -18,6 +18,7 @@
package org.apache.flink.cdc.runtime.operators.transform;
import org.apache.flink.api.common.InvalidProgramException;
+import org.apache.flink.cdc.common.model.AiModelClient;
import
org.apache.flink.cdc.runtime.operators.transform.exceptions.TransformException;
import org.apache.flink.util.FlinkRuntimeException;
@@ -30,7 +31,9 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
/**
* The processor of the transform expression. It processes the expression of
projections and
@@ -54,6 +57,20 @@ public class TransformExpressionCompiler {
/** Compiles an expression code to a janino {@link ExpressionEvaluator}. */
public static ExpressionEvaluator compileExpression(
TransformExpressionKey key, List<UserDefinedFunctionDescriptor>
udfDescriptors) {
+ return compileExpression(key, udfDescriptors, Collections.emptyMap());
+ }
+
+ /**
+ * Compiles an expression code to a janino {@link ExpressionEvaluator},
with additional {@link
+ * AiModelClient} instances appended after UDF instances.
+ *
+ * <p>{@code modelClients} maps model names (e.g. {@code myModel}) to the
corresponding client
+ * instances.
+ */
+ public static ExpressionEvaluator compileExpression(
+ TransformExpressionKey key,
+ List<UserDefinedFunctionDescriptor> udfDescriptors,
+ Map<String, AiModelClient> modelClients) {
try {
return COMPILED_EXPRESSION_CACHE.get(
key,
@@ -68,6 +85,11 @@ public class TransformExpressionCompiler {
argumentClasses.add(Class.forName(udfFunction.getClasspath()));
}
+ for (String paramName : modelClients.keySet()) {
+ argumentNames.add(paramName);
+ argumentClasses.add(AiModelClient.class);
+ }
+
// Input args
expressionEvaluator.setParameters(
argumentNames.toArray(new String[0]),
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilterProcessor.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilterProcessor.java
index 56ce510c1..6781ea191 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilterProcessor.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilterProcessor.java
@@ -19,6 +19,7 @@ package org.apache.flink.cdc.runtime.operators.transform;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cdc.common.converter.JavaClassConverter;
+import org.apache.flink.cdc.common.model.AiModelClient;
import org.apache.flink.cdc.common.schema.Column;
import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.runtime.parser.JaninoCompiler;
@@ -47,6 +48,7 @@ public class TransformFilterProcessor {
private final String timezone;
private final List<Object> udfFunctionInstances;
private final Map<String, SupportedMetadataColumn>
supportedMetadataColumns;
+ private final Map<String, AiModelClient> modelClients;
private final TransformExpressionKey transformExpressionKey;
private final ExpressionEvaluator expressionEvaluator;
@@ -58,13 +60,15 @@ public class TransformFilterProcessor {
String timezone,
List<UserDefinedFunctionDescriptor> udfDescriptors,
List<Object> udfFunctionInstances,
- Map<String, SupportedMetadataColumn> supportedMetadataColumns) {
+ Map<String, SupportedMetadataColumn> supportedMetadataColumns,
+ Map<String, AiModelClient> modelClients) {
this.isNoOp = isNoOp;
this.tableInfo = tableInfo;
this.transformFilter = transformFilter;
this.timezone = timezone;
this.udfFunctionInstances = udfFunctionInstances;
this.supportedMetadataColumns = supportedMetadataColumns;
+ this.modelClients = modelClients;
if (isNoOp) {
this.transformExpressionKey = null;
@@ -79,12 +83,12 @@ public class TransformFilterProcessor {
.toArray(new SupportedMetadataColumn[0]));
this.expressionEvaluator =
TransformExpressionCompiler.compileExpression(
- transformExpressionKey, udfDescriptors);
+ transformExpressionKey, udfDescriptors,
modelClients);
}
}
public static TransformFilterProcessor ofNoOp() {
- return new TransformFilterProcessor(true, null, null, null, null,
null, null);
+ return new TransformFilterProcessor(true, null, null, null, null,
null, null, null);
}
public static TransformFilterProcessor of(
@@ -93,7 +97,8 @@ public class TransformFilterProcessor {
String timezone,
List<UserDefinedFunctionDescriptor> udfDescriptors,
List<Object> udfFunctionInstances,
- SupportedMetadataColumn[] supportedMetadataColumns) {
+ SupportedMetadataColumn[] supportedMetadataColumns,
+ Map<String, AiModelClient> modelClients) {
Map<String, SupportedMetadataColumn> supportedMetadataColumnsMap = new
HashMap<>();
for (SupportedMetadataColumn supportedMetadataColumn :
supportedMetadataColumns) {
supportedMetadataColumnsMap.put(
@@ -106,7 +111,8 @@ public class TransformFilterProcessor {
timezone,
udfDescriptors,
udfFunctionInstances,
- supportedMetadataColumnsMap);
+ supportedMetadataColumnsMap,
+ modelClients);
}
public boolean test(Object[] preRow, Object[] postRow, TransformContext
context) {
@@ -209,6 +215,9 @@ public class TransformFilterProcessor {
// 3 - Add UDF function instances
params.addAll(udfFunctionInstances);
+
+ // 4 - Add AI model client instances
+ params.addAll(modelClients.values());
return params.toArray();
}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformProjectionProcessor.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformProjectionProcessor.java
index 231e72875..09fe5fb19 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformProjectionProcessor.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformProjectionProcessor.java
@@ -17,6 +17,7 @@
package org.apache.flink.cdc.runtime.operators.transform;
+import org.apache.flink.cdc.common.model.AiModelClient;
import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.common.utils.Preconditions;
import org.apache.flink.cdc.runtime.parser.TransformParser;
@@ -52,6 +53,7 @@ public class TransformProjectionProcessor {
private final List<ProjectionColumnProcessor> columnProcessors;
private final SupportedMetadataColumn[] supportedMetadataColumns;
private final Map<String, SupportedMetadataColumn>
supportedMetadataColumnsMap;
+ private final Map<String, AiModelClient> modelClients;
public TransformProjectionProcessor(
PostTransformChangeInfo changeInfo,
@@ -59,13 +61,15 @@ public class TransformProjectionProcessor {
String timezone,
List<UserDefinedFunctionDescriptor> udfDescriptors,
List<Object> udfFunctionInstances,
- SupportedMetadataColumn[] supportedMetadataColumns) {
+ SupportedMetadataColumn[] supportedMetadataColumns,
+ Map<String, AiModelClient> modelClients) {
this.changeInfo = changeInfo;
this.projectionExpression = projectionExpression;
this.timezone = timezone;
this.udfDescriptors = udfDescriptors;
this.udfFunctionInstances = udfFunctionInstances;
this.supportedMetadataColumns = supportedMetadataColumns;
+ this.modelClients = modelClients;
// Construct a mapping table ad-hoc to accelerate looking-up
Map<String, SupportedMetadataColumn> supportedMetadataColumnsMap = new
HashMap<>();
@@ -105,7 +109,8 @@ public class TransformProjectionProcessor {
timezone,
udfDescriptors,
udfFunctionInstances,
- supportedMetadataColumnsMap))
+ supportedMetadataColumnsMap,
+ modelClients))
.collect(Collectors.toList());
LOG.info("Successfully created projection column processors cache.");
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
index 7dda0a26e..abe2fb8c2 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
@@ -28,6 +28,8 @@ import org.apache.flink.cdc.common.types.DataTypeRoot;
import org.apache.flink.cdc.common.types.DecimalType;
import org.apache.flink.cdc.common.utils.Preconditions;
import org.apache.flink.cdc.common.utils.StringUtils;
+import org.apache.flink.cdc.runtime.ai.AiEmbeddingFunctionDef;
+import org.apache.flink.cdc.runtime.ai.AiTextFunctionDef;
import
org.apache.flink.cdc.runtime.operators.transform.UserDefinedFunctionDescriptor;
import org.apache.flink.cdc.runtime.parser.metadata.MetadataColumns;
@@ -117,7 +119,7 @@ public class JaninoCompiler {
public static final String DEFAULT_TIME_ZONE = "__time_zone__";
private static final String[] BUILTIN_FUNCTION_MODULES = {
- "Arithmetic", "Casting", "Comparison", "Logical", "String", "Struct",
"Temporal"
+ "Ai", "Arithmetic", "Casting", "Comparison", "Logical", "String",
"Struct", "Temporal"
};
@VisibleForTesting
@@ -858,6 +860,13 @@ public class JaninoCompiler {
} else if (operationName.equals("NULLIF")) {
return generateNullIfOperation(context, sqlBasicCall, atoms);
} else {
+ if (isAiFunction(operationName) && atoms.length >= 1) {
+ if (!(sqlBasicCall.operand(0) instanceof
SqlCharStringLiteral)) {
+ throw new ParseException(
+ "The model argument of an AI function must be a
string constant.");
+ }
+ rewriteAiFunctionModelArg(atoms);
+ }
return new Java.MethodInvocation(
Location.NOWHERE,
null,
@@ -950,6 +959,28 @@ public class JaninoCompiler {
}
}
+ private static boolean isAiFunction(String upperCaseName) {
+ for (AiTextFunctionDef def : AiTextFunctionDef.values()) {
+ if (def.getFunctionName().equals(upperCaseName)) {
+ return true;
+ }
+ }
+ for (AiEmbeddingFunctionDef def : AiEmbeddingFunctionDef.values()) {
+ if (def.getFunctionName().equals(upperCaseName)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static void rewriteAiFunctionModelArg(Java.Rvalue[] atoms) {
+ String modelName = atoms[0].toString();
+ if (modelName.startsWith("\"") && modelName.endsWith("\"")) {
+ modelName = modelName.substring(1, modelName.length() - 1);
+ }
+ atoms[0] = new Java.AmbiguousName(Location.NOWHERE, new String[]
{modelName});
+ }
+
private static Java.Rvalue generateTimezoneFreeTemporalFunctionOperation(
Context context, String operationName) {
return new Java.MethodInvocation(
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
index 6020245ba..8276d323c 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
@@ -24,6 +24,7 @@ import org.apache.flink.cdc.common.types.DataType;
import org.apache.flink.cdc.common.utils.Preconditions;
import org.apache.flink.cdc.runtime.operators.transform.ProjectionColumn;
import
org.apache.flink.cdc.runtime.operators.transform.UserDefinedFunctionDescriptor;
+import org.apache.flink.cdc.runtime.parser.metadata.AiFunctionSqlOperatorTable;
import org.apache.flink.cdc.runtime.parser.metadata.TransformSchemaFactory;
import org.apache.flink.cdc.runtime.parser.metadata.TransformSqlOperatorTable;
import org.apache.flink.cdc.runtime.typeutils.CalciteDataTypeConverter;
@@ -47,6 +48,7 @@ import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.schema.impl.ScalarFunctionImpl;
import org.apache.calcite.sql.SqlBasicCall;
import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlCharStringLiteral;
import org.apache.calcite.sql.SqlFunction;
import org.apache.calcite.sql.SqlFunctionCategory;
import org.apache.calcite.sql.SqlIdentifier;
@@ -164,9 +166,13 @@ public class TransformParser {
new CalciteConnectionConfigImpl(new Properties()));
TransformSqlOperatorTable transformSqlOperatorTable =
TransformSqlOperatorTable.instance();
SqlOperatorTable udfOperatorTable = SqlOperatorTables.of(udfFunctions);
+ SqlOperatorTable aiFunctionOperatorTable =
AiFunctionSqlOperatorTable.create();
SqlValidator validator =
SqlValidatorUtil.newValidator(
- SqlOperatorTables.chain(transformSqlOperatorTable,
udfOperatorTable),
+ SqlOperatorTables.chain(
+ transformSqlOperatorTable,
+ udfOperatorTable,
+ aiFunctionOperatorTable),
calciteCatalogReader,
factory,
SqlValidator.Config.DEFAULT
@@ -724,6 +730,54 @@ public class TransformParser {
return parseSelect(statement.toString());
}
+ /** Validates model arguments and references in the supported AI
functions. */
+ public static void validateAiModelReferences(
+ @Nullable String projection, @Nullable String filter, Set<String>
declaredModelNames) {
+ if (!isNullOrWhitespaceOnly(projection)) {
+ validateAiModelReferences(parseProjectionExpression(projection),
declaredModelNames);
+ }
+ if (!isNullOrWhitespaceOnly(filter)) {
+ validateAiModelReferences(parseFilterExpression(filter),
declaredModelNames);
+ }
+ }
+
+ private static void validateAiModelReferences(SqlNode node, Set<String>
declaredModelNames) {
+ if (node instanceof SqlCall) {
+ SqlCall call = (SqlCall) node;
+ if (isAiFunction(call.getOperator().getName())) {
+ if (call.operandCount() == 0) {
+ return;
+ }
+ SqlNode modelArgument = call.operand(0);
+ Preconditions.checkArgument(
+ modelArgument instanceof SqlCharStringLiteral,
+ "The model argument of %s must be a string constant,
but was %s.",
+ call.getOperator().getName(),
+ modelArgument);
+ String modelName = ((SqlCharStringLiteral)
modelArgument).getNlsString().getValue();
+ Preconditions.checkArgument(
+ declaredModelNames.contains(modelName),
+ "Model '%s' referenced by %s has not been declared.",
+ modelName,
+ call.getOperator().getName());
+ }
+ for (SqlNode operand : call.getOperandList()) {
+ if (operand != null) {
+ validateAiModelReferences(operand, declaredModelNames);
+ }
+ }
+ } else if (node instanceof SqlNodeList) {
+ for (SqlNode child : (SqlNodeList) node) {
+ validateAiModelReferences(child, declaredModelNames);
+ }
+ }
+ }
+
+ private static boolean isAiFunction(String functionName) {
+ return "AI_COMPLETE".equalsIgnoreCase(functionName)
+ || "AI_EMBED".equalsIgnoreCase(functionName);
+ }
+
public static boolean hasAsterisk(@Nullable String projection) {
if (isNullOrWhitespaceOnly(projection)) {
// Providing an empty projection expression is equivalent to
writing `*` explicitly.
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/AiFunctionSqlOperatorTable.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/AiFunctionSqlOperatorTable.java
new file mode 100644
index 000000000..c1d1fa995
--- /dev/null
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/AiFunctionSqlOperatorTable.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.runtime.parser.metadata;
+
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.RowType;
+import org.apache.flink.cdc.runtime.ai.AiEmbeddingFunctionDef;
+import org.apache.flink.cdc.runtime.ai.AiTextFunctionDef;
+import org.apache.flink.cdc.runtime.typeutils.CalciteDataTypeConverter;
+
+import org.apache.calcite.sql.SqlFunction;
+import org.apache.calcite.sql.SqlFunctionCategory;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperatorTable;
+import org.apache.calcite.sql.type.OperandTypes;
+import org.apache.calcite.sql.type.ReturnTypes;
+import org.apache.calcite.sql.type.SqlTypeFamily;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.sql.util.SqlOperatorTables;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** Creates SqlOperatorTable from {@link AiTextFunctionDef} definitions. */
+public class AiFunctionSqlOperatorTable {
+
+ private AiFunctionSqlOperatorTable() {}
+
+ /** Creates an SqlOperatorTable containing all AI functions defined in
AiFunctionDef. */
+ public static SqlOperatorTable create() {
+ List<SqlFunction> functions = new ArrayList<>();
+ for (AiTextFunctionDef def : AiTextFunctionDef.values()) {
+ functions.add(createTextSqlFunction(def));
+ }
+ for (AiEmbeddingFunctionDef def : AiEmbeddingFunctionDef.values()) {
+ functions.add(createEmbeddingSqlFunction(def));
+ }
+ return SqlOperatorTables.of(functions);
+ }
+
+ private static SqlFunction createTextSqlFunction(AiTextFunctionDef def) {
+ return new SqlFunction(
+ def.getFunctionName(),
+ SqlKind.OTHER_FUNCTION,
+ ReturnTypes.explicit(SqlTypeName.VARIANT),
+ null,
+
OperandTypes.family(toSqlTypeFamiliesWithAdditionalParams(def.getInputType())),
+ SqlFunctionCategory.USER_DEFINED_FUNCTION);
+ }
+
+ private static SqlFunction
createEmbeddingSqlFunction(AiEmbeddingFunctionDef def) {
+ return new SqlFunction(
+ def.getFunctionName(),
+ SqlKind.OTHER_FUNCTION,
+ opBinding ->
+ CalciteDataTypeConverter.convertCalciteType(
+ opBinding.getTypeFactory(),
def.getOutputType()),
+ null,
+ OperandTypes.family(SqlTypeFamily.STRING,
toSqlTypeFamily(def.getInputType())),
+ SqlFunctionCategory.USER_DEFINED_FUNCTION);
+ }
+
+ /**
+ * Converts inputType to SqlTypeFamily array, prepending additional
parameters: modelName
+ * (STRING) and input (STRING).
+ */
+ private static SqlTypeFamily[]
toSqlTypeFamiliesWithAdditionalParams(RowType inputType) {
+ List<SqlTypeFamily> families = new ArrayList<>();
+ families.add(SqlTypeFamily.STRING); // modelName
+ families.add(SqlTypeFamily.STRING); // input
+ for (DataType fieldType : inputType.getFieldTypes()) {
+ families.add(toSqlTypeFamily(fieldType));
+ }
+ return families.toArray(new SqlTypeFamily[0]);
+ }
+
+ private static SqlTypeFamily toSqlTypeFamily(DataType dataType) {
+ switch (dataType.getTypeRoot()) {
+ case VARCHAR:
+ case CHAR:
+ return SqlTypeFamily.STRING;
+ case INTEGER:
+ return SqlTypeFamily.INTEGER;
+ case BIGINT:
+ return SqlTypeFamily.NUMERIC;
+ case FLOAT:
+ case DOUBLE:
+ return SqlTypeFamily.APPROXIMATE_NUMERIC;
+ case BOOLEAN:
+ return SqlTypeFamily.BOOLEAN;
+ default:
+ return SqlTypeFamily.ANY;
+ }
+ }
+}
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/AiFunctionsTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/AiFunctionsTest.java
new file mode 100644
index 000000000..3cf9873fc
--- /dev/null
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/AiFunctionsTest.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.runtime.functions.impl;
+
+import org.apache.flink.cdc.common.model.AiModelClient;
+import org.apache.flink.cdc.common.model.abilities.SupportsEmbedding;
+import org.apache.flink.cdc.common.model.abilities.SupportsTextGeneration;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link AiFunctions}. */
+class AiFunctionsTest {
+
+ private static class TestModelClient
+ implements AiModelClient, SupportsTextGeneration,
SupportsEmbedding {
+
+ private static final long serialVersionUID = 1L;
+
+ private String lastPrompt;
+
+ @Override
+ public String generate(String systemPrompt, String userInput) {
+ lastPrompt = systemPrompt;
+ return "{\"result\":\"ABC\"}";
+ }
+
+ @Override
+ public float[] embed(String text) {
+ return new float[] {0.1f, 0.2f, 0.3f};
+ }
+ }
+
+ private static class UnsupportedModelClient implements AiModelClient {
+ private static final long serialVersionUID = 1L;
+ }
+
+ @Test
+ void testAiFunctions() {
+ TestModelClient model = new TestModelClient();
+
+ assertThat(AiFunctions.aiComplete(model, "input", "Return three
letters"))
+ .hasToString("{\"result\":\"ABC\"}");
+ assertThat(model.lastPrompt).contains("Return three
letters").contains("\"result\"");
+ assertThat(AiFunctions.aiEmbed(model, "input")).containsExactly(0.1f,
0.2f, 0.3f);
+ }
+
+ @Test
+ void testUnsupportedCapabilities() {
+ UnsupportedModelClient model = new UnsupportedModelClient();
+
+ assertThatThrownBy(() -> AiFunctions.aiComplete(model, "input",
"prompt"))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("does not support text generation");
+ assertThatThrownBy(() -> AiFunctions.aiEmbed(model, "input"))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("does not support embedding");
+ }
+
+ @Test
+ void testInvalidJsonResponse() {
+ TestModelClient model =
+ new TestModelClient() {
+ @Override
+ public String generate(String systemPrompt, String
userInput) {
+ return "not-json";
+ }
+ };
+
+ assertThatThrownBy(() -> AiFunctions.aiComplete(model, "input",
"prompt"))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to parse AI response as JSON");
+ }
+}
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/AiFunctionParserTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/AiFunctionParserTest.java
new file mode 100644
index 000000000..e7af51542
--- /dev/null
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/AiFunctionParserTest.java
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.runtime.parser;
+
+import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.runtime.operators.transform.ProjectionColumn;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Parser and Janino tests for the generic AI functions. */
+class AiFunctionParserTest {
+
+ private static final List<Column> COLUMNS =
+ List.of(
+ Column.physicalColumn("id", DataTypes.INT()),
+ Column.physicalColumn("content", DataTypes.STRING()));
+
+ @Test
+ void testTranslateAiFunctions() {
+ List<ProjectionColumn> columns =
+ translate(
+ "AI_COMPLETE('completer', content, 'You are helpful')
AS completed, "
+ + "AI_EMBED('embedder', content) AS
embedding");
+
+ assertThat(columns)
+ .extracting(ProjectionColumn::getScriptExpression)
+ .containsExactly(
+ "aiComplete(completer, $0, \"You are helpful\")",
"aiEmbed(embedder, $0)");
+ assertThat(columns)
+ .extracting(ProjectionColumn::getDataType)
+ .containsExactly(DataTypes.VARIANT(),
DataTypes.ARRAY(DataTypes.FLOAT()));
+ }
+
+ @Test
+ void testModelArgumentMustBeStringConstant() {
+ assertThatThrownBy(
+ () ->
+ TransformParser.validateAiModelReferences(
+ "AI_COMPLETE(content, content,
'prompt') AS completed",
+ null,
+ Set.of("content")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must be a string constant");
+ }
+
+ @Test
+ void testReferencedModelMustBeDeclared() {
+ assertThatThrownBy(
+ () ->
+ TransformParser.validateAiModelReferences(
+ "AI_EMBED('missing', content) AS
embedding",
+ null,
+ Set.of("declared")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Model 'missing'")
+ .hasMessageContaining("has not been declared");
+
+ assertThatCode(
+ () ->
+ TransformParser.validateAiModelReferences(
+ "AI_EMBED('declared', content) AS
embedding",
+ null,
+ Set.of("declared")))
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ void testFunctionArityValidation() {
+ assertThatThrownBy(() -> translate("AI_EMBED('model') AS embedding"))
+ .hasMessageContaining("Invalid number of arguments to function
'AI_EMBED'");
+ assertThatThrownBy(() -> translate("AI_COMPLETE('model', content) AS
completed"))
+ .hasMessageContaining("Invalid number of arguments to function
'AI_COMPLETE'");
+ assertThatThrownBy(() -> translate("AI_COMPLETE() AS completed"))
+ .hasMessageContaining("Invalid number of arguments to function
'AI_COMPLETE'");
+ assertThatCode(
+ () ->
+ TransformParser.validateAiModelReferences(
+ "AI_COMPLETE() AS completed", null,
Collections.emptySet()))
+ .doesNotThrowAnyException();
+ }
+
+ private List<ProjectionColumn> translate(String expression) {
+ return TransformParser.generateProjectionColumns(
+ expression, COLUMNS, Collections.emptyList(), new
SupportedMetadataColumn[0]);
+ }
+}
