Simplify to unconditional backtick quoting, gate behind flag

Following review feedback: the conditional check based on whether the
name matches a bare-identifier pattern is unnecessary. The Spark/
Databricks SQL grammar accepts :`name` for every valid identifier
(verified empirically against a live SQL warehouse), so wrapping
unconditionally keeps the compiler simpler and removes a class of
edge-case bugs that the condition could miss.

Add a ``quote_bind_params`` flag on DatabricksDialect (default True)
that can be turned off via the URL query parameter
``?quote_bind_params=false`` as an escape hatch if the quoting ever
introduces an unexpected regression. When disabled, we fall through
entirely to stock SQLAlchemy bind-name rendering.

Co-authored-by: Isaac

Author: msrathore-db

src/databricks/sqlalchemy/_ddl.py

@@ -84,50 +84,52 @@ def get_column_specification(self, column, **kwargs):
 
 
 class DatabricksStatementCompiler(compiler.SQLCompiler):
-    # Names that a bare Databricks named-parameter marker (`:name`) accepts:
-    # a letter or underscore followed by letters, digits, or underscores.
-    # Anything outside that set — hyphens, spaces, dots, brackets, a leading
-    # digit, etc. — must be wrapped in backticks (`:`name``), which the
-    # Spark/Databricks SQL grammar accepts as a quoted parameter identifier.
-    _bindname_is_bare_identifier = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
-
     def bindparam_string(self, name, **kw):
-        """Render a bind parameter marker.
-
-        Databricks named parameter markers only accept bare identifiers
-        ([A-Za-z_][A-Za-z0-9_]*) out of the box. DataFrame-origin column
-        names frequently contain hyphens (e.g. ``col-with-hyphen``), which
-        SQLAlchemy would otherwise pass through verbatim and produce an
-        invalid marker ``:col-with-hyphen`` — the parser splits on ``-``
-        and reports UNBOUND_SQL_PARAMETER.
-
-        The Spark SQL grammar accepts a quoted form ``:`col-with-hyphen```,
-        mirroring Oracle's ``:"name"`` pattern. The backticks are *quoting*
-        only: the parameter's logical name is still the text between them,
-        so the params dict sent to the driver must keep the original
-        unquoted key. We therefore emit the backticked marker directly
-        without populating ``escaped_bind_names`` — leaving the key
-        translation in ``construct_params`` a no-op.
-
-        For bare identifiers (the common case), we fall through to the
-        default implementation so INSERT/SELECT output stays unchanged.
+        """Render a bind parameter marker wrapped in backticks.
+
+        Databricks named parameter markers accept two identifier forms
+        (per ``SqlBaseParser.g4``): a bare ``IDENTIFIER``
+        (``[A-Za-z_][A-Za-z0-9_]*``) or a ``quotedIdentifier`` wrapped in
+        backticks. DataFrame-origin column names frequently contain
+        hyphens (e.g. ``col-with-hyphen``), which SQLAlchemy would
+        otherwise pass through verbatim and produce an invalid marker
+        ``:col-with-hyphen`` — the parser splits on ``-`` and reports
+        UNBOUND_SQL_PARAMETER.
+
+        Backticks are valid for *every* identifier (plain names included),
+        verified empirically against a Databricks SQL warehouse, so we
+        wrap unconditionally. This mirrors Oracle's ``:"name"`` approach
+        to the same grammar constraint and eliminates the collision risk
+        that any single-character escape map would carry (e.g. ``col-name``
+        vs ``col_name`` both mapping to ``:col_name``).
+
+        The backticks are SQL-side *quoting* only: the parameter's
+        logical name is still the text between them, so the params dict
+        sent to the driver keeps the original unquoted key. We therefore
+        leave ``escaped_bind_names`` untouched — ``construct_params``
+        passes keys through unchanged.
+
+        Gated by ``DatabricksDialect.quote_bind_params``. Set
+        ``?quote_bind_params=false`` on the SQLAlchemy URL to fall back
+        to stock bind-name rendering.
         """
         if (
-            not kw.get("escaped_from")
-            and not kw.get("post_compile", False)
-            and not self._bindname_is_bare_identifier.match(name)
+            kw.get("post_compile", False)
+            or kw.get("escaped_from")
+            or not getattr(self.dialect, "quote_bind_params", True)
         ):
-            accumulate = kw.get("accumulate_bind_names")
-            if accumulate is not None:
-                accumulate.add(name)
-            visited = kw.get("visited_bindparam")
-            if visited is not None:
-                visited.append(name)
-            quoted = f"`{name}`"
-            if self.state is compiler.CompilerState.COMPILING:
-                return self.compilation_bindtemplate % {"name": quoted}
-            return self.bindtemplate % {"name": quoted}
-        return super().bindparam_string(name, **kw)
+            return super().bindparam_string(name, **kw)
+
+        accumulate = kw.get("accumulate_bind_names")
+        if accumulate is not None:
+            accumulate.add(name)
+        visited = kw.get("visited_bindparam")
+        if visited is not None:
+            visited.append(name)
+        quoted = f"`{name}`"
+        if self.state is compiler.CompilerState.COMPILING:
+            return self.compilation_bindtemplate % {"name": quoted}
+        return self.bindtemplate % {"name": quoted}
 
     def limit_clause(self, select, **kw):
         """Identical to the default implementation of SQLCompiler.limit_clause except it writes LIMIT ALL instead of LIMIT -1,

