This is an automated email from the ASF dual-hosted git repository. wenjin272 pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-agents.git
commit 1126dbdb42e56c47d72adb8cfb66c600032826e0 Author: WenjinXie <[email protected]> AuthorDate: Tue Oct 20 11:11:32 2026 +0800 [runtime][python] Fix Table environment creation in to_table Create the implicit TableEnvironment from the input DataStream environment. Cover explicit and implicit environments with both output schemas and types. Generated-by: Codex CLI 0.153.4 (GPT-6) Co-authored-by: Codex <[email protected]> --- .../runtime/remote_execution_environment.py | 4 +-- .../tests/test_remote_execution_environment.py | 35 +++++++++++++++++++++- 2 files changed, 36 insertions(+), 3 deletions(-) diff --git a/python/flink_agents/runtime/remote_execution_environment.py b/python/flink_agents/runtime/remote_execution_environment.py index e1e21df7d..8f9e35aab 100644 --- a/python/flink_agents/runtime/remote_execution_environment.py +++ b/python/flink_agents/runtime/remote_execution_environment.py @@ -99,10 +99,10 @@ class RemoteAgentBuilder(AgentBuilder): @property def t_env(self) -> StreamTableEnvironment: - """Get or crate table environment.""" + """Get or create the table environment for the input stream.""" if self.__t_env is None: self.__t_env = StreamTableEnvironment.create( - stream_execution_environment=self.__env + stream_execution_environment=self.__input.get_execution_environment() ) return self.__t_env diff --git a/python/flink_agents/runtime/tests/test_remote_execution_environment.py b/python/flink_agents/runtime/tests/test_remote_execution_environment.py index b3b549adf..5256fffa7 100644 --- a/python/flink_agents/runtime/tests/test_remote_execution_environment.py +++ b/python/flink_agents/runtime/tests/test_remote_execution_environment.py @@ -28,7 +28,8 @@ import pytest import yaml from pyflink.common import Row from pyflink.common.typeinfo import ExternalTypeInfo, Types -from pyflink.table import DataTypes, Schema +from pyflink.datastream import StreamExecutionEnvironment +from pyflink.table import DataTypes, Schema, StreamTableEnvironment from flink_agents.api.agents.agent import Agent from flink_agents.api.events.event import Event @@ -606,3 +607,35 @@ def test_to_table_rejects_schema_in_output_type_slot(): pytest.raises(TypeError, match="pass a Table Schema as schema="), ): builder.to_table(output_type=_output_schema()) + + [email protected]("provide_table_env", [False, True]) [email protected]("use_schema", [False, True]) +def test_datastream_to_table_uses_input_environment(provide_table_env, use_schema): + """Build real Table views with and without an explicit TableEnvironment. + + Stub only agent execution: stream ownership, keying, row conversion and + Table planning use PyFlink. This catches a missing environment on the lazy path. + """ + env = StreamExecutionEnvironment.get_execution_environment() + t_env = StreamTableEnvironment.create(env) if provide_table_env else None + remote_env = RemoteExecutionEnvironment(env=env, t_env=t_env) + payload = cloudpickle.dumps({"id": 1, "label": "a", "score": 1.5}) + raw = env.from_collection([payload]) + builder = remote_env.from_datastream(raw, key_selector=lambda _: "key") + with patch.object(RemoteAgentBuilder, "_raw_output_stream", return_value=raw): + if use_schema: + table = builder.to_table(schema=_output_schema()) + else: + table = builder.to_table(output_type=ModelOutput) + cached_env = builder.t_env + builder.to_table(output_type=ModelOutput) + assert table.get_schema().get_field_names() == ["id", "label", "score"] + assert builder.t_env is cached_env + if provide_table_env: + assert builder.t_env is t_env + # The Table view belongs to the original job, rather than a new environment. + output = builder.t_env.to_data_stream(table) + assert output.get_execution_environment()._j_stream_execution_environment.equals( + env._j_stream_execution_environment + )
