[FLINK-38882][table][python] Introduce rules for async python scalar function

flink-python/pyflink/fn_execution/flink_fn_execution_pb2.pyi

@@ -56,18 +56,28 @@ class UserDefinedFunction(_message.Message):
     def __init__(self, payload: _Optional[bytes] = ..., inputs: _Optional[_Iterable[_Union[Input, _Mapping]]] = ..., window_index: _Optional[int] = ..., takes_row_as_input: bool = ..., is_pandas_udf: bool = ...) -> None: ...
 
 class UserDefinedFunctions(_message.Message):
-    __slots__ = ("udfs", "metric_enabled", "windows", "profile_enabled", "job_parameters")
+    __slots__ = ("udfs", "metric_enabled", "windows", "profile_enabled", "job_parameters", "async_buffer_capacity", "async_timeout_ms", "async_retry_enabled", "async_max_attempts", "async_retry_delay_ms")
     UDFS_FIELD_NUMBER: _ClassVar[int]
     METRIC_ENABLED_FIELD_NUMBER: _ClassVar[int]
     WINDOWS_FIELD_NUMBER: _ClassVar[int]
     PROFILE_ENABLED_FIELD_NUMBER: _ClassVar[int]
     JOB_PARAMETERS_FIELD_NUMBER: _ClassVar[int]
+    ASYNC_BUFFER_CAPACITY_FIELD_NUMBER: _ClassVar[int]
+    ASYNC_TIMEOUT_MS_FIELD_NUMBER: _ClassVar[int]
+    ASYNC_RETRY_ENABLED_FIELD_NUMBER: _ClassVar[int]
+    ASYNC_MAX_ATTEMPTS_FIELD_NUMBER: _ClassVar[int]
+    ASYNC_RETRY_DELAY_MS_FIELD_NUMBER: _ClassVar[int]
     udfs: _containers.RepeatedCompositeFieldContainer[UserDefinedFunction]
     metric_enabled: bool
     windows: _containers.RepeatedCompositeFieldContainer[OverWindow]
     profile_enabled: bool
     job_parameters: _containers.RepeatedCompositeFieldContainer[JobParameter]
-    def __init__(self, udfs: _Optional[_Iterable[_Union[UserDefinedFunction, _Mapping]]] = ..., metric_enabled: bool = ..., windows: _Optional[_Iterable[_Union[OverWindow, _Mapping]]] = ..., profile_enabled: bool = ..., job_parameters: _Optional[_Iterable[_Union[JobParameter, _Mapping]]] = ...) -> None: ...
+    async_buffer_capacity: int
+    async_timeout_ms: int
+    async_retry_enabled: bool
+    async_max_attempts: int
+    async_retry_delay_ms: int
+    def __init__(self, udfs: _Optional[_Iterable[_Union[UserDefinedFunction, _Mapping]]] = ..., metric_enabled: bool = ..., windows: _Optional[_Iterable[_Union[OverWindow, _Mapping]]] = ..., profile_enabled: bool = ..., job_parameters: _Optional[_Iterable[_Union[JobParameter, _Mapping]]] = ..., async_buffer_capacity: _Optional[int] = ..., async_timeout_ms: _Optional[int] = ..., async_retry_enabled: bool = ..., async_max_attempts: _Optional[int] = ..., async_retry_delay_ms: _Optional[int] = ...) -> None: ...
 
 class OverWindow(_message.Message):
     __slots__ = ("window_type", "lower_boundary", "upper_boundary")

flink-python/pyflink/proto/flink-fn-execution.proto

@@ -71,6 +71,13 @@ message UserDefinedFunctions {
   repeated OverWindow windows = 3;
   bool profile_enabled = 4;
   repeated JobParameter job_parameters = 5;
+  // Async execution configuration for async scalar functions
+  int32 async_buffer_capacity = 6;  // Max number of concurrent async operations
+  int64 async_timeout_ms = 7;  // Timeout in milliseconds for async operations
+  // Async retry strategy configuration
+  bool async_retry_enabled = 8;  // Whether retry is enabled
+  int32 async_max_attempts = 9;  // Maximum number of retry attempts
+  int64 async_retry_delay_ms = 10;  // Delay between retries in milliseconds
 }
 
 // Used to describe the info of over window in pandas batch over window aggregation

flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/scalar/async/PythonAsyncScalarFunctionOperator.java

