You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sh...@apache.org on 2025/06/12 01:19:21 UTC
(flink) branch master updated: [hotfix][table] Fix StreamExecMLPredictTableFunction doesn't respect object reuse option and add consumed options in the metadata (#26669)
This is an automated email from the ASF dual-hosted git repository.
shengkai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 228de51fb23 [hotfix][table] Fix StreamExecMLPredictTableFunction doesn't respect object reuse option and add consumed options in the metadata (#26669)
228de51fb23 is described below
commit 228de51fb2396ef35d1323cd506ac7f76b0827f3
Author: Shengkai <33...@users.noreply.github.com>
AuthorDate: Thu Jun 12 09:19:14 2025 +0800
[hotfix][table] Fix StreamExecMLPredictTableFunction doesn't respect object reuse option and add consumed options in the metadata (#26669)
---
.../stream/StreamExecMLPredictTableFunction.java | 8 +++++-
.../nodes/exec/stream/MLPredictRestoreTest.java | 30 ++++++++++++++++++++++
2 files changed, 37 insertions(+), 1 deletion(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecMLPredictTableFunction.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecMLPredictTableFunction.java
index 62f5a684f25..fde1783f6ca 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecMLPredictTableFunction.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecMLPredictTableFunction.java
@@ -22,6 +22,7 @@ import org.apache.flink.FlinkVersion;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.operators.ProcessOperator;
@@ -80,6 +81,11 @@ import java.util.Optional;
@ExecNodeMetadata(
name = "stream-exec-ml-predict-table-function",
version = 1,
+ consumedOptions = {
+ "table.exec.async-ml-predict.buffer-capacity",
+ "table.exec.async-ml-predict.timeout",
+ "table.exec.async-ml-predict.output-mode"
+ },
producedTransformations = StreamExecMLPredictTableFunction.ML_PREDICT_TRANSFORMATION,
minPlanVersion = FlinkVersion.v2_1,
minStateVersion = FlinkVersion.v2_1)
@@ -201,7 +207,7 @@ public class StreamExecMLPredictTableFunction extends ExecNodeBase<RowData>
mlPredictSpec.getFeatures(),
predictFunction,
"MLPredict",
- true);
+ config.get(PipelineOptions.OBJECT_REUSE));
GeneratedCollector<ListenableCollector<RowData>> generatedCollector =
LookupJoinCodeGenerator.generateCollector(
new CodeGeneratorContext(config, classLoader),
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MLPredictRestoreTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MLPredictRestoreTest.java
index 843b1e99da6..0a94096c833 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MLPredictRestoreTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MLPredictRestoreTest.java
@@ -18,14 +18,24 @@
package org.apache.flink.table.planner.plan.nodes.exec.stream;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
import org.apache.flink.table.planner.plan.nodes.exec.testutils.RestoreTestBase;
+import org.apache.flink.table.planner.plan.utils.ExecNodeMetadataUtil;
import org.apache.flink.table.test.program.TableTestProgram;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.HashSet;
import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
import static org.apache.flink.table.planner.plan.nodes.exec.stream.MLPredictTestPrograms.ASYNC_UNORDERED_ML_PREDICT;
import static org.apache.flink.table.planner.plan.nodes.exec.stream.MLPredictTestPrograms.SYNC_ML_PREDICT;
import static org.apache.flink.table.planner.plan.nodes.exec.stream.MLPredictTestPrograms.SYNC_ML_PREDICT_WITH_RUNTIME_CONFIG;
+import static org.assertj.core.api.Assertions.assertThat;
/** Restore tests for {@link StreamExecMLPredictTableFunction}. */
public class MLPredictRestoreTest extends RestoreTestBase {
@@ -34,6 +44,26 @@ public class MLPredictRestoreTest extends RestoreTestBase {
super(StreamExecMLPredictTableFunction.class);
}
+ @Test
+ public void testExecNodeMetadataContainsRequiredOptions() {
+ assertThat(
+ new HashSet<>(
+ Arrays.asList(
+ Objects.requireNonNull(
+ ExecNodeMetadataUtil.consumedOptions(
+ StreamExecMLPredictTableFunction.class)))))
+ .isEqualTo(
+ Arrays.asList(
+ ExecutionConfigOptions
+ .TABLE_EXEC_ASYNC_ML_PREDICT_BUFFER_CAPACITY,
+ ExecutionConfigOptions.TABLE_EXEC_ASYNC_ML_PREDICT_TIMEOUT,
+ ExecutionConfigOptions
+ .TABLE_EXEC_ASYNC_ML_PREDICT_OUTPUT_MODE)
+ .stream()
+ .map(ConfigOption::key)
+ .collect(Collectors.toSet()));
+ }
+
@Override
public List<TableTestProgram> programs() {
return List.of(