This is an automated email from the ASF dual-hosted git repository.
lvyanquan 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 48944c6fcd [FLINK-40409][runtime] CDC YAML supports inline Python UDF
definition (#4501)
48944c6fcd is described below
commit 48944c6fcd7f952464abee9e2a1c8829e5ac05d6
Author: yuxiqian <[email protected]>
AuthorDate: Wed Aug 19 11:19:52 2026 +0800
[FLINK-40409][runtime] CDC YAML supports inline Python UDF definition
(#4501)
---
.github/workflows/flink_cdc_base.yml | 9 +
.github/workflows/modules.py | 1 +
docs/content.zh/docs/core-concept/transform.md | 110 ++++++++++
docs/content/docs/core-concept/transform.md | 112 ++++++++++
.../cli/parser/YamlPipelineDefinitionParser.java | 84 +++++++-
.../parser/YamlPipelineDefinitionParserTest.java | 198 +++++++++++++++++
.../pipeline-definition-with-python-udf.yaml | 37 ++++
.../flink/cdc/common/udf/UserDefinedFunction.java | 15 +-
flink-cdc-dist/pom.xml | 5 +
.../cdc/pipeline/tests/PythonUdfE2eITCase.java | 221 +++++++++++++++++++
.../src/test/resources/ddl/python_udf_test.sql | 37 ++++
.../src/test/resources/rules/unexpected.yaml | 2 +-
flink-cdc-python/pom.xml | 76 +++++++
.../org/apache/flink/cdc/python/PythonUdf.java | 236 +++++++++++++++++++++
.../flink/cdc/python/utils/PythonUdfSignature.java | 85 ++++++++
.../org/apache/flink/cdc/python/signature.py | 54 +++++
.../org/apache/flink/cdc/python/PythonUdfTest.java | 219 +++++++++++++++++++
.../flink/cdc/python/utils/PemjaTestSupport.java | 53 +++++
.../cdc/python/utils/PythonUdfSignatureTest.java | 90 ++++++++
.../transform/TransformExpressionCompiler.java | 2 +-
.../transform/UserDefinedFunctionDescriptor.java | 11 +-
.../flink/cdc/runtime/parser/JaninoCompiler.java | 6 +-
.../UserDefinedFunctionDescriptorTest.java | 24 +++
.../cdc/runtime/parser/TransformParserTest.java | 97 +++------
pom.xml | 1 +
25 files changed, 1699 insertions(+), 86 deletions(-)
diff --git a/.github/workflows/flink_cdc_base.yml
b/.github/workflows/flink_cdc_base.yml
index 982ea81e3e..c71195e171 100644
--- a/.github/workflows/flink_cdc_base.yml
+++ b/.github/workflows/flink_cdc_base.yml
@@ -89,6 +89,15 @@ jobs:
with:
maven-version: 3.8.6
+ - name: Install Python and Pemja
+ # Required by flink-cdc-python unit tests.
+ if: ${{ matrix.module == 'core' }}
+ run: |
+ set -euo pipefail
+ sudo apt-get update
+ sudo apt-get install -y python3 python3-pip python3-dev
+ python3 -m pip install --disable-pip-version-check pemja==0.5.7
+
- name: Compile and test
timeout-minutes: 90
run: |
diff --git a/.github/workflows/modules.py b/.github/workflows/modules.py
index d92a4fc653..123b20b620 100755
--- a/.github/workflows/modules.py
+++ b/.github/workflows/modules.py
@@ -14,6 +14,7 @@ MODULES_CORE = [
"flink-cdc-composer",
"flink-cdc-runtime",
"flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible",
+ "flink-cdc-python",
"flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base",
"flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values"
]
diff --git a/docs/content.zh/docs/core-concept/transform.md
b/docs/content.zh/docs/core-concept/transform.md
index 0409d7a6a1..aad269c88a 100644
--- a/docs/content.zh/docs/core-concept/transform.md
+++ b/docs/content.zh/docs/core-concept/transform.md
@@ -528,3 +528,113 @@ transform:
```
有关 AI 模型函数和配置，请参见 [AI 模型]({{< ref "docs/core-concept/ai-model" >}})。
+
+### Python UDF
+
+Flink CDC 支持直接在 Pipeline YAML 中定义 Python UDF。发行包内置了 `flink-cdc-python`
+模块，通过 [Pemja](https://pypi.org/project/pemja/) 在每个 TaskManager 中嵌入 Python。
+
+```yaml
+transform:
+ - source-table: db.users
+ projection: ID, py_normalize(EMAIL) AS EMAIL_NORM, py_double(AGE) AS
DOUBLED
+
+pipeline:
+ user-defined-function:
+ - name: py_normalize
+ python-code: |
+ def eval(s: str) -> str:
+ return None if s is None else s.strip().lower()
+ python-executable: /usr/bin/python3
+ - name: py_double
+ python-code: |
+ def eval(x: int) -> int:
+ return None if x is None else x * 2
+ python-files:
+ - /opt/flink/python-deps
+ - /opt/flink/python-deps.zip
+```
+
+每个条目必须包含一个名为 `eval` 的顶层函数，并声明返回值类型。目前支持 `bool`、
+`bytes`、`float`、`int` 和 `str`，依次映射为 Flink CDC 的 `BOOLEAN`、`BYTES`、
+`DOUBLE`、`BIGINT` 和 `STRING`。
+
+运行时还需满足以下条件：
+
+* 每个 TaskManager 都要安装 Python 和 `pemja==0.5.7`。`python-executable` 默认使用
+ `PATH` 中找到的第一个 `python3`。
+* 可选的 `python-files` 是一个 YAML 列表，元素只能是已存在的目录或 `.zip` 压缩包；
+ 这些路径必须在每个 TaskManager 上都可访问。压缩包解压后才会加入 Python import path。
+* `python-code` 不能与 `classpath` 或 `options` 同时使用。每个 Python 函数应单独声明为
+ 一个 UDF 条目。
+
+## Embedding AI 模型
+
+内置 AI 模型可以在 transform 规则中使用。
+为了使用内置 AI 模型，你需要下载内置模型的 jar ，然后在 `flink-cdc.sh` 命令中添加 `--jar
{$BUILT_IN_MODEL_PATH}`。
+
+如何定义一个 Embedding AI 模型：
+
+```yaml
+pipeline:
+ model:
+ - model-name: CHAT
+ class-name: OpenAIChatModel
+ openai.model: gpt-4o-mini
+ openai.host: https://xxxx
+ openai.apikey: abcd1234
+ openai.chat.prompt: please summary this
+ - model-name: GET_EMBEDDING
+ class-name: OpenAIEmbeddingModel
+ openai.model: text-embedding-3-small
+ openai.host: https://xxxx
+ openai.apikey: abcd1234
+```
+注意：
+* `model-name` 是一个通用的必填参数，用于所有支持的模型，表示在 `projection` 或 `filter` 中调用的函数名称。
+* `class-name` 是一个通用的必填参数，用于所有支持的模型，可用值可以在[所有支持的模型](#all-support-models)中找到。
+* `openai.model`，`openai.host`，`openai.apiKey` 和 `openai.chat.prompt`
是在各个模型中特别的可选参数。
+
+如何使用一个 Embedding AI 模型：
+
+```yaml
+transform:
+ - source-table: db.\.*
+ projection: "*, inc(inc(inc(id))) as inc_id, GET_EMBEDDING(page) as emb,
CHAT(page) as summary"
+ filter: inc(id) < 100
+pipeline:
+ model:
+ - model-name: CHAT
+ class-name: OpenAIChatModel
+ openai.model: gpt-4o-mini
+ openai.host: http://langchain4j.dev/demo/openai/v1
+ openai.apikey: demo
+ openai.chat.prompt: please summary this
+ - model-name: GET_EMBEDDING
+ class-name: OpenAIEmbeddingModel
+ openai.model: text-embedding-3-small
+ openai.host: http://langchain4j.dev/demo/openai/v1
+ openai.apikey: demo
+```
+这里，GET_EMBEDDING 是通过 `model-name` 在 `pipeline` 中定义的。
+
+### 所有支持的模型
+
+下面列出了所有支持的模型：
+
+#### OpenAIChatModel
+
+| 参数 | 类型 | 是否必填 | 含义
|
+|--------------------|--------|----------|-----------------------------------------------------------------------------------------------------|
+| openai.model | STRING | 必填 | 要调用的模型名称，例如："gpt-4o-mini"，可用选项有
"gpt-4o-mini"、"gpt-4o"、"gpt-4-32k"、"gpt-3.5-turbo"。 |
+| openai.host | STRING | 必填 |
要连接的模型服务器地址，例如：`http://langchain4j.dev/demo/openai/v1`。
|
+| openai.apikey | STRING | 必填 | 模型服务器验证的 API Key，例如："demo"。
|
+| openai.chat.prompt | STRING | 可选 | 与 OpenAI 聊天的提示词，例如："Please summary
this"。 |
+
+#### OpenAIEmbeddingModel
+
+| 参数 | 类型 | 是否必填 | 含义
|
+|---------------|--------|----------|----------------------------------------------------------------------------------------------------------------------|
+| openai.model | STRING | 必填 |
要调用的模型名称，例如："text-embedding-3-small"，可用选项有
"text-embedding-3-small"、"text-embedding-3-large"、"text-embedding-ada-002"。 |
+| openai.host | STRING | 必填 |
要连接的模型服务器地址，例如：`http://langchain4j.dev/demo/openai/v1`。
|
+| openai.apikey | STRING | 必填 | 模型服务器验证的 API Key，例如："demo"。
|
diff --git a/docs/content/docs/core-concept/transform.md
b/docs/content/docs/core-concept/transform.md
index 8e2f9815c7..5ff24b9b54 100644
--- a/docs/content/docs/core-concept/transform.md
+++ b/docs/content/docs/core-concept/transform.md
@@ -533,3 +533,115 @@ transform:
```
For AI model functions and configuration, see [AI Model]({{< ref
"docs/core-concept/ai-model" >}}).
+
+### Python UDFs
+
+Flink CDC supports defining Python UDFs inline in the pipeline YAML. The
distribution includes the
+`flink-cdc-python` module, which embeds Python on each TaskManager through
+[Pemja](https://pypi.org/project/pemja/).
+
+```yaml
+transform:
+ - source-table: db.users
+ projection: ID, py_normalize(EMAIL) AS EMAIL_NORM, py_double(AGE) AS
DOUBLED
+
+pipeline:
+ user-defined-function:
+ - name: py_normalize
+ python-code: |
+ def eval(s: str) -> str:
+ return None if s is None else s.strip().lower()
+ python-executable: /usr/bin/python3
+ - name: py_double
+ python-code: |
+ def eval(x: int) -> int:
+ return None if x is None else x * 2
+ python-files:
+ - /opt/flink/python-deps
+ - /opt/flink/python-deps.zip
+```
+
+Each entry must contain one top-level function named `eval` with a return type
annotation. Supported
+annotations are `bool`, `bytes`, `float`, `int`, and `str`, mapped to Flink
CDC `BOOLEAN`, `BYTES`,
+`DOUBLE`, `BIGINT`, and `STRING`, respectively.
+
+The following runtime requirements apply:
+
+* Every TaskManager must have Python and `pemja==0.5.7` installed.
`python-executable` defaults to
+ the first `python3` on `PATH`.
+* `python-files` is optional and accepts a YAML list of existing directories
or `.zip` archives.
+ These paths must be available on every TaskManager. Zip archives are
extracted before being added
+ to Python's import path.
+* `python-code` cannot be combined with `classpath` or `options`. Use one UDF
entry for each Python
+ function.
+
+## Embedding AI Model
+
+Embedding AI Model can be used in transform rules.
+To use Embedding AI Model, you need to download the jar of build-in model, and
then add `--jar {$BUILT_IN_MODEL_PATH}` to your flink-cdc.sh command.
+
+How to define a Embedding AI Model:
+
+```yaml
+pipeline:
+ model:
+ - model-name: CHAT
+ class-name: OpenAIChatModel
+ openai.model: text-embedding-3-small
+ openai.host: https://xxxx
+ openai.apikey: abcd1234
+ openai.chat.prompt: please summary this
+ - model-name: GET_EMBEDDING
+ class-name: OpenAIEmbeddingModel
+ openai.model: text-embedding-3-small
+ openai.host: https://xxxx
+ openai.apikey: abcd1234
+```
+Note:
+* `model-name` is a common required parameter for all support models, which
represent the function name called in `projection` or `filter`.
+* `class-name` is a common required parameter for all support models,
available values can be found in [All Support models](#all-support-models).
+* `openai.model`, `openai.host`, `openai.apiKey` and `openai.chat.prompt` is
option parameters that defined in specific model.
+
+How to use a Embedding AI Model:
+
+```yaml
+transform:
+ - source-table: db.\.*
+ projection: "*, inc(inc(inc(id))) as inc_id, GET_EMBEDDING(page) as emb,
CHAT(page) as summary"
+ filter: inc(id) < 100
+pipeline:
+ model:
+ - model-name: CHAT
+ class-name: OpenAIChatModel
+ openai.model: gpt-4o-mini
+ openai.host: http://langchain4j.dev/demo/openai/v1
+ openai.apikey: demo
+ openai.chat.prompt: please summary this
+ - model-name: GET_EMBEDDING
+ class-name: OpenAIEmbeddingModel
+ openai.model: text-embedding-3-small
+ openai.host: http://langchain4j.dev/demo/openai/v1
+ openai.apikey: demo
+```
+Here, GET_EMBEDDING is defined though `model-name` in `pipeline`.
+
+### All Support models
+
+The following built-in models are provided:
+
+#### OpenAIChatModel
+
+| parameter | type | optional/required | meaning
|
+|--------------------|--------|-------------------|--------------------------------------------------------------------------------------------------------------------------------------|
+| openai.model | STRING | required | Name of model to be
called, for example: "gpt-4o-mini", Available options are "gpt-4o-mini",
"gpt-4o", "gpt-4-32k", "gpt-3.5-turbo". |
+| openai.host | STRING | required | Host of the Model server
to be connected, for example: `http://langchain4j.dev/demo/openai/v1`.
|
+| openai.apikey | STRING | required | Api Key for verification
of the Model server, for example, "demo".
|
+| openai.chat.prompt | STRING | optional | Prompt for chatting with
OpenAI, for example: "Please summary this ".
|
+
+#### OpenAIEmbeddingModel
+
+| parameter | type | optional/required | meaning
|
+|---------------|--------|-------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| openai.model | STRING | required | Name of model to be called, for
example: "text-embedding-3-small", Available options are
"text-embedding-3-small", "text-embedding-3-large", "text-embedding-ada-002". |
+| openai.host | STRING | required | Host of the Model server to be
connected, for example: `http://langchain4j.dev/demo/openai/v1`.
|
+| openai.apikey | STRING | required | Api Key for verification of the
Model server, for example, "demo".
|
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 b5519afd5a..98573104df 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
@@ -44,6 +44,7 @@ import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.yaml.YA
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
@@ -94,7 +95,11 @@ public class YamlPipelineDefinitionParser implements
PipelineDefinitionParser {
private static final String UDF_KEY = "user-defined-function";
private static final String UDF_FUNCTION_NAME_KEY = "name";
private static final String UDF_CLASSPATH_KEY = "classpath";
+ private static final String UDF_PYTHON_CODE_KEY = "python-code";
+ private static final String UDF_PYTHON_EXECUTABLE_KEY =
"python-executable";
+ private static final String UDF_PYTHON_FILES_KEY = "python-files";
private static final String UDF_OPTIONS_KEY = "options";
+ private static final String PYTHON_UDF_CLASSPATH =
"org.apache.flink.cdc.python.PythonUdf";
// Model related keys
private static final String MODEL_NAME_KEY = "name";
@@ -313,8 +318,13 @@ public class YamlPipelineDefinitionParser implements
PipelineDefinitionParser {
validateJsonNodeKeys(
"UDF",
udfNode,
- Arrays.asList(UDF_FUNCTION_NAME_KEY, UDF_CLASSPATH_KEY),
- Collections.singletonList(UDF_OPTIONS_KEY));
+ Collections.singletonList(UDF_FUNCTION_NAME_KEY),
+ Arrays.asList(
+ UDF_CLASSPATH_KEY,
+ UDF_PYTHON_CODE_KEY,
+ UDF_PYTHON_EXECUTABLE_KEY,
+ UDF_PYTHON_FILES_KEY,
+ UDF_OPTIONS_KEY));
String functionName =
checkNotNull(
@@ -322,12 +332,22 @@ public class YamlPipelineDefinitionParser implements
PipelineDefinitionParser {
"Missing required field \"%s\" in UDF
configuration",
UDF_FUNCTION_NAME_KEY)
.asText();
- String classpath =
- checkNotNull(
- udfNode.get(UDF_CLASSPATH_KEY),
- "Missing required field \"%s\" in UDF
configuration",
- UDF_CLASSPATH_KEY)
- .asText();
+ JsonNode classpathNode = udfNode.get(UDF_CLASSPATH_KEY);
+ JsonNode pythonCodeNode = udfNode.get(UDF_PYTHON_CODE_KEY);
+
+ Preconditions.checkArgument(
+ classpathNode != null || pythonCodeNode != null,
+ "Missing required field \"%s\" or \"%s\" in UDF configuration",
+ UDF_CLASSPATH_KEY,
+ UDF_PYTHON_CODE_KEY);
+ Preconditions.checkArgument(
+ classpathNode == null || pythonCodeNode == null,
+ "UDF configuration cannot define both \"%s\" and \"%s\"",
+ UDF_CLASSPATH_KEY,
+ UDF_PYTHON_CODE_KEY);
+
+ JsonNode pythonExecutableNode = udfNode.get(UDF_PYTHON_EXECUTABLE_KEY);
+ JsonNode pythonFilesNode = udfNode.get(UDF_PYTHON_FILES_KEY);
Map<String, String> options =
Optional.ofNullable(udfNode.get(UDF_OPTIONS_KEY))
@@ -335,11 +355,57 @@ public class YamlPipelineDefinitionParser implements
PipelineDefinitionParser {
node ->
mapper.convertValue(
node, new
TypeReference<Map<String, String>>() {}))
- .orElse(null);
+ .orElseGet(HashMap::new);
+
+ if (pythonCodeNode != null) {
+ Preconditions.checkArgument(
+ udfNode.get(UDF_OPTIONS_KEY) == null,
+ "UDF configuration using \"%s\" cannot define \"%s\"; use
top-level \"%s\" and \"%s\" instead",
+ UDF_PYTHON_CODE_KEY,
+ UDF_OPTIONS_KEY,
+ UDF_PYTHON_EXECUTABLE_KEY,
+ UDF_PYTHON_FILES_KEY);
+ options.put("source", pythonCodeNode.asText());
+ Optional.ofNullable(pythonExecutableNode)
+ .map(JsonNode::asText)
+ .ifPresent(value -> options.put("python-executable",
value));
+ Optional.ofNullable(normalizePythonFiles(pythonFilesNode))
+ .ifPresent(value -> options.put("python-files", value));
+ return new UdfDef(functionName, PYTHON_UDF_CLASSPATH, options);
+ }
+ Preconditions.checkArgument(
+ pythonExecutableNode == null && pythonFilesNode == null,
+ "UDF configuration using \"%s\" or \"%s\" requires \"%s\"",
+ UDF_PYTHON_EXECUTABLE_KEY,
+ UDF_PYTHON_FILES_KEY,
+ UDF_PYTHON_CODE_KEY);
+
+ String classpath = classpathNode.asText();
return new UdfDef(functionName, classpath, options);
}
+ private String normalizePythonFiles(JsonNode pythonFilesNode) {
+ if (pythonFilesNode == null || pythonFilesNode.isNull()) {
+ return null;
+ }
+ if (pythonFilesNode.isTextual()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "YAML UDF field `%s` should be a list when used
with `python-code`.",
+ UDF_PYTHON_FILES_KEY));
+ }
+ if (!pythonFilesNode.isArray()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "YAML UDF field `%s` should be a list, but got
%s.",
+ UDF_PYTHON_FILES_KEY,
pythonFilesNode.getNodeType()));
+ }
+ List<String> pythonFiles = new ArrayList<>();
+ pythonFilesNode.forEach(node -> pythonFiles.add(node.asText()));
+ return String.join(",", pythonFiles);
+ }
+
private TransformDef toTransformDef(JsonNode transformNode) {
validateJsonNodeKeys(
"transform",
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 834831b6cb..a1673723c1 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
@@ -207,6 +207,139 @@ class YamlPipelineDefinitionParserTest {
assertThat(pipelineDef).isEqualTo(pipelineDefWithUdfOptions);
}
+ @Test
+ void testPythonUdfDefinition() throws Exception {
+ URL resource =
+
Resources.getResource("definitions/pipeline-definition-with-python-udf.yaml");
+ YamlPipelineDefinitionParser parser = new
YamlPipelineDefinitionParser();
+ PipelineDef pipelineDef = parser.parse(new Path(resource.toURI()), new
Configuration());
+ assertThat(pipelineDef).isEqualTo(pipelineDefWithPythonUdf);
+ }
+
+ @Test
+ void testPythonUdfRejectsClasspathAndPythonCodeTogether() {
+ YamlPipelineDefinitionParser parser = new
YamlPipelineDefinitionParser();
+ assertThatThrownBy(
+ () ->
+ parser.parse(
+ buildPipelineDefWithPythonUdf(
+ "py_identity",
+ "classpath:
org.example.MyFunction\n"
+ + " python-code:
|\n"
+ + " def eval(x:
int) -> int:\n"
+ + " return
x\n"
+ + "
python-executable: /usr/bin/python3\n"),
+ new Configuration()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "UDF configuration cannot define both \"classpath\"
and \"python-code\"");
+ }
+
+ @Test
+ void testPythonUdfRejectsOptionsAlongsidePythonCode() {
+ YamlPipelineDefinitionParser parser = new
YamlPipelineDefinitionParser();
+ assertThatThrownBy(
+ () ->
+ parser.parse(
+ buildPipelineDefWithPythonUdf(
+ "py_identity",
+ "python-code: |\n"
+ + " def eval(x:
int) -> int:\n"
+ + " return
x\n"
+ + " options:\n"
+ + "
cache.enabled: true\n"),
+ new Configuration()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "UDF configuration using \"python-code\" cannot define
\"options\"; use top-level \"python-executable\" and \"python-files\" instead");
+ }
+
+ @Test
+ void testPythonExecutableRequiresPythonCode() {
+ YamlPipelineDefinitionParser parser = new
YamlPipelineDefinitionParser();
+ assertThatThrownBy(
+ () ->
+ parser.parse(
+ buildPipelineDefWithPythonUdf(
+ "py_identity",
+ "python-executable:
/usr/bin/python3\n"
+ + " classpath:
org.example.MyFunction\n"),
+ new Configuration()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "UDF configuration using \"python-executable\" or
\"python-files\" requires \"python-code\"");
+ }
+
+ @Test
+ void testPythonFilesRequiresPythonCode() {
+ YamlPipelineDefinitionParser parser = new
YamlPipelineDefinitionParser();
+ assertThatThrownBy(
+ () ->
+ parser.parse(
+ buildPipelineDefWithPythonUdf(
+ "py_identity",
+ "python-files:\n"
+ + " -
/flink/usrlib/deps.zip\n"
+ + " classpath:
org.example.MyFunction\n"),
+ new Configuration()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "UDF configuration using \"python-executable\" or
\"python-files\" requires \"python-code\"");
+ }
+
+ @Test
+ void testPythonFilesMustUseListSyntax() {
+ YamlPipelineDefinitionParser parser = new
YamlPipelineDefinitionParser();
+ assertThatThrownBy(
+ () ->
+ parser.parse(
+ buildPipelineDefWithPythonUdf(
+ "py_identity",
+ "python-code: |\n"
+ + " def eval(x:
int) -> int:\n"
+ + " return
x\n"
+ + " python-files:
/flink/usrlib/deps.zip\n"),
+ new Configuration()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "YAML UDF field `python-files` should be a list when
used with `python-code`.");
+ }
+
+ @Test
+ void testUdfRequiresClasspathOrPythonCode() {
+ YamlPipelineDefinitionParser parser = new
YamlPipelineDefinitionParser();
+ assertThatThrownBy(
+ () ->
+ parser.parse(
+
buildPipelineDefWithPythonUdf("py_identity", ""),
+ new Configuration()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "Missing required field \"classpath\" or
\"python-code\" in UDF configuration");
+ }
+
+ @Test
+ void testPythonUdfRequiresName() {
+ YamlPipelineDefinitionParser parser = new
YamlPipelineDefinitionParser();
+ assertThatThrownBy(
+ () ->
+ parser.parse(
+ "source:\n"
+ + " type: values\n"
+ + "\n"
+ + "sink:\n"
+ + " type: values\n"
+ + "\n"
+ + "pipeline:\n"
+ + " user-defined-function:\n"
+ + " - python-code: |\n"
+ + " def eval(x: int) ->
int:\n"
+ + " return x\n",
+ new Configuration()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Missing required field \"name\" in UDF
configuration");
+ }
+
@Test
void testRouteMode() throws Exception {
URL resource =
@@ -899,6 +1032,49 @@ class YamlPipelineDefinitionParserTest {
.put("parallelism", "1")
.build()));
+ private final PipelineDef pipelineDefWithPythonUdf =
+ new PipelineDef(
+ new SourceDef("values", null, new Configuration()),
+ new SinkDef(
+ "values",
+ null,
+ new Configuration(),
+ ImmutableSet.of(
+ ALTER_TABLE_COMMENT,
+ DROP_COLUMN,
+ ALTER_COLUMN_TYPE,
+ ADD_COLUMN,
+ CREATE_TABLE,
+ RENAME_COLUMN)),
+ Collections.emptyList(),
+ Collections.singletonList(
+ new TransformDef(
+ "mydb.web_order",
+ "*, py_identity(id) as py_id",
+ null,
+ null,
+ null,
+ null,
+ ",",
+ null,
+ null)),
+ Collections.singletonList(
+ new UdfDef(
+ "py_identity",
+ "org.apache.flink.cdc.python.PythonUdf",
+ ImmutableMap.<String, String>builder()
+ .put("source", "def eval(x: int)
-> int:\n return x\n")
+ .put("python-executable",
"/usr/bin/python3")
+ .put(
+ "python-files",
+
"/flink/usrlib/deps.zip,/flink/usrlib/shared")
+ .build())),
+ Collections.emptyList(),
+ Configuration.fromMap(
+ ImmutableMap.<String, String>builder()
+ .put("parallelism", "1")
+ .build()));
+
private final PipelineDef pipelineDefWithRouteMode =
new PipelineDef(
new SourceDef(
@@ -955,4 +1131,26 @@ class YamlPipelineDefinitionParserTest {
.put("parallelism", "2")
.put("route-mode", "FIRST_MATCH")
.build()));
+
+ private static String buildPipelineDefWithPythonUdf(String name, String
udfBody) {
+ return "source:\n"
+ + " type: values\n"
+ + "\n"
+ + "sink:\n"
+ + " type: values\n"
+ + "\n"
+ + "transform:\n"
+ + " - source-table: mydb.web_order\n"
+ + " projection: \"*, "
+ + name
+ + "(id) as py_id\"\n"
+ + "\n"
+ + "pipeline:\n"
+ + " parallelism: 1\n"
+ + " user-defined-function:\n"
+ + " - name: "
+ + name
+ + "\n"
+ + (udfBody.isEmpty() ? "" : " " + udfBody);
+ }
}
diff --git
a/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-with-python-udf.yaml
b/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-with-python-udf.yaml
new file mode 100644
index 0000000000..361f60c211
--- /dev/null
+++
b/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-with-python-udf.yaml
@@ -0,0 +1,37 @@
+################################################################################
+# 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.
+################################################################################
+source:
+ type: values
+
+sink:
+ type: values
+
+transform:
+ - source-table: mydb.web_order
+ projection: "*, py_identity(id) as py_id"
+
+pipeline:
+ parallelism: 1
+ user-defined-function:
+ - name: py_identity
+ python-code: |
+ def eval(x: int) -> int:
+ return x
+ python-executable: /usr/bin/python3
+ python-files:
+ - /flink/usrlib/deps.zip
+ - /flink/usrlib/shared
diff --git
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/udf/UserDefinedFunction.java
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/udf/UserDefinedFunction.java
index 0133e43945..0fa5d78c8c 100644
---
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/udf/UserDefinedFunction.java
+++
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/udf/UserDefinedFunction.java
@@ -26,15 +26,26 @@ import org.apache.flink.cdc.common.types.DataType;
*/
@PublicEvolving
public interface UserDefinedFunction {
+
+ /**
+ * Returns the return type of this UDF when it cannot be derived from
per-UDF options.
+ *
+ * @deprecated Use {@link #getReturnType(UserDefinedFunctionContext)}
instead.
+ */
+ @Deprecated
default DataType getReturnType() {
return null;
}
+ /** Returns the return type of this UDF, given the per-UDF YAML options. */
+ default DataType getReturnType(UserDefinedFunctionContext context) {
+ return getReturnType();
+ }
+
/**
* This will be invoked every time when a UDF got created.
*
- * <p>this method is {@link Deprecated}, please use {@link
#open(UserDefinedFunctionContext)}
- * instead.
+ * @deprecated Use {@link #open(UserDefinedFunctionContext)} instead.
*/
@Deprecated
default void open() throws Exception {}
diff --git a/flink-cdc-dist/pom.xml b/flink-cdc-dist/pom.xml
index 8c918fddc9..951b52f70d 100644
--- a/flink-cdc-dist/pom.xml
+++ b/flink-cdc-dist/pom.xml
@@ -52,6 +52,11 @@ limitations under the License.
<artifactId>flink-cdc-composer</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-python</artifactId>
+ <version>${project.version}</version>
+ </dependency>
</dependencies>
<profiles>
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/PythonUdfE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/PythonUdfE2eITCase.java
new file mode 100644
index 0000000000..66c40f8205
--- /dev/null
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/PythonUdfE2eITCase.java
@@ -0,0 +1,221 @@
+/*
+ * 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.connectors.mysql.testutils.UniqueDatabase;
+import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.ExecConfig;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.images.builder.Transferable;
+
+import javax.annotation.Nullable;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/** E2e tests for pipelines that use Python UDFs. */
+class PythonUdfE2eITCase extends PipelineTestEnvironment {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(PythonUdfE2eITCase.class);
+
+ private static final String CONTAINER_PYTHON_EXECUTABLE =
"/usr/bin/python3";
+ private static final String PYTHON_FILES_DIRECTORY =
"/opt/flink/python-deps";
+ private static final String PEMJA_VERSION = "0.5.7";
+
+ private final UniqueDatabase pythonUdfTestDatabase =
+ new UniqueDatabase(MYSQL, "python_udf_test", MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD);
+
+ private final Function<String, String> databaseNameFormatter =
+ event -> String.format(event,
pythonUdfTestDatabase.getDatabaseName());
+
+ @BeforeEach
+ void initializeDatabaseAndPython() throws Exception {
+ pythonUdfTestDatabase.createAndInitialize();
+ installPythonAndPemja(taskManager);
+ preparePythonFiles(taskManager);
+ }
+
+ @AfterEach
+ void destroyDatabase() {
+ pythonUdfTestDatabase.dropDatabase();
+ }
+
+ @Test
+ void testMultiplePythonUdfsWithFilterAndPythonFiles() throws Exception {
+ String pipelineJob =
+ buildJobYaml(
+ "ID, py_normalize(EMAIL) AS EMAIL_NORM, py_double(AGE)
AS DOUBLED",
+ "py_age_ge_30(AGE)",
+ Arrays.asList(
+ new UdfEntry(
+ "py_normalize",
+ "from python_udf_helpers import
normalize_email\n"
+ + "def eval(value: str) ->
str:\n"
+ + " return
normalize_email(value)",
+ PYTHON_FILES_DIRECTORY),
+ new UdfEntry(
+ "py_double",
+ "def eval(value: int) -> int:\n"
+ + " return None if value is
None else value * 2",
+ null),
+ new UdfEntry(
+ "py_age_ge_30",
+ "def eval(value: int) -> bool:\n"
+ + " return value is not
None and value >= 30",
+ null)));
+
+ submitPipelineJob(pipelineJob);
+ waitUntilJobRunning(Duration.ofSeconds(60));
+ validateResult(
+ databaseNameFormatter,
+ "CreateTableEvent{tableId=%s.USERS, schema=columns={`ID` INT
NOT NULL,`EMAIL_NORM` STRING,`DOUBLED` BIGINT}, primaryKeys=ID, options=()}",
+ "DataChangeEvent{tableId=%s.USERS, before=[], after=[3,
[email protected], 70], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=%s.USERS, before=[], after=[4,
[email protected], 84], op=INSERT, meta=()}");
+ }
+
+ private String buildJobYaml(
+ String projection, @Nullable String filter, List<UdfEntry>
userDefinedFunctions) {
+ StringBuilder transform =
+ new StringBuilder("transform:\n")
+ .append(" - source-table: ")
+ .append(pythonUdfTestDatabase.getDatabaseName())
+ .append(".USERS\n")
+ .append(" projection: ")
+ .append(projection)
+ .append('\n');
+ if (filter != null) {
+ transform.append(" filter: ").append(filter).append('\n');
+ }
+
+ StringBuilder udfYaml = new StringBuilder();
+ for (UdfEntry udf : userDefinedFunctions) {
+ String indentedSource =
+ Arrays.stream(udf.source.split("\n", -1))
+ .map(line -> line.isEmpty() ? "" : " " +
line)
+ .collect(Collectors.joining("\n"));
+ udfYaml.append(" - name: ").append(udf.name).append('\n');
+ udfYaml.append(" python-code: |\n");
+ udfYaml.append(indentedSource).append('\n');
+ udfYaml.append(" python-executable: ")
+ .append(CONTAINER_PYTHON_EXECUTABLE)
+ .append('\n');
+ if (udf.pythonFiles != null) {
+ udfYaml.append(" python-files:\n");
+ udfYaml.append(" -
").append(udf.pythonFiles).append('\n');
+ }
+ }
+
+ return String.format(
+ "source:\n"
+ + " type: mysql\n"
+ + " hostname: %s\n"
+ + " port: 3306\n"
+ + " username: %s\n"
+ + " password: %s\n"
+ + " scan.startup.mode: earliest-offset\n"
+ + " tables: %s.USERS\n"
+ + " server-id: 5400-5404\n"
+ + " server-time-zone: UTC\n"
+ + "\n"
+ + "sink:\n"
+ + " type: values\n"
+ + "\n"
+ + "%s"
+ + "\n"
+ + "pipeline:\n"
+ + " parallelism: %d\n"
+ + " user-defined-function:\n"
+ + "%s",
+ INTER_CONTAINER_MYSQL_ALIAS,
+ MYSQL_TEST_USER,
+ MYSQL_TEST_PASSWORD,
+ pythonUdfTestDatabase.getDatabaseName(),
+ transform,
+ parallelism,
+ udfYaml);
+ }
+
+ private void installPythonAndPemja(GenericContainer<?> container) throws
Exception {
+ LOG.info(
+ "Installing Python and Pemja {} into {}",
+ PEMJA_VERSION,
+ container.getDockerImageName());
+ String externallyManagedOption =
+ flinkVersion.startsWith("2") ? "--break-system-packages " : "";
+ String script =
+ "set -euo pipefail; "
+ + "apt-get update && "
+ + "apt-get install -y --no-install-recommends python3
python3-pip python3-dev && "
+ + "rm -rf /var/lib/apt/lists/* && "
+ + "python3 -m pip install "
+ + externallyManagedOption
+ + "--disable-pip-version-check --no-cache-dir pemja=="
+ + PEMJA_VERSION
+ + " && "
+ + CONTAINER_PYTHON_EXECUTABLE
+ + " -c 'import pemja'";
+ Container.ExecResult result =
+ container.execInContainer(
+ ExecConfig.builder()
+ .user("root")
+ .command(new String[] {"bash", "-c", script})
+ .build());
+ if (result.getExitCode() != 0) {
+ throw new IllegalStateException(
+ "Failed to install Pemja into "
+ + container.getDockerImageName()
+ + " (exit="
+ + result.getExitCode()
+ + ").\nstdout:\n"
+ + result.getStdout()
+ + "\nstderr:\n"
+ + result.getStderr());
+ }
+ }
+
+ private void preparePythonFiles(GenericContainer<?> container) throws
Exception {
+ runInContainerAsRoot(container, "mkdir", "-p", PYTHON_FILES_DIRECTORY);
+ container.copyFileToContainer(
+ Transferable.of(
+ "def normalize_email(value):\n"
+ + " return None if value is None else
value.strip().lower()\n"),
+ PYTHON_FILES_DIRECTORY + "/python_udf_helpers.py");
+ }
+
+ private static final class UdfEntry {
+ private final String name;
+ private final String source;
+ @Nullable private final String pythonFiles;
+
+ private UdfEntry(String name, String source, @Nullable String
pythonFiles) {
+ this.name = name;
+ this.source = source;
+ this.pythonFiles = pythonFiles;
+ }
+ }
+}
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/resources/ddl/python_udf_test.sql
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/resources/ddl/python_udf_test.sql
new file mode 100644
index 0000000000..fba07b4449
--- /dev/null
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/resources/ddl/python_udf_test.sql
@@ -0,0 +1,37 @@
+-- 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.
+
+-- Fixture for PythonUdfE2eITCase. Rows are intentionally messy (mixed case,
+-- whitespace, varied phone formats) so the Python UDFs have something to do
+-- that built-in transform expressions can't easily match.
+
+DROP TABLE IF EXISTS USERS;
+
+CREATE TABLE USERS (
+ ID INT NOT NULL,
+ EMAIL VARCHAR(128),
+ PHONE VARCHAR(64),
+ AGE INT,
+ SCORE DOUBLE,
+ ACTIVE BOOLEAN,
+ AVATAR VARBINARY(64),
+ PRIMARY KEY (ID)
+);
+
+INSERT INTO USERS VALUES (1, '[email protected]', '+1 (415) 555-0100', 28,
87.5, TRUE, X'48656C6C6F');
+INSERT INTO USERS VALUES (2, ' [email protected] ', '(212)-555-0199', 25,
64.0, FALSE, X'576F726C64');
+INSERT INTO USERS VALUES (3, '[email protected]', '+44 20 7946 0958', 35,
92.5, TRUE, X'00FF7F');
+INSERT INTO USERS VALUES (4, '[email protected]', '+1.617.555.0173', 42,
78.0, TRUE, X'CAFEBABE');
+INSERT INTO USERS VALUES (5, NULL, NULL, NULL,
NULL, NULL, NULL);
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/resources/rules/unexpected.yaml
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/resources/rules/unexpected.yaml
index 0b9257b1a9..b2a96ce011 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/resources/rules/unexpected.yaml
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/resources/rules/unexpected.yaml
@@ -79,5 +79,5 @@ steps:
language: clojure
error: |
Unexpected key `language` in YAML UDF block.
- Allowed keys in this context are: [name, classpath, options]
+ Allowed keys in this context are: [name, classpath, python-code,
python-executable, python-files, options]
Note: option language: "clojure" is unexpected. It was silently ignored
in previous versions, and probably should be removed.
diff --git a/flink-cdc-python/pom.xml b/flink-cdc-python/pom.xml
new file mode 100644
index 0000000000..5fc3f2666d
--- /dev/null
+++ b/flink-cdc-python/pom.xml
@@ -0,0 +1,76 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ 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";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-parent</artifactId>
+ <version>${revision}</version>
+ </parent>
+
+ <artifactId>flink-cdc-python</artifactId>
+ <name>Flink CDC : Pipeline UDF : Python</name>
+
+ <properties>
+ <pemja.version>0.5.7</pemja.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-common</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.alibaba</groupId>
+ <artifactId>pemja</artifactId>
+ <version>${pemja.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>${maven.shade.plugin.version}</version>
+ <executions>
+ <execution>
+ <id>shade-flink</id>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <shadeTestJar>false</shadeTestJar>
+ <artifactSet>
+ <includes>
+ <include>com.alibaba:pemja</include>
+ </includes>
+ </artifactSet>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git
a/flink-cdc-python/src/main/java/org/apache/flink/cdc/python/PythonUdf.java
b/flink-cdc-python/src/main/java/org/apache/flink/cdc/python/PythonUdf.java
new file mode 100644
index 0000000000..4eff12ea86
--- /dev/null
+++ b/flink-cdc-python/src/main/java/org/apache/flink/cdc/python/PythonUdf.java
@@ -0,0 +1,236 @@
+/*
+ * 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.python;
+
+import org.apache.flink.cdc.common.annotation.Experimental;
+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.types.DataType;
+import org.apache.flink.cdc.common.udf.UserDefinedFunction;
+import org.apache.flink.cdc.common.udf.UserDefinedFunctionContext;
+import org.apache.flink.cdc.python.utils.PythonUdfSignature;
+
+import pemja.core.PythonInterpreter;
+import pemja.core.PythonInterpreterConfig;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Locale;
+import java.util.stream.Stream;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipInputStream;
+
+/** Generic UDF that delegates to a Python function defined inline in YAML. */
+@Experimental
+public final class PythonUdf implements UserDefinedFunction {
+
+ public static final ConfigOption<String> OPTION_SOURCE =
+ ConfigOptions.key("source")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Inline Python source containing a `def
eval(...)`.");
+
+ public static final ConfigOption<String> OPTION_PYTHON_EXECUTABLE =
+ ConfigOptions.key("python-executable")
+ .stringType()
+ .defaultValue("python3")
+ .withDescription(
+ "Path to the Python interpreter Pemja embeds on
every TaskManager."
+ + " The interpreter must have a matching
`pemja` package"
+ + " installed; defaults to the first
`python3` on PATH.");
+
+ public static final ConfigOption<String> OPTION_PYTHON_FILES =
+ ConfigOptions.key("python-files")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Comma-separated directories or zip archives that
will be added"
+ + " to the embedded Python import search
path. Zip archives"
+ + " are extracted to a temporary directory
first so packages"
+ + " with native extensions can be
imported.");
+
+ private static final String PYTHON_FUNCTION_NAME = "eval";
+
+ private transient PythonInterpreter interpreter;
+ private transient Path extractedPythonFilesDirectory;
+
+ @Override
+ public void open(UserDefinedFunctionContext context) {
+ Configuration config = context.configuration();
+ String source = requireSource(config);
+ String pythonExec = config.get(OPTION_PYTHON_EXECUTABLE);
+
+ PythonInterpreterConfig.PythonInterpreterConfigBuilder
pemjaConfigBuilder =
+ PythonInterpreterConfig.newBuilder().setPythonExec(pythonExec);
+ try {
+ configurePythonFiles(pemjaConfigBuilder, config);
+ this.interpreter = new
PythonInterpreter(pemjaConfigBuilder.build());
+ this.interpreter.exec(source);
+ } catch (Exception | Error failure) {
+ try {
+ close();
+ } catch (Exception | Error cleanupFailure) {
+ failure.addSuppressed(cleanupFailure);
+ }
+ throw failure;
+ }
+ }
+
+ @Override
+ public void close() {
+ try {
+ if (interpreter != null) {
+ interpreter.close();
+ }
+ } finally {
+ interpreter = null;
+ cleanupExtractedPythonFiles();
+ }
+ }
+
+ @Override
+ public DataType getReturnType(UserDefinedFunctionContext context) {
+ Configuration config = context.configuration();
+ String source = requireSource(config);
+ return PythonUdfSignature.parseReturnType(source,
config.get(OPTION_PYTHON_EXECUTABLE));
+ }
+
+ public Object eval(Object... args) {
+ if (interpreter == null) {
+ throw new IllegalStateException("PythonUdf invoked before open()
was called.");
+ }
+ return interpreter.invoke(PYTHON_FUNCTION_NAME, args);
+ }
+
+ private void configurePythonFiles(
+ PythonInterpreterConfig.PythonInterpreterConfigBuilder
pemjaConfigBuilder,
+ Configuration config) {
+ List<String> pythonPaths = new ArrayList<>();
+ for (String rawPath :
config.getOptional(OPTION_PYTHON_FILES).orElse("").split(",")) {
+ String path = rawPath.trim();
+ if (path.isEmpty()) {
+ continue;
+ }
+ pythonPaths.add(resolvePythonFilePath(path));
+ }
+ if (!pythonPaths.isEmpty()) {
+ pemjaConfigBuilder.addPythonPaths(String.join(File.pathSeparator,
pythonPaths));
+ }
+ }
+
+ private String resolvePythonFilePath(String configuredPath) {
+ Path path = new
File(configuredPath).toPath().toAbsolutePath().normalize();
+ if (Files.isDirectory(path)) {
+ return path.toString();
+ }
+ if (Files.isRegularFile(path)
+ &&
path.getFileName().toString().toLowerCase(Locale.ROOT).endsWith(".zip")) {
+ return extractPythonArchive(path).toString();
+ }
+ throw new IllegalArgumentException(
+ "Python UDF option '"
+ + OPTION_PYTHON_FILES.key()
+ + "' only supports existing directories or .zip
archives, but got: "
+ + configuredPath);
+ }
+
+ private Path extractPythonArchive(Path archivePath) {
+ try {
+ if (extractedPythonFilesDirectory == null) {
+ extractedPythonFilesDirectory =
Files.createTempDirectory("python-udf-files-");
+ }
+ String archiveName = archivePath.getFileName().toString();
+ int suffixIndex =
archiveName.toLowerCase(Locale.ROOT).lastIndexOf(".zip");
+ String targetDirectoryName =
+ suffixIndex > 0 ? archiveName.substring(0, suffixIndex) :
archiveName;
+ Path targetDirectory =
+ Files.createTempDirectory(
+ extractedPythonFilesDirectory, targetDirectoryName
+ "-");
+ unzipArchive(archivePath, targetDirectory);
+ return targetDirectory;
+ } catch (IOException e) {
+ throw new IllegalArgumentException(
+ "Failed to extract Python dependency archive: " +
archivePath, e);
+ }
+ }
+
+ private static void unzipArchive(Path archivePath, Path targetDirectory)
throws IOException {
+ try (InputStream inputStream = Files.newInputStream(archivePath);
+ ZipInputStream zipInputStream = new
ZipInputStream(inputStream)) {
+ ZipEntry zipEntry;
+ while ((zipEntry = zipInputStream.getNextEntry()) != null) {
+ Path targetPath =
targetDirectory.resolve(zipEntry.getName()).normalize();
+ if (!targetPath.startsWith(targetDirectory)) {
+ throw new IOException(
+ "Zip entry escapes extraction directory: " +
zipEntry.getName());
+ }
+ if (zipEntry.isDirectory()) {
+ Files.createDirectories(targetPath);
+ } else {
+ Path parent = targetPath.getParent();
+ if (parent != null) {
+ Files.createDirectories(parent);
+ }
+ Files.copy(zipInputStream, targetPath,
StandardCopyOption.REPLACE_EXISTING);
+ }
+ zipInputStream.closeEntry();
+ }
+ }
+ }
+
+ private void cleanupExtractedPythonFiles() {
+ if (extractedPythonFilesDirectory == null) {
+ return;
+ }
+ try {
+ Path[] filesToDelete;
+ try (Stream<Path> files =
Files.walk(extractedPythonFilesDirectory)) {
+ filesToDelete =
files.sorted(Comparator.reverseOrder()).toArray(Path[]::new);
+ }
+ for (Path path : filesToDelete) {
+ try {
+ Files.deleteIfExists(path);
+ } catch (IOException ignored) {
+ // Best-effort cleanup only.
+ }
+ }
+ } catch (IOException ignored) {
+ // Best-effort cleanup only.
+ } finally {
+ extractedPythonFilesDirectory = null;
+ }
+ }
+
+ private static String requireSource(Configuration config) {
+ return config.getOptional(OPTION_SOURCE)
+ .orElseThrow(
+ () ->
+ new IllegalArgumentException(
+ "Python UDF is missing required option
'"
+ + OPTION_SOURCE.key()
+ + "'."));
+ }
+}
diff --git
a/flink-cdc-python/src/main/java/org/apache/flink/cdc/python/utils/PythonUdfSignature.java
b/flink-cdc-python/src/main/java/org/apache/flink/cdc/python/utils/PythonUdfSignature.java
new file mode 100644
index 0000000000..a9a3976aa5
--- /dev/null
+++
b/flink-cdc-python/src/main/java/org/apache/flink/cdc/python/utils/PythonUdfSignature.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.python.utils;
+
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+
+import pemja.core.PythonInterpreter;
+import pemja.core.PythonInterpreterConfig;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/** Resolves the return type of a {@code PythonUdf} from its inline Python
source. */
+public final class PythonUdfSignature {
+
+ private static final String SIGNATURE_PARSER_RESOURCE =
+ "/org/apache/flink/cdc/python/signature.py";
+
+ private static final Map<String, DataType> SUPPORTED_ANNOTATIONS =
+ Map.of(
+ "bool", DataTypes.BOOLEAN(),
+ "int", DataTypes.BIGINT(),
+ "float", DataTypes.DOUBLE(),
+ "str", DataTypes.STRING(),
+ "bytes", DataTypes.BYTES());
+
+ private PythonUdfSignature() {}
+
+ public static DataType parseReturnType(String source, String pythonExec) {
+ String annotation = parseReturnAnnotation(source, pythonExec).trim();
+ DataType type = SUPPORTED_ANNOTATIONS.get(annotation);
+ if (type == null) {
+ throw new IllegalArgumentException(
+ "Unsupported Python UDF return type '"
+ + annotation
+ + "'. Supported annotations: "
+ + SUPPORTED_ANNOTATIONS.keySet().stream()
+ .sorted()
+ .collect(Collectors.toList()));
+ }
+ return type;
+ }
+
+ private static String parseReturnAnnotation(String source, String
pythonExec) {
+ PythonInterpreterConfig pemjaConfig =
+
PythonInterpreterConfig.newBuilder().setPythonExec(pythonExec).build();
+ try (PythonInterpreter parser = new PythonInterpreter(pemjaConfig)) {
+ parser.exec(loadSignatureScript());
+ return (String) parser.invoke("eval_return_type", source);
+ }
+ }
+
+ private static String loadSignatureScript() {
+ try (InputStream in =
+
PythonUdfSignature.class.getResourceAsStream(SIGNATURE_PARSER_RESOURCE)) {
+ if (in == null) {
+ throw new IllegalStateException(
+ "Bundled signature parser resource not found: "
+ + SIGNATURE_PARSER_RESOURCE);
+ }
+ return new String(in.readAllBytes(), StandardCharsets.UTF_8);
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Failed to parse Python UDF
signature", e);
+ }
+ }
+}
diff --git
a/flink-cdc-python/src/main/resources/org/apache/flink/cdc/python/signature.py
b/flink-cdc-python/src/main/resources/org/apache/flink/cdc/python/signature.py
new file mode 100644
index 0000000000..9bdc11f784
--- /dev/null
+++
b/flink-cdc-python/src/main/resources/org/apache/flink/cdc/python/signature.py
@@ -0,0 +1,54 @@
+################################################################################
+# 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.
+################################################################################
+"""Resolve the Calcite return type of a Python UDF from its inline source."""
+
+import ast
+
+
+def eval_return_type(source):
+ """Return the raw return-type annotation string of the top-level ``eval``
+ function. Raises ``ValueError`` if it can't be determined.
+ """
+ for node in ast.parse(source).body:
+ if isinstance(node, ast.FunctionDef) and node.name == 'eval':
+ if node.returns is None:
+ raise ValueError(
+ "Function 'eval' has no return type annotation."
+ )
+ annotation = _annotation_string(node.returns)
+ if annotation is None:
+ raise ValueError(
+ "Return type annotation of 'eval' could not be rendered "
+ "(needs Python 3.9+ for non-trivial annotations)."
+ )
+ return annotation
+ raise ValueError(
+ "Python UDF source does not define a top-level 'eval' function."
+ )
+
+
+def _annotation_string(annotation):
+ if isinstance(annotation, ast.Name):
+ return annotation.id
+ unparse = getattr(ast, 'unparse', None)
+ if unparse is None:
+ return None
+ try:
+ return unparse(annotation)
+ except (AttributeError, TypeError):
+ return None
diff --git
a/flink-cdc-python/src/test/java/org/apache/flink/cdc/python/PythonUdfTest.java
b/flink-cdc-python/src/test/java/org/apache/flink/cdc/python/PythonUdfTest.java
new file mode 100644
index 0000000000..fdff958616
--- /dev/null
+++
b/flink-cdc-python/src/test/java/org/apache/flink/cdc/python/PythonUdfTest.java
@@ -0,0 +1,219 @@
+/*
+ * 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.python;
+
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.common.udf.UserDefinedFunctionContext;
+import org.apache.flink.cdc.python.utils.PemjaTestSupport;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipOutputStream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link PythonUdf}. */
+class PythonUdfTest {
+
+ @BeforeAll
+ static void requirePemja() {
+ PemjaTestSupport.requirePemja();
+ }
+
+ @Test
+ void evalBeforeOpenThrows() {
+ PythonUdf udf = new PythonUdf();
+ assertThatThrownBy(() -> udf.eval(1L))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("before open()");
+ }
+
+ @Test
+ void closeIsIdempotent() throws Exception {
+ PythonUdf udf = new PythonUdf();
+ udf.close();
+ udf.close();
+ try {
+ udf.open(contextFor("def eval(x: int) -> int:\n return x\n"));
+ udf.close();
+ udf.close();
+ } finally {
+ udf.close();
+ }
+ }
+
+ @Test
+ void openRequiresSourceOption() {
+ PythonUdf udf = new PythonUdf();
+ Map<String, String> opts = new HashMap<>();
+ opts.put(PythonUdf.OPTION_PYTHON_EXECUTABLE.key(),
PemjaTestSupport.PYTHON_EXEC);
+ assertThatThrownBy(() -> udf.open(() -> Configuration.fromMap(opts)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(PythonUdf.OPTION_SOURCE.key());
+ }
+
+ @Test
+ void getReturnTypeReadsFromSource() {
+ PythonUdf udf = new PythonUdf();
+ assertThat(udf.getReturnType(contextFor("def eval(x: int) -> int:\n
return x * 2\n")))
+ .isEqualTo(DataTypes.BIGINT());
+ }
+
+ @Test
+ void evalRoundTripsInt() throws Exception {
+ PythonUdf udf = new PythonUdf();
+ udf.open(contextFor("def eval(x: int) -> int:\n return x * 2\n"));
+ try {
+ assertThat(udf.eval(21L)).isEqualTo(42L);
+ } finally {
+ udf.close();
+ }
+ }
+
+ @Test
+ void evalRoundTripsString() throws Exception {
+ PythonUdf udf = new PythonUdf();
+ udf.open(contextFor("def eval(s: str) -> str:\n return
s.upper()\n"));
+ try {
+ assertThat(udf.eval("abc")).isEqualTo("ABC");
+ } finally {
+ udf.close();
+ }
+ }
+
+ @Test
+ void evalForwardsNullToPython() throws Exception {
+ // Pemja maps Java null -> Python None; a guard-clause UDF should see
it and may handle it.
+ PythonUdf udf = new PythonUdf();
+ udf.open(
+ contextFor(
+ "def eval(s) -> str:\n" + " return 'null' if s is
None else str(s)\n"));
+ try {
+ assertThat(udf.eval(new Object[] {null})).isEqualTo("null");
+ } finally {
+ udf.close();
+ }
+ }
+
+ @Test
+ void evalImportsModuleFromDirectory(@TempDir Path tempDir) throws
Exception {
+ Path moduleDir = tempDir.resolve("python-dir");
+ Files.createDirectories(moduleDir);
+ Files.write(
+ moduleDir.resolve("helper_mod.py"),
+ ("def twice(x):\n" + " return x *
2\n").getBytes(StandardCharsets.UTF_8));
+
+ PythonUdf udf = new PythonUdf();
+ udf.open(
+ contextFor(
+ "import helper_mod\n"
+ + "def eval(x: int) -> int:\n"
+ + " return helper_mod.twice(x)\n",
+ moduleDir.toString()));
+ try {
+ assertThat(udf.eval(21L)).isEqualTo(42L);
+ } finally {
+ udf.close();
+ }
+ }
+
+ @Test
+ void evalImportsModuleFromZip(@TempDir Path tempDir) throws Exception {
+ Path zipFile = tempDir.resolve("python-deps.zip");
+ try (ZipOutputStream zipOutputStream =
+ new ZipOutputStream(Files.newOutputStream(zipFile))) {
+ zipOutputStream.putNextEntry(new ZipEntry("helper_zip.py"));
+ zipOutputStream.write(
+ ("def shout(s):\n" + " return s.upper()\n")
+ .getBytes(StandardCharsets.UTF_8));
+ zipOutputStream.closeEntry();
+ }
+
+ PythonUdf udf = new PythonUdf();
+ udf.open(
+ contextFor(
+ "import helper_zip\n"
+ + "def eval(s: str) -> str:\n"
+ + " return helper_zip.shout(s)\n",
+ zipFile.toString()));
+ try {
+ assertThat(udf.eval("abc")).isEqualTo("ABC");
+ } finally {
+ udf.close();
+ }
+ }
+
+ @Test
+ void openFailureCleansUpInterpreterAndExtractedPythonFiles(@TempDir Path
tempDir)
+ throws Exception {
+ Path zipFile = tempDir.resolve("python-deps.zip");
+ try (ZipOutputStream zipOutputStream =
+ new ZipOutputStream(Files.newOutputStream(zipFile))) {
+ zipOutputStream.putNextEntry(new ZipEntry("helper.py"));
+ zipOutputStream.write("VALUE =
1\n".getBytes(StandardCharsets.UTF_8));
+ zipOutputStream.closeEntry();
+ }
+
+ Path extractedPathMarker = tempDir.resolve("extracted-path.txt");
+ String markerPath =
+ extractedPathMarker.toString().replace("\\",
"\\\\").replace("'", "\\'");
+ String source =
+ "import sys\n"
+ + "from pathlib import Path\n"
+ + "extracted = next(p for p in sys.path if
'python-udf-files-' in p)\n"
+ + "Path('"
+ + markerPath
+ + "').write_text(extracted)\n"
+ + "raise RuntimeError('expected open failure')\n";
+
+ PythonUdf udf = new PythonUdf();
+ assertThatThrownBy(() -> udf.open(contextFor(source,
zipFile.toString())))
+ .hasMessageContaining("expected open failure");
+
+ Path extractedPath = Paths.get(Files.readString(extractedPathMarker));
+ assertThat(extractedPath).doesNotExist();
+ assertThat(udf)
+ .extracting("interpreter", "extractedPythonFilesDirectory")
+ .containsExactly(null, null);
+ }
+
+ private static UserDefinedFunctionContext contextFor(String source) {
+ return contextFor(source, null);
+ }
+
+ private static UserDefinedFunctionContext contextFor(String source, String
pythonFiles) {
+ Map<String, String> opts = new HashMap<>();
+ opts.put(PythonUdf.OPTION_SOURCE.key(), source);
+ opts.put(PythonUdf.OPTION_PYTHON_EXECUTABLE.key(),
PemjaTestSupport.PYTHON_EXEC);
+ if (pythonFiles != null) {
+ opts.put(PythonUdf.OPTION_PYTHON_FILES.key(), pythonFiles);
+ }
+ return () -> Configuration.fromMap(opts);
+ }
+}
diff --git
a/flink-cdc-python/src/test/java/org/apache/flink/cdc/python/utils/PemjaTestSupport.java
b/flink-cdc-python/src/test/java/org/apache/flink/cdc/python/utils/PemjaTestSupport.java
new file mode 100644
index 0000000000..e852da68a3
--- /dev/null
+++
b/flink-cdc-python/src/test/java/org/apache/flink/cdc/python/utils/PemjaTestSupport.java
@@ -0,0 +1,53 @@
+/*
+ * 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.python.utils;
+
+import pemja.core.PythonInterpreter;
+import pemja.core.PythonInterpreterConfig;
+
+/**
+ * Fails fast with an actionable message if Pemja or its Python sidecar isn't
reachable. The
+ * interpreter is resolved from {@code pemja.python.executable} (system
property) or {@code
+ * PEMJA_PYTHON_EXECUTABLE} (env var), falling back to {@code python3} on PATH.
+ */
+public final class PemjaTestSupport {
+
+ public static final String PYTHON_EXEC =
+ System.getProperty(
+ "pemja.python.executable",
+ System.getenv().getOrDefault("PEMJA_PYTHON_EXECUTABLE",
"python3"));
+
+ private PemjaTestSupport() {}
+
+ public static void requirePemja() {
+ try (PythonInterpreter ignored =
+ new PythonInterpreter(
+
PythonInterpreterConfig.newBuilder().setPythonExec(PYTHON_EXEC).build())) {
+ // Smoke-test: starting the interpreter loads libpython + the
pemja package.
+ } catch (Throwable t) {
+ throw new IllegalStateException(
+ "Pemja could not be initialized using '"
+ + PYTHON_EXEC
+ + "'. Install Python 3.9+ and `pip install
pemja==0.5.7`, or point"
+ + " the tests at a usable interpreter via"
+ + " -Dpemja.python.executable=/path/to/python3"
+ + " (or PEMJA_PYTHON_EXECUTABLE env var).",
+ t);
+ }
+ }
+}
diff --git
a/flink-cdc-python/src/test/java/org/apache/flink/cdc/python/utils/PythonUdfSignatureTest.java
b/flink-cdc-python/src/test/java/org/apache/flink/cdc/python/utils/PythonUdfSignatureTest.java
new file mode 100644
index 0000000000..090f8e2ac0
--- /dev/null
+++
b/flink-cdc-python/src/test/java/org/apache/flink/cdc/python/utils/PythonUdfSignatureTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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.python.utils;
+
+import org.apache.flink.cdc.common.types.DataTypes;
+
+import org.junit.jupiter.api.BeforeAll;
+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 PythonUdfSignature}. */
+class PythonUdfSignatureTest {
+
+ @BeforeAll
+ static void requirePemja() {
+ PemjaTestSupport.requirePemja();
+ }
+
+ @Test
+ void resolvesEvenWhenOtherFunctionsArePresent() {
+ String source =
+ "def helper(x: int) -> int:\n"
+ + " return x\n"
+ + "def eval(x: str) -> str:\n"
+ + " return helper(len(x)) and x\n";
+ assertThat(PythonUdfSignature.parseReturnType(source,
PemjaTestSupport.PYTHON_EXEC))
+ .isEqualTo(DataTypes.STRING());
+ }
+
+ @Test
+ void throwsWhenNoEvalFunction() {
+ String source = "def other(x: int) -> int:\n return x\n";
+ assertThatThrownBy(
+ () ->
+ PythonUdfSignature.parseReturnType(
+ source, PemjaTestSupport.PYTHON_EXEC))
+ .hasMessageContaining(
+ "Python UDF source does not define a top-level 'eval'
function.");
+ }
+
+ @Test
+ void throwsWhenEvalHasNoReturnAnnotation() {
+ String source = "def eval(x: int):\n return x\n";
+ assertThatThrownBy(
+ () ->
+ PythonUdfSignature.parseReturnType(
+ source, PemjaTestSupport.PYTHON_EXEC))
+ .hasMessageContaining("Function 'eval' has no return type
annotation.");
+ }
+
+ @Test
+ void throwsWhenAnnotationUnsupported() {
+ String source = "def eval(x: int) -> bytearray:\n return
bytearray(x)\n";
+ assertThatThrownBy(
+ () ->
+ PythonUdfSignature.parseReturnType(
+ source, PemjaTestSupport.PYTHON_EXEC))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Unsupported Python UDF return type
'bytearray'.")
+ .hasMessageContaining("Supported annotations: [bool, bytes,
float, int, str]");
+ }
+
+ @Test
+ void throwsOnInvalidPythonSource() {
+ // Not "no eval" — outright syntax error so ast.parse blows up inside
the parser.
+ String source = "def eval(x ->\n";
+ assertThatThrownBy(
+ () ->
+ PythonUdfSignature.parseReturnType(
+ source, PemjaTestSupport.PYTHON_EXEC))
+ .hasMessageContaining("invalid syntax");
+ }
+}
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 776b4af2d0..47e42eb90b 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
@@ -81,7 +81,7 @@ public class TransformExpressionCompiler {
List<Class<?>> argumentClasses = new
ArrayList<>(key.getArgumentClasses());
for (UserDefinedFunctionDescriptor udfFunction :
udfDescriptors) {
- argumentNames.add("__instanceOf" +
udfFunction.getClassName());
+ argumentNames.add("__udf_" +
udfFunction.getName());
argumentClasses.add(Class.forName(udfFunction.getClasspath()));
}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/UserDefinedFunctionDescriptor.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/UserDefinedFunctionDescriptor.java
index 4e792e8719..4df805f88b 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/UserDefinedFunctionDescriptor.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/UserDefinedFunctionDescriptor.java
@@ -19,8 +19,10 @@ package org.apache.flink.cdc.runtime.operators.transform;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.configuration.Configuration;
import org.apache.flink.cdc.common.types.DataType;
import org.apache.flink.cdc.common.udf.UserDefinedFunction;
+import org.apache.flink.cdc.common.udf.UserDefinedFunctionContext;
import java.io.Serializable;
import java.lang.reflect.InvocationTargetException;
@@ -62,13 +64,10 @@ public class UserDefinedFunctionDescriptor implements
Serializable {
Class<?> clazz = Class.forName(classpath);
isCdcPipelineUdf = isCdcPipelineUdf(clazz);
if (isCdcPipelineUdf) {
- // We use reflection to invoke UDF methods since we may add
more methods
- // into UserDefinedFunction interface, thus the provided UDF
classes
- // might not be compatible with the interface definition in
CDC common.
+ UserDefinedFunctionContext context = () ->
Configuration.fromMap(parameters);
returnTypeHint =
- (DataType)
- clazz.getMethod("getReturnType")
-
.invoke(clazz.getConstructor().newInstance());
+ ((UserDefinedFunction)
clazz.getConstructor().newInstance())
+ .getReturnType(context);
} else {
returnTypeHint = null;
}
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 d03664a35e..1497d6cda2 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
@@ -1256,12 +1256,12 @@ public class JaninoCompiler {
private static String
generateInvokeExpression(UserDefinedFunctionDescriptor udfFunction) {
if (udfFunction.getReturnTypeHint() != null) {
return String.format(
- "(%s) __instanceOf%s.eval",
+ "(%s) __udf_%s.eval",
JavaClassConverter.toJavaClass(udfFunction.getReturnTypeHint())
.getCanonicalName(),
- udfFunction.getClassName());
+ udfFunction.getName());
} else {
- return String.format("__instanceOf%s.eval",
udfFunction.getClassName());
+ return String.format("__udf_%s.eval", udfFunction.getName());
}
}
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/UserDefinedFunctionDescriptorTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/UserDefinedFunctionDescriptorTest.java
index 79f9c92af5..3ca5ecac15 100644
---
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/UserDefinedFunctionDescriptorTest.java
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/UserDefinedFunctionDescriptorTest.java
@@ -20,12 +20,15 @@ package org.apache.flink.cdc.runtime.operators.transform;
import org.apache.flink.cdc.common.types.DataType;
import org.apache.flink.cdc.common.types.DataTypes;
import org.apache.flink.cdc.common.udf.UserDefinedFunction;
+import org.apache.flink.cdc.common.udf.UserDefinedFunctionContext;
import org.apache.flink.cdc.runtime.model.OpenAIEmbeddingModel;
import org.apache.flink.table.functions.ScalarFunction;
import com.fasterxml.jackson.core.JsonProcessingException;
import org.junit.jupiter.api.Test;
+import java.util.Collections;
+
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -43,6 +46,16 @@ class UserDefinedFunctionDescriptorTest {
}
}
+ /** This is a Flink CDC UDF whose type hint depends on per-UDF options. */
+ public static class CdcUdfWithOptionTypeHint implements
UserDefinedFunction {
+ @Override
+ public DataType getReturnType(UserDefinedFunctionContext context) {
+ return
"string".equals(context.configuration().toMap().get("return-type"))
+ ? DataTypes.STRING()
+ : DataTypes.BIGINT();
+ }
+ }
+
/** This is a Flink ScalarFunction. */
public static class FlinkUdf extends ScalarFunction {}
@@ -105,4 +118,15 @@ class UserDefinedFunctionDescriptorTest {
DataTypes.ARRAY(DataTypes.FLOAT()),
true);
}
+
+ @Test
+ void testReturnTypeHintUsesUdfOptions() {
+ UserDefinedFunctionDescriptor descriptor =
+ new UserDefinedFunctionDescriptor(
+ "option_type_hint",
+ CdcUdfWithOptionTypeHint.class.getName(),
+ Collections.singletonMap("return-type", "string"));
+
+
assertThat(descriptor.getReturnTypeHint()).isEqualTo(DataTypes.STRING());
+ }
}
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
index 16136323f7..50fcc1e455 100644
---
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
@@ -1166,50 +1166,34 @@ class TransformParserTest {
void testTranslateUdfFilterToJaninoExpression() {
testFilterExpressionWithUdf(
"IFNULL(format(id), 'fallback')",
- "(java.lang.String)
ifNull(__instanceOfFormatFunctionClass.eval(id), \"fallback\")");
+ "(java.lang.String) ifNull(__udf_format.eval(id),
\"fallback\")");
testFilterExpressionWithUdf(
"NULLIF(format(id), '1')",
- "(java.lang.String)
nullIf(__instanceOfFormatFunctionClass.eval(id), \"1\")");
+ "(java.lang.String) nullIf(__udf_format.eval(id), \"1\")");
+ testFilterExpressionWithUdf("format(upper(id))",
"__udf_format.eval(upper(id))");
+ testFilterExpressionWithUdf("format(lower(id))",
"__udf_format.eval(lower(id))");
+ testFilterExpressionWithUdf("format(concat(a,b))",
"__udf_format.eval(concat(a, b))");
+ testFilterExpressionWithUdf("format(SUBSTR(a,1))",
"__udf_format.eval(substr(a, 1))");
testFilterExpressionWithUdf(
- "format(upper(id))",
"__instanceOfFormatFunctionClass.eval(upper(id))");
+ "typeof(id like '^[a-zA-Z]')", "__udf_typeof.eval(like(id,
\"^[a-zA-Z]\"))");
testFilterExpressionWithUdf(
- "format(lower(id))",
"__instanceOfFormatFunctionClass.eval(lower(id))");
- testFilterExpressionWithUdf(
- "format(concat(a,b))",
"__instanceOfFormatFunctionClass.eval(concat(a, b))");
- testFilterExpressionWithUdf(
- "format(SUBSTR(a,1))",
"__instanceOfFormatFunctionClass.eval(substr(a, 1))");
- testFilterExpressionWithUdf(
- "typeof(id like '^[a-zA-Z]')",
- "__instanceOfTypeOfFunctionClass.eval(like(id,
\"^[a-zA-Z]\"))");
- testFilterExpressionWithUdf(
- "typeof(id not like '^[a-zA-Z]')",
- "__instanceOfTypeOfFunctionClass.eval(notLike(id,
\"^[a-zA-Z]\"))");
- testFilterExpressionWithUdf(
- "typeof(abs(2))",
"__instanceOfTypeOfFunctionClass.eval(abs(2))");
- testFilterExpressionWithUdf(
- "typeof(ceil(2))",
"__instanceOfTypeOfFunctionClass.eval(ceil(2))");
- testFilterExpressionWithUdf(
- "typeof(ceiling(2))",
"__instanceOfTypeOfFunctionClass.eval(ceil(2))");
- testFilterExpressionWithUdf(
- "typeof(floor(2))",
"__instanceOfTypeOfFunctionClass.eval(floor(2))");
- testFilterExpressionWithUdf(
- "typeof(round(2,2))",
"__instanceOfTypeOfFunctionClass.eval(round(2, 2))");
- testFilterExpressionWithUdf(
- "typeof(id + 2)", "__instanceOfTypeOfFunctionClass.eval(id +
2)");
- testFilterExpressionWithUdf(
- "typeof(id - 2)", "__instanceOfTypeOfFunctionClass.eval(id -
2)");
- testFilterExpressionWithUdf(
- "typeof(id * 2)", "__instanceOfTypeOfFunctionClass.eval(id *
2)");
- testFilterExpressionWithUdf(
- "typeof(id / 2)", "__instanceOfTypeOfFunctionClass.eval(id /
2)");
- testFilterExpressionWithUdf(
- "typeof(id % 2)", "__instanceOfTypeOfFunctionClass.eval(id %
2)");
+ "typeof(id not like '^[a-zA-Z]')",
"__udf_typeof.eval(notLike(id, \"^[a-zA-Z]\"))");
+ testFilterExpressionWithUdf("typeof(abs(2))",
"__udf_typeof.eval(abs(2))");
+ testFilterExpressionWithUdf("typeof(ceil(2))",
"__udf_typeof.eval(ceil(2))");
+ testFilterExpressionWithUdf("typeof(ceiling(2))",
"__udf_typeof.eval(ceil(2))");
+ testFilterExpressionWithUdf("typeof(floor(2))",
"__udf_typeof.eval(floor(2))");
+ testFilterExpressionWithUdf("typeof(round(2,2))",
"__udf_typeof.eval(round(2, 2))");
+ testFilterExpressionWithUdf("typeof(id + 2)", "__udf_typeof.eval(id +
2)");
+ testFilterExpressionWithUdf("typeof(id - 2)", "__udf_typeof.eval(id -
2)");
+ testFilterExpressionWithUdf("typeof(id * 2)", "__udf_typeof.eval(id *
2)");
+ testFilterExpressionWithUdf("typeof(id / 2)", "__udf_typeof.eval(id /
2)");
+ testFilterExpressionWithUdf("typeof(id % 2)", "__udf_typeof.eval(id %
2)");
testFilterExpressionWithUdf(
"addone(addone(id)) > 4 OR typeof(id) <> 'bool' AND
format('from %s to %s is %s', 'a', 'z', 'lie') <> ''",
-
"greaterThan(__instanceOfAddOneFunctionClass.eval(__instanceOfAddOneFunctionClass.eval(id)),
4) || !valueEquals(__instanceOfTypeOfFunctionClass.eval(id), \"bool\") &&
!valueEquals(__instanceOfFormatFunctionClass.eval(\"from %s to %s is %s\",
\"a\", \"z\", \"lie\"), \"\")");
+ "greaterThan(__udf_addone.eval(__udf_addone.eval(id)), 4) ||
!valueEquals(__udf_typeof.eval(id), \"bool\") &&
!valueEquals(__udf_format.eval(\"from %s to %s is %s\", \"a\", \"z\", \"lie\"),
\"\")");
testFilterExpressionWithUdf(
"ADDONE(ADDONE(id)) > 4 OR TYPEOF(id) <> 'bool' AND
FORMAT('from %s to %s is %s', 'a', 'z', 'lie') <> ''",
-
"greaterThan(__instanceOfAddOneFunctionClass.eval(__instanceOfAddOneFunctionClass.eval(id)),
4) || !valueEquals(__instanceOfTypeOfFunctionClass.eval(id), \"bool\") &&
!valueEquals(__instanceOfFormatFunctionClass.eval(\"from %s to %s is %s\",
\"a\", \"z\", \"lie\"), \"\")");
+ "greaterThan(__udf_addone.eval(__udf_addone.eval(id)), 4) ||
!valueEquals(__udf_typeof.eval(id), \"bool\") &&
!valueEquals(__udf_format.eval(\"from %s to %s is %s\", \"a\", \"z\", \"lie\"),
\"\")");
}
@Test
@@ -1228,19 +1212,19 @@ class TransformParserTest {
testFilterExpressionWithUdf(
"IFNULL(id)",
- "__instanceOfAddOneFunctionClass.eval(id)",
+ "__udf_ifnull.eval(id)",
DUMMY_COLUMNS,
Collections.emptyMap(),
udfDescriptors);
testFilterExpressionWithUdf(
"TRY_CAST(id)",
- "__instanceOfTypeOfFunctionClass.eval(id)",
+ "__udf_try_cast.eval(id)",
DUMMY_COLUMNS,
Collections.emptyMap(),
udfDescriptors);
testFilterExpressionWithUdf(
"NULLIF('%s', 'udf')",
- "__instanceOfFormatFunctionClass.eval(\"%s\", \"udf\")",
+ "__udf_nullif.eval(\"%s\", \"udf\")",
DUMMY_COLUMNS,
Collections.emptyMap(),
udfDescriptors);
@@ -1260,53 +1244,38 @@ class TransformParserTest {
columnNameMap.put("a-b", "$2");
testFilterExpressionWithUdf(
- "format(upper(a))",
- "__instanceOfFormatFunctionClass.eval(upper($0))",
- columns,
- columnNameMap);
+ "format(upper(a))", "__udf_format.eval(upper($0))", columns,
columnNameMap);
testFilterExpressionWithUdf(
- "format(lower(b))",
- "__instanceOfFormatFunctionClass.eval(lower($1))",
- columns,
- columnNameMap);
+ "format(lower(b))", "__udf_format.eval(lower($1))", columns,
columnNameMap);
testFilterExpressionWithUdf(
- "format(concat(a,b))",
- "__instanceOfFormatFunctionClass.eval(concat($0, $1))",
- columns,
- columnNameMap);
+ "format(concat(a,b))", "__udf_format.eval(concat($0, $1))",
columns, columnNameMap);
testFilterExpressionWithUdf(
"format(SUBSTR(`a-b`,1))",
- "__instanceOfFormatFunctionClass.eval(substr($2, 1))",
+ "__udf_format.eval(substr($2, 1))",
columns,
columnNameMap);
testFilterExpressionWithUdf(
"typeof(`a-b` like '^[a-zA-Z]')",
- "__instanceOfTypeOfFunctionClass.eval(like($2,
\"^[a-zA-Z]\"))",
+ "__udf_typeof.eval(like($2, \"^[a-zA-Z]\"))",
columns,
columnNameMap);
testFilterExpressionWithUdf(
"typeof(`a-b` not like '^[a-zA-Z]')",
- "__instanceOfTypeOfFunctionClass.eval(notLike($2,
\"^[a-zA-Z]\"))",
+ "__udf_typeof.eval(notLike($2, \"^[a-zA-Z]\"))",
columns,
columnNameMap);
testFilterExpressionWithUdf(
- "typeof(a-b-`a-b`)",
- "__instanceOfTypeOfFunctionClass.eval($0 - $1 - $2)",
- columns,
- columnNameMap);
+ "typeof(a-b-`a-b`)", "__udf_typeof.eval($0 - $1 - $2)",
columns, columnNameMap);
testFilterExpressionWithUdf(
- "typeof(a-b-2)",
- "__instanceOfTypeOfFunctionClass.eval($0 - $1 - 2)",
- columns,
- columnNameMap);
+ "typeof(a-b-2)", "__udf_typeof.eval($0 - $1 - 2)", columns,
columnNameMap);
testFilterExpressionWithUdf(
"addone(addone(`a-b`)) > 4 OR typeof(a-b) <> 'bool' AND
format('from %s to %s is %s', 'a', 'z', 'lie') <> ''",
-
"greaterThan(__instanceOfAddOneFunctionClass.eval(__instanceOfAddOneFunctionClass.eval($2)),
4) || !valueEquals(__instanceOfTypeOfFunctionClass.eval($0 - $1), \"bool\") &&
!valueEquals(__instanceOfFormatFunctionClass.eval(\"from %s to %s is %s\",
\"a\", \"z\", \"lie\"), \"\")",
+ "greaterThan(__udf_addone.eval(__udf_addone.eval($2)), 4) ||
!valueEquals(__udf_typeof.eval($0 - $1), \"bool\") &&
!valueEquals(__udf_format.eval(\"from %s to %s is %s\", \"a\", \"z\", \"lie\"),
\"\")",
columns,
columnNameMap);
testFilterExpressionWithUdf(
"ADDONE(ADDONE(`a-b`)) > 4 OR TYPEOF(a-b) <> 'bool' AND
FORMAT('from %s to %s is %s', 'a', 'z', 'lie') <> ''",
-
"greaterThan(__instanceOfAddOneFunctionClass.eval(__instanceOfAddOneFunctionClass.eval($2)),
4) || !valueEquals(__instanceOfTypeOfFunctionClass.eval($0 - $1), \"bool\") &&
!valueEquals(__instanceOfFormatFunctionClass.eval(\"from %s to %s is %s\",
\"a\", \"z\", \"lie\"), \"\")",
+ "greaterThan(__udf_addone.eval(__udf_addone.eval($2)), 4) ||
!valueEquals(__udf_typeof.eval($0 - $1), \"bool\") &&
!valueEquals(__udf_format.eval(\"from %s to %s is %s\", \"a\", \"z\", \"lie\"),
\"\")",
columns,
columnNameMap);
}
diff --git a/pom.xml b/pom.xml
index 54e5f69c34..32d3d021fd 100644
--- a/pom.xml
+++ b/pom.xml
@@ -43,6 +43,7 @@ limitations under the License.
<module>flink-cdc-runtime</module>
<module>flink-cdc-e2e-tests</module>
<module>flink-cdc-pipeline-udf-examples</module>
+ <module>flink-cdc-python</module>
<module>flink-cdc-pipeline-model</module>
</modules>
