diff --git a/ibis/backends/flink/__init__.py b/ibis/backends/flink/__init__.py
index e45615c4fa30..d4964fa8e3e8 100644
--- a/ibis/backends/flink/__init__.py
+++ b/ibis/backends/flink/__init__.py
@@ -906,18 +906,22 @@ def insert(
             )
             return self.raw_sql(statement.compile())
 
+        identifier = sg.table(
+            table_name, db=database, catalog=catalog, quoted=self.compiler.quoted
+        ).sql(self.dialect)
+
         if isinstance(obj, pa.Table):
             obj = obj.to_pandas()
         if isinstance(obj, dict):
             obj = pd.DataFrame.from_dict(obj)
         if isinstance(obj, pd.DataFrame):
             table = self._table_env.from_pandas(obj)
-            return table.execute_insert(table_name, overwrite=overwrite)
+            return table.execute_insert(identifier, overwrite=overwrite)
 
         if isinstance(obj, list):
             # pyflink infers datatypes, which may sometimes result in incompatible types
             table = self._table_env.from_elements(obj)
-            return table.execute_insert(table_name, overwrite=overwrite)
+            return table.execute_insert(identifier, overwrite=overwrite)
 
         raise ValueError(
             "No operation is being performed. Either the obj parameter "