add python models session submission method

dbt/adapters/databricks/__version__.py

@@ -1 +1 @@
-version: str = "1.8.6"
+version: str = "1.8.6a1"

dbt/adapters/databricks/impl.py

@@ -45,6 +45,7 @@
 from dbt.adapters.databricks.connections import USE_LONG_SESSIONS
 from dbt.adapters.databricks.python_submissions import (
     DbtDatabricksAllPurposeClusterPythonJobHelper,
+    SessionHelper
 )
 from dbt.adapters.databricks.python_submissions import (
     DbtDatabricksJobClusterPythonJobHelper,
@@ -611,6 +612,7 @@ def python_submission_helpers(self) -> Dict[str, Type[PythonJobHelper]]:
         return {
             "job_cluster": DbtDatabricksJobClusterPythonJobHelper,
             "all_purpose_cluster": DbtDatabricksAllPurposeClusterPythonJobHelper,
+            "session": SessionHelper,
         }
 
     @available

dbt/adapters/databricks/python_submissions.py

@@ -600,3 +600,16 @@ def check_credentials(self) -> None:
                 "Databricks `http_path` or `cluster_id` of an all-purpose cluster is required "
                 "for the `all_purpose_cluster` submission method."
             )
+
+
+class SessionHelper(PythonJobHelper):
+    def __init__(self, parsed_model: Dict, credentials: DatabricksCredentials) -> None:
+        pass
+
+    def submit(self, compiled_code: str) -> Any:
+        try:
+            from pyspark.sql import SparkSession
+            spark = SparkSession.getActiveSession()
+            exec(compiled_code, {"spark": spark})
+        except Exception as e:
+            raise DbtRuntimeError(f"Python model failed with traceback as:\n{e}")