This is an automated email from the ASF dual-hosted git repository.
RocMarshal 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 13e17094a55 [FLINK-39014][table] Fix the conversion to relational
algebra issue in Batch Mode (#28500)
13e17094a55 is described below
commit 13e17094a556d933520ac1febb464c993c1a81ef
Author: Archit Goyal <[email protected]>
AuthorDate: Fri Aug 7 18:12:52 2026 -0700
[FLINK-39014][table] Fix the conversion to relational algebra issue in
Batch Mode (#28500)
Co-authored-by: Yuepeng Pan <[email protected]>
---
flink-python/pyflink/table/environment_settings.py | 15 ++++++
.../api/bridge/java/StreamTableEnvironment.java | 11 +++-
.../flink/table/api/EnvironmentSettings.java | 16 ++++++
.../api/bridge/scala/StreamTableEnvironment.scala | 12 ++++-
.../StreamTableEnvironmentRuntimeModeTest.java | 60 ++++++++++++++++++++++
5 files changed, 112 insertions(+), 2 deletions(-)
diff --git a/flink-python/pyflink/table/environment_settings.py
b/flink-python/pyflink/table/environment_settings.py
index e5883b2b525..621944e2e91 100644
--- a/flink-python/pyflink/table/environment_settings.py
+++ b/flink-python/pyflink/table/environment_settings.py
@@ -21,6 +21,7 @@ from pyflink.util.api_stability_decorators import
PublicEvolving
from pyflink.util.java_utils import create_url_class_loader
from pyflink.common import Configuration
+from pyflink.datastream.execution_mode import RuntimeExecutionMode
__all__ = ['EnvironmentSettings']
@@ -82,6 +83,20 @@ class EnvironmentSettings(object):
self._j_builder = self._j_builder.inStreamingMode()
return self
+ def in_runtime_execution_mode(
+ self, mode: RuntimeExecutionMode) ->
'EnvironmentSettings.Builder':
+ """
+ Sets the :class:`~pyflink.datastream.RuntimeExecutionMode` that
the components
+ should work in. Only an explicit
+ :class:`~pyflink.datastream.RuntimeExecutionMode`.STREAMING or
+ :class:`~pyflink.datastream.RuntimeExecutionMode`.BATCH mode is
supported in the
+ Table API.
+
+ :return: This object.
+ """
+ self._j_builder =
self._j_builder.inRuntimeExecutionMode(mode._to_j_execution_mode())
+ return self
+
def with_built_in_catalog_name(self, built_in_catalog_name: str) \
-> 'EnvironmentSettings.Builder':
"""
diff --git
a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/StreamTableEnvironment.java
b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/StreamTableEnvironment.java
index e580efc6357..c852ce27571 100644
---
a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/StreamTableEnvironment.java
+++
b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/StreamTableEnvironment.java
@@ -19,11 +19,13 @@
package org.apache.flink.table.api.bridge.java;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.common.typeutils.CompositeType;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.ExecutionOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
@@ -88,7 +90,14 @@ public interface StreamTableEnvironment extends
TableEnvironment {
* TableEnvironment}.
*/
static StreamTableEnvironment create(StreamExecutionEnvironment
executionEnvironment) {
- return create(executionEnvironment,
EnvironmentSettings.newInstance().build());
+ final RuntimeExecutionMode runtimeMode =
+
executionEnvironment.getConfiguration().get(ExecutionOptions.RUNTIME_MODE)
+ == RuntimeExecutionMode.BATCH
+ ? RuntimeExecutionMode.BATCH
+ : RuntimeExecutionMode.STREAMING;
+ return create(
+ executionEnvironment,
+
EnvironmentSettings.newInstance().inRuntimeExecutionMode(runtimeMode).build());
}
/**
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java
index 92a375bf0bf..e052c02bdba 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java
@@ -20,12 +20,14 @@ package org.apache.flink.table.api;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.config.TableConfigOptions;
import org.apache.flink.table.catalog.CatalogStore;
import org.apache.flink.table.expressions.SqlFactory;
import org.apache.flink.table.functions.UserDefinedFunction;
import org.apache.flink.table.secret.SecretStore;
+import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable;
@@ -187,6 +189,20 @@ public class EnvironmentSettings {
return this;
}
+ /** Sets the {@link RuntimeExecutionMode} that the components should
work in. */
+ public Builder inRuntimeExecutionMode(RuntimeExecutionMode mode) {
+ Preconditions.checkArgument(
+ mode == STREAMING || mode == BATCH,
+ "Unsupported value '%s' for '%s'. "
+ + "Only an explicit %s or %s mode is supported in
Table API.",
+ mode,
+ RUNTIME_MODE.key(),
+ STREAMING,
+ BATCH);
+ configuration.set(RUNTIME_MODE, mode);
+ return this;
+ }
+
/**
* Specifies the name of the initial catalog to be created when
instantiating a {@link
* TableEnvironment}.
diff --git
a/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/bridge/scala/StreamTableEnvironment.scala
b/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/bridge/scala/StreamTableEnvironment.scala
index a0d3df3338a..f71a1b5b837 100644
---
a/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/bridge/scala/StreamTableEnvironment.scala
+++
b/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/bridge/scala/StreamTableEnvironment.scala
@@ -18,8 +18,10 @@
package org.apache.flink.table.api.bridge.scala
import org.apache.flink.annotation.PublicEvolving
+import org.apache.flink.api.common.RuntimeExecutionMode
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.common.typeutils.CompositeType
+import org.apache.flink.configuration.ExecutionOptions
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.table.api.{TableEnvironment, _}
@@ -833,7 +835,15 @@ object StreamTableEnvironment {
* The Scala [[StreamExecutionEnvironment]] of the [[TableEnvironment]].
*/
def create(executionEnvironment: StreamExecutionEnvironment):
StreamTableEnvironment = {
- create(executionEnvironment, EnvironmentSettings.newInstance().build)
+ val runtimeMode =
+ if (
+
executionEnvironment.getConfiguration.get(ExecutionOptions.RUNTIME_MODE) ==
+ RuntimeExecutionMode.BATCH
+ ) RuntimeExecutionMode.BATCH
+ else RuntimeExecutionMode.STREAMING
+ create(
+ executionEnvironment,
+
EnvironmentSettings.newInstance().inRuntimeExecutionMode(runtimeMode).build)
}
/**
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/table/StreamTableEnvironmentRuntimeModeTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/table/StreamTableEnvironmentRuntimeModeTest.java
new file mode 100644
index 00000000000..99dbcba266f
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/table/StreamTableEnvironmentRuntimeModeTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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.runtime.stream.table;
+
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ExecutionOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests that {@link
StreamTableEnvironment#create(StreamExecutionEnvironment)} inherits the runtime
+ * execution mode from the given {@link StreamExecutionEnvironment} instead of
always defaulting to
+ * streaming (FLINK-39014).
+ */
+class StreamTableEnvironmentRuntimeModeTest {
+
+ @Test
+ void testCreateInheritsBatchRuntimeMode() {
+ final Configuration configuration = new Configuration();
+ configuration.set(ExecutionOptions.RUNTIME_MODE,
RuntimeExecutionMode.BATCH);
+ final StreamExecutionEnvironment env =
+
StreamExecutionEnvironment.getExecutionEnvironment(configuration);
+
+ final StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ assertThat(tEnv.getConfig().get(ExecutionOptions.RUNTIME_MODE))
+ .isEqualTo(RuntimeExecutionMode.BATCH);
+ }
+
+ @Test
+ void testCreateDefaultsToStreamingRuntimeMode() {
+ final StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+
+ final StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ assertThat(tEnv.getConfig().get(ExecutionOptions.RUNTIME_MODE))
+ .isEqualTo(RuntimeExecutionMode.STREAMING);
+ }
+}
