(flink-agents) 01/05: [runtime][python] Fix Table environment creation in to_table

[we...@apache.org]

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 <we...@gmail.com>
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 <co...@openai.com>
---
 .../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())
+
+
+@pytest.mark.parametrize("provide_table_env", [False, True])
+@pytest.mark.parametrize("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
+    )