Fix: Spark partition by on state tables (#1238)

* convert strings to columns

* dry up mock engine adapter creation

Author: eakmanrq

sqlmesh/core/engine_adapter/spark.py

@@ -212,7 +212,7 @@ def create_state_table(
         self.create_table(
             table_name,
             columns_to_types,
-            partitioned_by=primary_key,
+            partitioned_by=[exp.column(x) for x in primary_key] if primary_key else None,
         )
 
     def create_view(

tests/conftest.py

@@ -23,6 +23,8 @@
 
 pytest_plugins = ["tests.common_fixtures"]
 
+T = t.TypeVar("T", bound=EngineAdapter)
+
 
 class DuckDBMetadata:
     def __init__(self, engine_adapter: EngineAdapter):
@@ -244,6 +246,17 @@ def sushi_fixed_date_data_validator(sushi_context_fixed_date: Context) -> SushiD
     return SushiDataValidator.from_context(sushi_context_fixed_date)
 
 
+@pytest.fixture
+def make_mocked_engine_adapter(mocker: MockerFixture) -> t.Callable:
+    def _make_function(klass: t.Type[T], dialect: t.Optional[str] = None) -> T:
+        connection_mock = mocker.NonCallableMock()
+        cursor_mock = mocker.Mock()
+        connection_mock.cursor.return_value = cursor_mock
+        return klass(lambda: connection_mock, dialect=dialect or klass.DIALECT)
+
+    return _make_function
+
+
 def delete_cache(project_paths: str | t.List[str]) -> None:
     for path in ensure_list(project_paths):
         try: