Copilot commented on code in PR #4501:
URL: https://github.com/apache/flink-cdc/pull/4501#discussion_r3800369546
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformExpressionCompiler.java:
##########
@@ -64,7 +64,7 @@ public static ExpressionEvaluator compileExpression(
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()));
Review Comment:
The shared cache key does not include UDF parameter types. With bindings now
based only on the UDF name, two transforms with the same expression/schema and
UDF name but different classpaths produce the same `TransformExpressionKey`;
the second transform can reuse an evaluator whose `__udf_*` parameter is
compiled for the first class and fail with an argument-type mismatch. Include
the UDF binding names and classes in the cache identity, or do not share cached
evaluators for UDF expressions.
##########
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."
+ )
Review Comment:
This returns the annotation from the first top-level `eval`, while executing
the source leaves the last definition bound. If duplicate definitions have
different return annotations, planning uses the wrong type for the function
that actually runs. Reject duplicate top-level `eval` definitions (matching the
documented one-function contract) or otherwise resolve the effective definition.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java:
##########
@@ -1082,12 +1082,12 @@ private static Java.Rvalue generateTypeConvertMethod(
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());
Review Comment:
The YAML/API UDF name is not constrained to a Java identifier, but it is now
embedded verbatim in generated Janino code. A valid quoted SQL function name
such as `my-fn` is accepted by the parser and registered by Calcite, then
generates `__udf_my-fn.eval` and fails compilation. Use an opaque/sanitized
binding identifier shared with `TransformExpressionCompiler`, rather than the
user-facing name.
##########
flink-cdc-python/src/main/java/org/apache/flink/cdc/python/PythonUdf.java:
##########
@@ -0,0 +1,226 @@
+/*
+ * 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);
+ configurePythonFiles(pemjaConfigBuilder, config);
+
+ this.interpreter = new PythonInterpreter(pemjaConfigBuilder.build());
+ this.interpreter.exec(source);
Review Comment:
If interpreter construction or top-level source execution fails (for
example, an import is missing), `open()` exits without closing the interpreter
or deleting archives extracted by `configurePythonFiles`. Task retries can
therefore leak native interpreter resources and temporary disk space. Clean up
both resources on every failure path while preserving the original exception.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