src/databricks/sqlalchemy/base.py

@@ -66,6 +66,15 @@ class DatabricksDialect(default.DefaultDialect):
     supports_sequences: bool = False
     supports_native_boolean: bool = True
 
+    # When True (default), every named bind parameter is rendered wrapped in
+    # backticks (`` :`name` ``) so that column names containing characters
+    # which are illegal in bare Databricks parameter identifiers (hyphens,
+    # spaces, dots, leading digits, etc.) work transparently. Set to False
+    # via the URL query string — ``?quote_bind_params=false`` — to fall back
+    # to stock SQLAlchemy bind-name rendering if this quoting causes an
+    # unexpected regression.
+    quote_bind_params: bool = True
+
     colspecs = {
         sqlalchemy.types.DateTime: dialect_type_impl.TIMESTAMP_NTZ,
         sqlalchemy.types.Time: dialect_type_impl.DatabricksTimeType,
@@ -117,6 +126,10 @@ def create_connect_args(self, url):
         self.schema = kwargs["schema"]
         self.catalog = kwargs["catalog"]
 
+        raw_quote_flag = url.query.get("quote_bind_params")
+        if raw_quote_flag is not None:
+            self.quote_bind_params = raw_quote_flag.lower() not in ("false", "0", "no")
+
         self._force_paramstyle_to_native_mode()
 
         return [], kwargs

tests/test_local/test_ddl.py

@@ -117,18 +117,24 @@ def test_create_table_with_complex_type(self, metadata):
 
 
 class TestBindParamQuoting(DDLTestBase):
-    """Regression tests for column names that contain characters which are not
-    legal inside a bare Databricks named-parameter marker (`:name`). Without
-    the custom ``bindparam_string`` override, a column like
-    ``col-with-hyphen`` produces SQL like ``VALUES (:col-with-hyphen)`` which
-    fails with UNBOUND_SQL_PARAMETER on the server. The fix wraps such names
-    in backticks (``VALUES (:`col-with-hyphen`)``), which the Databricks SQL
-    grammar accepts as a quoted parameter identifier.
+    """Regression tests for bind-parameter quoting.
+
+    Databricks named parameter markers (``:name``) must be bare identifiers
+    (``[A-Za-z_][A-Za-z0-9_]*``) unless wrapped in backticks. Because
+    DataFrame-origin column names frequently contain hyphens (a character
+    that's legal inside a backtick-quoted column identifier but not in a
+    bare bind marker), the dialect wraps every bind name in backticks
+    unconditionally. The backticks are SQL-side quoting only — the params
+    dict sent to the driver keeps the original unquoted key.
+
+    The behavior is gated by ``DatabricksDialect.quote_bind_params`` which
+    defaults to True; set ``?quote_bind_params=false`` in the URL to
+    disable.
     """
 
-    def _compile_insert(self, table, values):
+    def _compile_insert(self, table, values, engine=None):
         stmt = insert(table).values(values)
-        return stmt.compile(bind=self.engine)
+        return stmt.compile(bind=engine or self.engine)
 
     def test_hyphenated_column_renders_backticked_bind_marker(self):
         metadata = MetaData()
@@ -143,18 +149,18 @@ def test_hyphenated_column_renders_backticked_bind_marker(self):
         )
 
         sql = str(compiled)
-        # Hyphenated name is wrapped in backticks at the marker site
+        # Both names are backticked at the marker site
         assert ":`col-with-hyphen`" in sql
-        # Plain name is untouched
-        assert ":normal_col" in sql
+        assert ":`normal_col`" in sql
         # The params dict sent to the driver keeps the ORIGINAL unquoted key
         # — this matches what the Databricks server expects (verified
-        # empirically: a backticked marker `:`name`` binds against a plain
-        # `name` key in the params dict).
+        # empirically: a backticked marker ``:`name``` binds against a plain
+        # ``name`` key in the params dict).
         params = compiled.construct_params()
         assert params["col-with-hyphen"] == "x"
         assert params["normal_col"] == "y"
         assert "`col-with-hyphen`" not in params
+        assert "`normal_col`" not in params
 
     def test_hyphen_and_underscore_columns_do_not_collide(self):
         """A table containing both ``col-name`` and ``col_name`` must produce
@@ -174,14 +180,17 @@ def test_hyphen_and_underscore_columns_do_not_collide(self):
 
         sql = str(compiled)
         assert ":`col-name`" in sql
-        assert ":col_name" in sql
+        assert ":`col_name`" in sql
 
         params = compiled.construct_params()
         assert params["col-name"] == "hyphen_value"
         assert params["col_name"] == "underscore_value"
 
-    def test_plain_identifier_bind_names_are_unchanged(self):
-        """No regression: ordinary column names must not be backticked."""
+    def test_plain_identifier_bind_names_are_also_backticked(self):
+        """Every bind name is wrapped unconditionally — the Databricks SQL
+        grammar accepts ``:`id``` identically to ``:id`` for plain names
+        (verified against a live warehouse).
+        """
         metadata = MetaData()
         table = Table(
             "t",
@@ -191,15 +200,10 @@ def test_plain_identifier_bind_names_are_unchanged(self):
         )
         compiled = self._compile_insert(table, {"id": "1", "name": "n"})
         sql = str(compiled)
-        assert ":id" in sql
-        assert ":name" in sql
-        assert ":`id`" not in sql
-        assert ":`name`" not in sql
+        assert ":`id`" in sql
+        assert ":`name`" in sql
 
-    def test_space_and_dot_in_column_name_also_backticked(self):
-        """The bare-identifier check covers all non-[A-Za-z0-9_] characters,
-        not just hyphens — spaces, dots, etc. should also be wrapped.
-        """
+    def test_space_and_dot_in_column_name_are_backticked(self):
         metadata = MetaData()
         table = Table(
             "t",
@@ -218,13 +222,42 @@ def test_space_and_dot_in_column_name_also_backticked(self):
         assert params["col with space"] == "s"
         assert params["col.with.dot"] == "d"
 
-    def test_leading_digit_column_is_backticked(self):
-        """Databricks bind names cannot start with a digit either."""
+    def test_quote_bind_params_can_be_disabled(self):
+        """Setting ``quote_bind_params=False`` on the dialect reverts to
+        stock SQLAlchemy bind-name rendering (the pre-fix behavior).
+        """
+        from databricks.sqlalchemy.base import DatabricksDialect
+
+        dialect = DatabricksDialect()
+        dialect.paramstyle = "named"
+        dialect.quote_bind_params = False
+
         metadata = MetaData()
-        table = Table("t", metadata, Column("1col", String()))
-        compiled = self._compile_insert(table, {"1col": "x"})
+        table = Table("t", metadata, Column("id", String()))
+        compiled = insert(table).values({"id": "1"}).compile(dialect=dialect)
         sql = str(compiled)
-        assert ":`1col`" in sql
+        assert ":id" in sql
+        assert ":`id`" not in sql
 
-        params = compiled.construct_params()
-        assert params["1col"] == "x"
+    def test_url_query_string_disables_quoting(self):
+        """The URL query parameter ``?quote_bind_params=false`` turns the
+        flag off on the dialect.
+        """
+        from sqlalchemy import create_engine
+
+        engine = create_engine(
+            "databricks://token:****@****?http_path=****&catalog=****"
+            "&schema=****&quote_bind_params=false"
+        )
+        # create_engine lazy-initializes; force the dialect to process the URL
+        engine.dialect.create_connect_args(engine.url)
+        assert engine.dialect.quote_bind_params is False
+
+    def test_url_query_string_defaults_to_quoting(self):
+        from sqlalchemy import create_engine
+
+        engine = create_engine(
+            "databricks://token:****@****?http_path=****&catalog=****&schema=****"
+        )
+        engine.dialect.create_connect_args(engine.url)
+        assert engine.dialect.quote_bind_params is True