feat(flink): add from_connection API for backend

ibis/backends/flink/__init__.py

@@ -71,6 +71,17 @@ def do_connect(self, table_env: TableEnvironment) -> None:
         """
         self._table_env = table_env
 
+    @classmethod
+    def from_connection(cls, table_env: TableEnvironment) -> None:
+        """Create a Flink `Backend` from an existing table environment.
+
+        Parameters
+        ----------
+        table_env
+            A table environment.
+        """
+        return ibis.flink.connect(table_env)
+
     def disconnect(self) -> None:
         pass
 

ibis/backends/tests/test_client.py

@@ -1616,6 +1616,10 @@ def test_insert_using_col_name_not_position(con, first_row, second_row, monkeypa
     con.drop_table(table_name)
 
 
+CON_ATTR = {"flink": "_table_env"}
+DEFAULT_CON_ATTR = "con"
+
+
 def test_from_connection(con):
-    new_con = type(con).from_connection(con.con)
+    new_con = type(con).from_connection(getattr(con, CON_ATTR.get(con.name, "con")))
     assert {"astronauts", "batting", "diamonds"} <= set(new_con.list_tables())