@@ -0,0 +1,103 @@
+/*
+ * 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.table.runtime.operators.python.scalar.async;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.fnexecution.v1.FlinkFnApi;
+import org.apache.flink.table.functions.AsyncScalarFunction;
+import org.apache.flink.table.functions.python.PythonFunctionInfo;
+import org.apache.flink.table.runtime.generated.GeneratedProjection;
+import org.apache.flink.table.runtime.operators.python.scalar.PythonScalarFunctionOperator;
+import org.apache.flink.table.types.logical.RowType;
+
+/** The Python {@link AsyncScalarFunction} operator for Table API. */
+@Internal
+public class PythonAsyncScalarFunctionOperator extends PythonScalarFunctionOperator {
+
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * The maximum number of async operations that can be in-flight at the same time. This controls
+     * the buffer capacity for async execution.
+     */
+    private final int asyncBufferCapacity;
+
+    /** The timeout in milliseconds for async operations. */
+    private final long asyncTimeout;
+
+    /** Whether retry is enabled for async operations. */
+    private final boolean asyncRetryEnabled;
+
+    /** Maximum number of retry attempts. */
+    private final int asyncMaxAttempts;
+
+    /** Delay between retries in milliseconds. */
+    private final long asyncRetryDelayMs;
+
+    public PythonAsyncScalarFunctionOperator(
+            Configuration config,
+            PythonFunctionInfo[] scalarFunctions,
+            RowType inputType,
+            RowType udfInputType,
+            RowType udfOutputType,
+            GeneratedProjection udfInputGeneratedProjection,
+            GeneratedProjection forwardedFieldGeneratedProjection,
+            int asyncBufferCapacity,
+            long asyncTimeout,
+            boolean asyncRetryEnabled,
+            int asyncMaxAttempts,
+            long asyncRetryDelayMs) {
+        super(
+                config,
+                scalarFunctions,
+                inputType,
+                udfInputType,
+                udfOutputType,
+                udfInputGeneratedProjection,
+                forwardedFieldGeneratedProjection);
+        this.asyncBufferCapacity = asyncBufferCapacity;
+        this.asyncTimeout = asyncTimeout;
+        this.asyncRetryEnabled = asyncRetryEnabled;
+        this.asyncMaxAttempts = asyncMaxAttempts;
+        this.asyncRetryDelayMs = asyncRetryDelayMs;
+    }
+
+    @Override
+    public String getFunctionUrn() {
+        // Use a dedicated URN for async scalar functions
+        return "flink:transform:async_scalar_function:v1";
+    }
+
+    @Override
+    public FlinkFnApi.UserDefinedFunctions createUserDefinedFunctionsProto() {
+        // Create the base proto with scalar functions
+        FlinkFnApi.UserDefinedFunctions.Builder builder =
+                super.createUserDefinedFunctionsProto().toBuilder();
+
+        // Add async-specific configurations
+        builder.setAsyncBufferCapacity(asyncBufferCapacity);
+        builder.setAsyncTimeoutMs(asyncTimeout);
+        builder.setAsyncRetryEnabled(asyncRetryEnabled);
+        builder.setAsyncMaxAttempts(asyncMaxAttempts);
+        builder.setAsyncRetryDelayMs(asyncRetryDelayMs);
+
+        return builder.build();
+    }
+}

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecPythonAsyncCalc.java

@@ -0,0 +1,68 @@
+/*
+ * 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.table.planner.plan.nodes.exec.batch;
+
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeContext;
+import org.apache.flink.table.planner.plan.nodes.exec.InputProperty;
+import org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecPythonAsyncCalc;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import org.apache.calcite.rex.RexNode;
+
+import java.util.Collections;
+import java.util.List;
+
+/** Batch {@link ExecNode} for Python Async ScalarFunctions. */
+public class BatchExecPythonAsyncCalc extends CommonExecPythonAsyncCalc
+        implements BatchExecNode<RowData> {
+
+    public BatchExecPythonAsyncCalc(
+            ReadableConfig tableConfig,
+            List<RexNode> projection,
+            InputProperty inputProperty,
+            RowType outputType,
+            String description) {
+        this(
+                ExecNodeContext.newNodeId(),
+                ExecNodeContext.newContext(BatchExecPythonAsyncCalc.class),
+                ExecNodeContext.newPersistedConfig(BatchExecPythonAsyncCalc.class, tableConfig),
+                projection,
+                Collections.singletonList(inputProperty),
+                outputType,
+                description);
+    }
+
+    @JsonCreator
+    public BatchExecPythonAsyncCalc(
+            @JsonProperty(FIELD_NAME_ID) int id,
+            @JsonProperty(FIELD_NAME_TYPE) ExecNodeContext context,
+            @JsonProperty(FIELD_NAME_CONFIGURATION) ReadableConfig persistedConfig,
+            @JsonProperty(FIELD_NAME_PROJECTION) List<RexNode> projection,
+            @JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty> inputProperties,
+            @JsonProperty(FIELD_NAME_OUTPUT_TYPE) RowType outputType,
+            @JsonProperty(FIELD_NAME_DESCRIPTION) String description) {
+        super(id, context, persistedConfig, projection, inputProperties, outputType, description);
+    }
+}