fix(flink): quote SQL identifiers indiscriminately

ibis/backends/base/sql/registry/identifiers.py

@@ -1,21 +1,5 @@
 from __future__ import annotations
 
-# Copyright 2014 Cloudera Inc.
-#
-# Licensed 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.
-
-# Base identifiers
-
 base_identifiers = [
     "add",
     "aggregate",

ibis/backends/flink/compiler/core.py

@@ -18,6 +18,9 @@
 
 
 class FlinkTableSetFormatter(TableSetFormatter):
+    def _quote_identifier(self, name):
+        return quote_identifier(name, force=True)
+
     def _format_in_memory_table(self, op):
         names = op.schema.names
         raw_rows = []
@@ -121,7 +124,7 @@
         filter(
             None,
             [
-                f"TABLE {quote_identifier(op.table.name)}",
+                f"TABLE {formatter._quote_identifier(op.table.name)}",
                 f"DESCRIPTOR({formatter._translate(op.time_col)})",
                 formatter._translate(op.window_size),
                 formatter._translate(op.offset) if op.offset else None,
@@ -136,7 +139,7 @@
         filter(
             None,
             [
-                f"TABLE {quote_identifier(op.table.name)}",
+                f"TABLE {formatter._quote_identifier(op.table.name)}",
                 f"DESCRIPTOR({formatter._translate(op.time_col)})",
                 formatter._translate(op.window_slide),
                 formatter._translate(op.window_size),
@@ -154,7 +157,7 @@
         filter(
             None,
             [
-                f"TABLE {quote_identifier(op.table.name)}",
+                f"TABLE {formatter._quote_identifier(op.table.name)}",
                 f"DESCRIPTOR({formatter._translate(op.time_col)})",
                 formatter._translate(op.window_step),
                 formatter._translate(op.window_size),

ibis/backends/tests/test_aggregation.py

@@ -1471,7 +1471,6 @@ def test_agg_name_in_output_column(alltypes):
     assert "max" in df.columns[1].lower()
 
 
-@pytest.mark.notimpl(["flink"], "WIP", raises=Py4JError)
 def test_grouped_case(backend, con):
     table = ibis.memtable({"key": [1, 1, 2, 2], "value": [10, 30, 20, 40]})
 

ibis/backends/tests/test_window.py

@@ -1215,7 +1215,7 @@ def test_first_last(backend):
     raises=com.OperationNotDefinedError,
     reason="not support by the backend",
 )
-@pytest.mark.notyet(["flink"], raises=Py4JJavaError, reason="not supported by Flink")
+@pytest.mark.broken(["flink"], raises=Py4JJavaError, reason="bug in Flink")
 def test_range_expression_bounds(backend):
     t = ibis.memtable(
         {