fix(sqlalchemy-spanner): quote and escape identifiers in generated DDL

packages/sqlalchemy-spanner/google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py

@@ -264,6 +264,16 @@ def __init__(self, dialect):
             dialect, initial_quote="`", final_quote="`"
         )
 
+    def _escape_identifier(self, value):
+        """Escape backslashes and backticks inside a backtick-quoted identifier.
+
+        The base preparer only doubles the ANSI double-quote, which does not
+        neutralize a backtick in a Spanner backtick-quoted identifier. Match the
+        backslash escaping used by ``parse_utils.escape_name`` so a name that
+        carries a backtick cannot terminate the quoted identifier.
+        """
+        return value.replace("\\", "\\\\").replace("`", "\\`")
+
     def _requires_quotes(self, value):
         """Return True if the given identifier requires quoting."""
         lc_value = value.lower()
@@ -697,15 +707,18 @@ def post_create_table(self, table):
         Returns:
             str: primary key difinition to add to the table CREATE request.
         """
-        cols = [col.name for col in table.primary_key.columns]
+        cols = [self.preparer.quote(col.name) for col in table.primary_key.columns]
         post_cmds = " PRIMARY KEY ({})".format(", ".join(cols))
 
         if "TEMPORARY" in table._prefixes:
             raise NotImplementedError("Temporary tables are not supported.")
 
-        if table.kwargs.get("spanner_interleave_in"):
+        parent = table.kwargs.get("spanner_interleave_in")
+        if parent is not None and hasattr(parent, "name"):
+            parent = parent.name
+        if parent:
             post_cmds += ",\nINTERLEAVE IN PARENT {}".format(
-                table.kwargs["spanner_interleave_in"]
+                self.preparer.quote(parent)
             )
 
             if table.kwargs.get("spanner_interleave_on_delete_cascade"):

packages/sqlalchemy-spanner/tests/unit/test_dialect.py

@@ -14,6 +14,8 @@
 
 from unittest.mock import MagicMock
 
+from sqlalchemy import Column, Integer, MetaData, Table
+from sqlalchemy.schema import CreateTable
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing.plugin.plugin_base import fixtures
 
@@ -100,3 +102,53 @@ def test_max_size_exported(self):
         eq_(SpannerDialect.max_size, MAX_SIZE)
         eq_(int_from_size("MAX"), 2621440)
         eq_(int_from_size("100"), 100)
+
+    def _compile_create_table(self, column):
+        """Compile ``CREATE TABLE`` for a one-column primary key table."""
+        table = Table("some_table", MetaData(), column)
+        return str(CreateTable(table).compile(dialect=SpannerDialect()))
+
+    def test_primary_key_reserved_word_is_quoted(self):
+        """A reserved-word primary key column is quoted in the PRIMARY KEY clause."""
+        ddl = self._compile_create_table(Column("from", Integer, primary_key=True))
+        assert "PRIMARY KEY (`from`)" in ddl
+
+    def test_primary_key_backtick_is_escaped(self):
+        """A backtick in a primary key column name cannot terminate the identifier."""
+        name = "id`) STORING (x); DROP TABLE t; --"
+        ddl = self._compile_create_table(Column(name, Integer, primary_key=True))
+        assert "PRIMARY KEY (`id\\`) STORING (x); DROP TABLE t; --`)" in ddl
+
+    def test_primary_key_plain_name_is_unquoted(self):
+        """A regular identifier is left unquoted, so existing DDL is unchanged."""
+        ddl = self._compile_create_table(Column("user_id", Integer, primary_key=True))
+        assert "PRIMARY KEY (user_id)" in ddl
+
+    def test_preparer_escapes_backtick_and_backslash(self):
+        """The identifier preparer backslash-escapes backticks and backslashes."""
+        preparer = SpannerDialect().identifier_preparer
+        eq_(preparer.quote("a`b"), "`a\\`b`")
+        eq_(preparer.quote("a\\b"), "`a\\\\b`")
+
+    def test_interleave_in_parent_string_is_quoted(self):
+        """A string interleave parent is routed through the identifier preparer."""
+        table = Table(
+            "child",
+            MetaData(),
+            Column("id", Integer, primary_key=True),
+            spanner_interleave_in="from",
+        )
+        ddl = str(CreateTable(table).compile(dialect=SpannerDialect()))
+        assert "INTERLEAVE IN PARENT `from`" in ddl
+
+    def test_interleave_in_parent_table_object_is_quoted(self):
+        """A Table interleave parent is quoted by its name rather than repr."""
+        parent = Table("from", MetaData(), Column("id", Integer, primary_key=True))
+        table = Table(
+            "child",
+            MetaData(),
+            Column("id", Integer, primary_key=True),
+            spanner_interleave_in=parent,
+        )
+        ddl = str(CreateTable(table).compile(dialect=SpannerDialect()))
+        assert "INTERLEAVE IN PARENT `from`" in ddl