Fix: Pass table properties when replacing a query in the Spark engine adapter

sqlmesh/core/engine_adapter/mixins.py

@@ -76,6 +76,7 @@ def overwrite_target_from_temp(
         query: Query,
         columns_to_types: t.Dict[str, exp.DataType],
         target_table: TableName,
+        **kwargs: t.Any,
     ) -> None:
         """
         Overwrites the target table from the temp table. This is used when the target table is self-referencing.
@@ -84,6 +85,7 @@ def overwrite_target_from_temp(
             exp.select(*columns_to_types).from_(target_table),
             target_table,
             columns_to_types,
+            **kwargs,
         ) as temp_table:
 
             def replace_table(

sqlmesh/core/engine_adapter/spark.py

@@ -372,10 +372,10 @@ def replace_query(
 
             if self_referencing:
                 return LogicalReplaceQueryMixin.overwrite_target_from_temp(
-                    self, query, columns_to_types, target_table
+                    self, query, columns_to_types, target_table, **kwargs
                 )
 
-        self.create_table(table_name, columns_to_types)
+        self.create_table(table_name, columns_to_types, **kwargs)
         return self._insert_overwrite_by_condition(
             table_name, source_queries, columns_to_types, where=exp.true()
         )

tests/core/engine_adapter/test_spark.py

@@ -56,6 +56,35 @@ def test_create_table_properties(make_mocked_engine_adapter: t.Callable):
         )
 
 
+def test_replace_query_table_properties(make_mocked_engine_adapter: t.Callable):
+    adapter = make_mocked_engine_adapter(SparkEngineAdapter)
+
+    columns_to_types = {
+        "cola": exp.DataType.build("INT"),
+        "colb": exp.DataType.build("TEXT"),
+        "colc": exp.DataType.build("TEXT"),
+    }
+    adapter.replace_query(
+        "test_table",
+        parse_one("SELECT 1 AS cola, '2' AS colb, '3' AS colc"),
+        columns_to_types=columns_to_types,
+        partitioned_by=[exp.to_column("colb")],
+        storage_format="ICEBERG",
+        table_properties={"a": exp.convert(1)},
+    )
+
+    adapter.cursor.execute.assert_has_calls(
+        [
+            call(
+                "CREATE TABLE IF NOT EXISTS `test_table` (`cola` INT, `colb` STRING, `colc` STRING) USING ICEBERG PARTITIONED BY (`colb`) TBLPROPERTIES ('a'=1)"
+            ),
+            call(
+                "INSERT OVERWRITE TABLE `test_table` (`cola`, `colb`, `colc`) SELECT `cola`, `colb`, `colc` FROM (SELECT 1 AS `cola`, '2' AS `colb`, '3' AS `colc`) AS `_subquery` WHERE TRUE"
+            ),
+        ]
+    )
+
+
 def test_create_view_properties(make_mocked_engine_adapter: t.Callable):
     adapter = make_mocked_engine_adapter(SparkEngineAdapter)