Add opt-out gate for multi-row insert casts.

Allow users to disable targeted multi-row insert cast rendering with enable_multirow_insert_casts=false in the Databricks SQLAlchemy engine URL while keeping the PECOBLR-2746 fix enabled by default.

Signed-off-by: Madhavendra Rathore <madhavendra.rathore@databricks.com>

Author: msrathore-db

src/databricks/sqlalchemy/_ddl.py

@@ -223,6 +223,9 @@ def _build_multi_value_cast_plan(self, insert_stmt):
         Spark inline-table incompatibility for object columns that mix
         primitive families (e.g. INT + STRING).
         """
+        if not self.dialect.enable_multirow_insert_casts:
+            return {}
+
         if not getattr(insert_stmt, "_multi_values", None):
             return {}
 

src/databricks/sqlalchemy/base.py

@@ -42,6 +42,12 @@ class DatabricksImpl(DefaultImpl):
 logger = logging.getLogger(__name__)
 
 
+def _parse_bool_url_param(value: Optional[str], default: bool) -> bool:
+    if value is None:
+        return default
+    return value.lower() not in ("0", "false", "no", "off")
+
+
 class DatabricksDialect(default.DefaultDialect):
     """This dialect implements only those methods required to pass our e2e tests"""
 
@@ -65,6 +71,7 @@ class DatabricksDialect(default.DefaultDialect):
     supports_server_side_cursors: bool = False
     supports_sequences: bool = False
     supports_native_boolean: bool = True
+    enable_multirow_insert_casts: bool = True
 
     colspecs = {
         sqlalchemy.types.DateTime: dialect_type_impl.TIMESTAMP_NTZ,
@@ -117,6 +124,9 @@ def create_connect_args(self, url):
 
         self.schema = kwargs["schema"]
         self.catalog = kwargs["catalog"]
+        self.enable_multirow_insert_casts = _parse_bool_url_param(
+            url.query.get("enable_multirow_insert_casts"), True
+        )
 
         self._force_paramstyle_to_native_mode()
 

tests/test_local/test_ddl.py

@@ -458,6 +458,22 @@ def test_multi_values_casts_mixed_type_column(self):
         assert "CAST(:`name_m1` AS STRING)" not in sql
         assert "CAST(:`name_m2` AS STRING)" not in sql
 
+    def test_multi_value_casts_can_be_disabled_by_url_param(self):
+        engine = create_engine(
+            "databricks://token:****@****"
+            "?http_path=****&catalog=****&schema=****"
+            "&enable_multirow_insert_casts=false"
+        )
+        metadata = MetaData()
+        table = Table("t", metadata, Column("value", String()))
+        stmt = insert(table).values([{"value": 1}, {"value": 0}, {"value": "NE"}])
+
+        sql = str(stmt.compile(bind=engine))
+        assert "CAST(:`value_m0` AS STRING)" not in sql
+        assert ":`value_m0`" in sql
+        assert ":`value_m1`" in sql
+        assert ":`value_m2`" in sql
+
     def test_homogeneous_multi_values_are_not_cast(self):
         metadata = MetaData()
         table = Table("t", metadata, Column("value", String()))