fix: prevent SQL parameterization of duplicate_key_handling in MySQL bulk_load_custom

providers/mysql/src/airflow/providers/mysql/hooks/mysql.py

@@ -21,7 +21,7 @@
 
 import json
 import logging
-from typing import TYPE_CHECKING, Any, Union
+from typing import TYPE_CHECKING, Any, Literal, Union
 from urllib.parse import quote_plus, urlencode
 
 from airflow.providers.common.compat.sdk import AirflowOptionalProviderFeatureException
@@ -318,7 +318,11 @@
         return token, port
 
     def bulk_load_custom(
-        self, table: str, tmp_file: str, duplicate_key_handling: str = "IGNORE", extra_options: str = ""
+        self,
+        table: str,
+        tmp_file: str,
+        duplicate_key_handling: Literal["IGNORE", "REPLACE", ""] = "IGNORE",
+        extra_options: str = "",
     ) -> None:
         """
         Load local data from a file into the database in a more configurable way.
@@ -339,11 +343,31 @@
 
             .. seealso:: https://dev.mysql.com/doc/refman/8.0/en/load-data.html
         """
+        _VALID_DUPLICATE_KEY_HANDLING = {"IGNORE", "REPLACE", ""}
+        if duplicate_key_handling not in _VALID_DUPLICATE_KEY_HANDLING:
+            raise ValueError(
+                f"Invalid duplicate_key_handling: {duplicate_key_handling!r}. "
+                f"Must be one of {_VALID_DUPLICATE_KEY_HANDLING}."
+            )
+
+        import re
+
+        if extra_options and not re.match(r"^[A-Z @=',;()\w\s.*/-]+$", extra_options):
+            raise ValueError(
+                f"Invalid extra_options: {extra_options!r}. "
+                "Only alphanumeric characters, spaces, and common SQL clauses are allowed."
+            )
+
         conn = self.get_conn()
         cursor = conn.cursor()
 
-        sql_statement = f"LOAD DATA LOCAL INFILE %s %s INTO TABLE `{table}` %s"
-        parameters = (tmp_file, duplicate_key_handling, extra_options)
+        # duplicate_key_handling and extra_options are SQL keywords (e.g. IGNORE, REPLACE)
+        # and must be interpolated into the statement, not passed as query parameters,
+        # because parameterized values get quoted as strings which produces invalid SQL.
+        sql_statement = (
+            f"LOAD DATA LOCAL INFILE %s {duplicate_key_handling} INTO TABLE `{table}` {extra_options}"
+        )
+        parameters = (tmp_file,)
         cursor.execute(
             sql_statement,
             parameters,

providers/mysql/tests/unit/mysql/hooks/test_mysql.py

@@ -506,16 +506,25 @@ def test_bulk_load_custom(self, table):
             IGNORE 1 LINES""",
         )
         self.cur.execute.assert_called_once_with(
-            f"LOAD DATA LOCAL INFILE %s %s INTO TABLE `{table}` %s",
-            (
-                "/tmp/file",
-                "IGNORE",
-                """FIELDS TERMINATED BY ';'
-            OPTIONALLY ENCLOSED BY '"'
-            IGNORE 1 LINES""",
-            ),
+            f"LOAD DATA LOCAL INFILE %s IGNORE INTO TABLE `{table}` FIELDS TERMINATED BY ';'\n            OPTIONALLY ENCLOSED BY '\"'\n            IGNORE 1 LINES",
+            ("/tmp/file",),
         )
 
+    @pytest.mark.parametrize("duplicate_key_handling", ["IGNORE", "REPLACE"])
+    def test_bulk_load_custom_duplicate_key_not_parameterized(self, duplicate_key_handling):
+        """Verify duplicate_key_handling is interpolated into SQL, not passed as a query parameter."""
+        self.db_hook.bulk_load_custom(
+            "table",
+            "/tmp/file",
+            duplicate_key_handling,
+            "",
+        )
+        executed_sql = self.cur.execute.call_args[0][0]
+        # The keyword must appear literally in the SQL, not as a %s placeholder
+        assert duplicate_key_handling in executed_sql
+        # Only tmp_file should be parameterized
+        assert self.cur.execute.call_args[0][1] == ("/tmp/file",)
+
     @mock.patch("airflow.providers.mysql.hooks.mysql.send_sql_hook_lineage")
     def test_bulk_load_custom_hook_lineage(self, mock_send_lineage):
         self.db_hook.bulk_load_custom(
@@ -527,8 +536,10 @@ def test_bulk_load_custom_hook_lineage(self, mock_send_lineage):
         mock_send_lineage.assert_called_once()
         call_kw = mock_send_lineage.call_args.kwargs
         assert call_kw["context"] is self.db_hook
-        assert call_kw["sql"] == "LOAD DATA LOCAL INFILE %s %s INTO TABLE `table` %s"
-        assert call_kw["sql_parameters"] == ("/tmp/file", "IGNORE", "FIELDS TERMINATED BY ';'")
+        assert (
+            call_kw["sql"] == "LOAD DATA LOCAL INFILE %s IGNORE INTO TABLE `table` FIELDS TERMINATED BY ';'"
+        )
+        assert call_kw["sql_parameters"] == ("/tmp/file",)
         assert call_kw["cur"] is self.cur
 
     def test_reserved_words(self):