fix: escape SQL identifiers in db engine spec prequeries and metadata queries (#39840)

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

tests/unit_tests/db_engine_specs/test_databricks.py

@@ -281,6 +281,13 @@ def test_get_prequeries(mocker: MockerFixture) -> None:
     assert DatabricksNativeEngineSpec.get_prequeries(
         database, catalog="`escaped-hyphen`", schema="`hyphen-escaped`"
     ) == [
-        "USE CATALOG `escaped-hyphen`",
-        "USE SCHEMA `hyphen-escaped`",
+        "USE CATALOG ```escaped-hyphen```",
+        "USE SCHEMA ```hyphen-escaped```",
+    ]
+
+    assert DatabricksNativeEngineSpec.get_prequeries(
+        database, catalog="evil` USE CATALOG bad", schema="evil` USE SCHEMA bad"
+    ) == [
+        "USE CATALOG `evil`` USE CATALOG bad`",
+        "USE SCHEMA `evil`` USE SCHEMA bad`",
     ]

tests/unit_tests/db_engine_specs/test_db2.py

@@ -81,6 +81,9 @@ def test_get_prequeries(mocker: MockerFixture) -> None:
     assert Db2EngineSpec.get_prequeries(database, schema="my_schema") == [
         'set current_schema "my_schema"'
     ]
+    assert Db2EngineSpec.get_prequeries(database, schema='evil"; SELECT 1--') == [
+        'set current_schema "evil""; SELECT 1--"'
+    ]
 
 
 @pytest.mark.parametrize(

tests/unit_tests/db_engine_specs/test_hive.py

@@ -99,3 +99,87 @@ SELECT * \nFROM my_schema.my_table
  LIMIT :param_1
     """.strip()
     )
+
+
+def test_get_view_names_escapes_schema(mocker: MockerFixture) -> None:
+    """
+    Test that ``get_view_names`` correctly escapes backticks in schema names
+    within the SHOW VIEWS statement.
+    """
+    from superset.db_engine_specs.hive import HiveEngineSpec
+
+    database = mocker.MagicMock()
+    inspector = mocker.MagicMock()
+
+    conn = mocker.MagicMock()
+    cursor = mocker.MagicMock()
+    cursor.fetchall.return_value = []
+    conn.__enter__ = mocker.MagicMock(return_value=conn)
+    conn.__exit__ = mocker.MagicMock(return_value=False)
+    conn.cursor.return_value = cursor
+    database.get_raw_connection.return_value = conn
+
+    HiveEngineSpec.get_view_names(database, inspector, schema="evil` UNION SELECT 1--")
+    cursor.execute.assert_called_once()
+    sql = cursor.execute.call_args[0][0]
+    assert "IN `evil`` UNION SELECT 1--`" in sql
+
+
+def test_df_to_sql_escapes_like_wildcards(mocker: MockerFixture) -> None:
+    """
+    Test that ``df_to_sql`` escapes ``%`` and ``_`` wildcard characters in the
+    SHOW TABLES LIKE pattern used to detect table existence.
+    """
+    import pandas as pd
+
+    from superset.db_engine_specs.hive import HiveEngineSpec
+    from superset.exceptions import SupersetException
+    from superset.sql.parse import Table
+
+    database = mocker.MagicMock()
+    # Simulate an existing table so df_to_sql raises before reaching the upload path
+    database.get_df.return_value = pd.DataFrame({"name": ["sales_%_2024"]})
+
+    with pytest.raises(SupersetException, match="Table already exists"):
+        HiveEngineSpec.df_to_sql(
+            database=database,
+            table=Table("sales_%_2024", "my_schema"),
+            df=pd.DataFrame({"a": [1]}),
+            to_sql_kwargs={"if_exists": "fail"},
+        )
+
+    database.get_df.assert_called_once()
+    sql = database.get_df.call_args[0][0]
+    assert r"\%" in sql
+    assert r"\_" in sql
+    assert "ESCAPE" in sql
+
+
+def test_partition_query_escapes_identifiers() -> None:
+    """
+    Test that ``_partition_query`` correctly backtick-quotes table and schema names
+    in the SHOW PARTITIONS statement.
+    """
+    from superset.db_engine_specs.hive import HiveEngineSpec
+    from superset.sql.parse import Table
+
+    result = HiveEngineSpec._partition_query(
+        table=Table("my_table", "my_schema"),
+        indexes=[],
+        database=None,  # type: ignore
+    )
+    assert result == "SHOW PARTITIONS `my_schema`.`my_table`"
+
+    result = HiveEngineSpec._partition_query(
+        table=Table("evil`tbl", "evil`schema"),
+        indexes=[],
+        database=None,  # type: ignore
+    )
+    assert result == "SHOW PARTITIONS `evil``schema`.`evil``tbl`"
+
+    result = HiveEngineSpec._partition_query(
+        table=Table("no_schema_tbl"),
+        indexes=[],
+        database=None,  # type: ignore
+    )
+    assert result == "SHOW PARTITIONS `no_schema_tbl`"

tests/unit_tests/db_engine_specs/test_postgres.py

@@ -147,6 +147,9 @@ def test_get_prequeries(mocker: MockerFixture) -> None:
 
     assert spec.get_prequeries(database) == []
     assert spec.get_prequeries(database, schema="test") == ['set search_path = "test"']
+    assert spec.get_prequeries(database, schema='evil"; SELECT 1--') == [
+        'set search_path = "evil""; SELECT 1--"'
+    ]
 
 
 def test_get_default_schema_for_query(mocker: MockerFixture) -> None:

tests/unit_tests/db_engine_specs/test_starrocks.py

@@ -169,6 +169,11 @@ def test_impersonation_username(mocker: MockerFixture) -> None:
         'EXECUTE AS "alice" WITH NO REVERT;'
     ]
 
+    database.get_effective_user.return_value = 'evil" WITH NO REVERT; DROP TABLE x--'
+    assert StarRocksEngineSpec.get_prequeries(database) == [
+        'EXECUTE AS "evil"" WITH NO REVERT; DROP TABLE x--" WITH NO REVERT;'
+    ]
+
 
 def test_impersonation_disabled(mocker: MockerFixture) -> None:
     """