You are viewing a plain text version of this content. The canonical link for it is here.

Posted to issues@flink.apache.org by "dianfu (via GitHub)" <gi...@apache.org> on 2026/01/09 08:40:05 UTC

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

dianfu opened a new pull request, #27395:
URL: https://github.com/apache/flink/pull/27395

   ## What is the purpose of the change
   
   *This pull request introduces the necessary rules and nodes to support async python scalar function*
   
   
   ## Brief change log
   
   *(for example:)*
     - *The TaskInfo is stored in the blob store on job creation time as a persistent artifact*
     - *Deployments RPC transmits only the blob storage reference*
     - *TaskManagers retrieve the TaskInfo from the blob cache*
   
   
   ## Verifying this change
   
   This change added tests and can be verified as follows:
   
   *(example:)*
     - *Added integration tests for end-to-end deployment with large payloads (100MB)*
     - *Extended integration test for recovery after master (JobManager) failure*
     - *Added test that validates that TaskInfo is transferred only once across recoveries*
     - *Manually verified the change by running a 4 node cluster with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.*
   
   ## Does this pull request potentially affect one of the following parts:
   
     - Dependencies (does it add or upgrade a dependency): (yes / no)
     - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / no)
     - The serializers: (yes / no / don't know)
     - The runtime per-record code paths (performance sensitive): (yes / no / don't know)
     - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
     - The S3 file system connector: (yes / no / don't know)
   
   ## Documentation
   
     - Does this pull request introduce a new feature? (yes / no)
     - If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
   


-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-38882][table][python] Introduce rules for async python scalar function [flink]

Posted by "bgeng777 (via GitHub)" <gi...@apache.org>.

bgeng777 commented on code in PR #27395:
URL: https://github.com/apache/flink/pull/27395#discussion_r2703070589


##########
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

Review Comment:
   I notice that for python batch exec ndoes, we do not add metadata annotations like the streaming counterparts. 
   I believe it is not strictly required but I wonder which one is preferred.  It looks like https://issues.apache.org/jira/browse/FLINK-35797 has declared some extra requirement.
   
   ```
   @ExecNodeMetadata(
       name = "batch-exec-python-async-calc",
       version = 1,
       producedTransformations = CommonExecPythonAsyncCalc.PYTHON_ASYNC_CALC_TRANSFORMATION,
       minPlanVersion = FlinkVersion.v2_0,
       minStateVersion = FlinkVersion.v2_0)
   ```
   



##########
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_max_concurrent_operations = 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_retry_max_attempts = 9;  // Maximum number of retry attempts
+  int64 async_retry_delay_ms = 10;  // Delay between retries in milliseconds

Review Comment:
   It looks like we need 5 extra fields for async execution. Is is possible to introduce a `AsyncOptions` messaage to hold these 5 fields? 



##########
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 asyncMaxConcurrentOperations;
+
+    /** 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 asyncRetryMaxAttempts;
+
+    /** 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 asyncRetryMaxAttempts,
+            long asyncRetryDelayMs) {
+        super(
+                config,
+                scalarFunctions,
+                inputType,
+                udfInputType,
+                udfOutputType,
+                udfInputGeneratedProjection,
+                forwardedFieldGeneratedProjection);
+        this.asyncMaxConcurrentOperations = asyncBufferCapacity;
+        this.asyncTimeout = asyncTimeout;
+        this.asyncRetryEnabled = asyncRetryEnabled;
+        this.asyncRetryMaxAttempts = asyncRetryMaxAttempts;
+        this.asyncRetryDelayMs = asyncRetryDelayMs;
+    }
+
+    @Override
+    public String getFunctionUrn() {
+        // Use a dedicated URN for async scalar functions
+        return "flink:transform:async_scalar_function:v1";

Review Comment:
   Introduce a `private static final String ASYNC_SCALAR_FUNCTION_URN = "flink:transform: async_scalar_function:v1";`?



-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-38882][table][python] Introduce rules for async python scalar function [flink]

Posted by "davidradl (via GitHub)" <gi...@apache.org>.

davidradl commented on code in PR #27395:
URL: https://github.com/apache/flink/pull/27395#discussion_r2681452432


##########
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

Review Comment:
   I am curious how we have come up with these numbers. `async_max_attempts = 9` seems a strange amount?



-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-38882][table][python] Introduce rules for async python scalar function [flink]

Posted by "flinkbot (via GitHub)" <gi...@apache.org>.

flinkbot commented on PR #27395:
URL: https://github.com/apache/flink/pull/27395#issuecomment-3727845592

   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "67d917ee9bdcd0bcbc7bc8209c68dd20e725aa1d",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "67d917ee9bdcd0bcbc7bc8209c68dd20e725aa1d",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 67d917ee9bdcd0bcbc7bc8209c68dd20e725aa1d UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org