chore: update to version and add session mode support

This commit updates the Databricks adapter to version 1.10.15+session, introducing support for session mode execution. Key changes include:

- Added `DatabricksSessionHandle` and `SessionCursorWrapper` for handling SparkSession-based execution.
- Enhanced `DatabricksCredentials` to manage connection methods and validate session mode configurations.
- Updated connection management to support session mode, including automatic selection of submission methods for Python models.
- Improve SparkSession retrieval in Databricks adapter. This commit enhances the `DatabricksSessionHandle` and `SessionPythonJobHelper` classes to improve the retrieval of the existing SparkSession. It introduces multiple methods to obtain the SparkSession, ensuring compatibility with various Databricks environments. Additionally, it refactors method signatures for consistency and readability.

Author: alexeyegorov

dbt/adapters/databricks/__version__.py

@@ -1 +1 @@
-version = "1.11.4"
+version = "1.10.15-5"

dbt/adapters/databricks/connections.py

@@ -288,6 +288,47 @@ def translate_bindings(bindings: Optional[Sequence[Any]]) -> Optional[Sequence[A
             return list(map(lambda x: float(x) if isinstance(x, decimal.Decimal) else x, bindings))
         return None
 
+    @staticmethod
+    def format_bindings_for_sql(bindings: Optional[Sequence[Any]]) -> Optional[Sequence[str]]:
+        """
+        Format bindings as SQL literals for string substitution in session mode.
+        
+        This method properly quotes string values and handles special cases to ensure
+        SQL injection safety and correct SQL syntax. Used when executing SQL via
+        SparkSession.sql() which doesn't support parameterized queries.
+        
+        Args:
+            bindings: Sequence of binding values (strings, numbers, None, etc.)
+            
+        Returns:
+            Sequence of SQL literal strings, or None if bindings is None/empty
+        """
+        if not bindings:
+            return None
+        
+        formatted = []
+        for value in bindings:
+            if value is None:
+                formatted.append("NULL")
+            elif isinstance(value, bool):
+                formatted.append("TRUE" if value else "FALSE")
+            elif isinstance(value, str):
+                # Escape single quotes by doubling them, then wrap in quotes
+                escaped = value.replace("'", "''")
+                formatted.append(f"'{escaped}'")
+            elif isinstance(value, (int, float, decimal.Decimal)):
+                # Numbers don't need quotes
+                if isinstance(value, decimal.Decimal):
+                    formatted.append(str(float(value)))
+                else:
+                    formatted.append(str(value))
+            else:
+                # For other types, convert to string and quote
+                escaped = str(value).replace("'", "''")
+                formatted.append(f"'{escaped}'")
+        
+        return formatted
+
     @staticmethod
     def clean_sql(sql: str) -> str:
         cleaned = sql.strip()

dbt/adapters/databricks/credentials.py

@@ -57,6 +57,7 @@
     AllPurposeClusterPythonJobHelper,
     JobClusterPythonJobHelper,
     ServerlessClusterPythonJobHelper,
+    SessionPythonJobHelper,
     WorkflowPythonJobHelper,
 )
 from dbt.adapters.databricks.relation import (
@@ -801,6 +802,7 @@ def python_submission_helpers(self) -> dict[str, type[PythonJobHelper]]:
             "all_purpose_cluster": AllPurposeClusterPythonJobHelper,
             "serverless_cluster": ServerlessClusterPythonJobHelper,
             "workflow_job": WorkflowPythonJobHelper,
+            "session": SessionPythonJobHelper,
         }
 
     @log_code_execution
@@ -809,6 +811,17 @@ def submit_python_job(self, parsed_model: dict, compiled_code: str) -> AdapterRe
             "user_folder_for_python",
             self.behavior.use_user_folder_for_python.setting,  # type: ignore[attr-defined]
         )
+
+        # Auto-select session submission when in session mode
+        from dbt.adapters.databricks.credentials import DatabricksCredentials
+        from dbt.adapters.databricks.logging import logger
+
+        creds = cast(DatabricksCredentials, self.config.credentials)
+        if creds.is_session_mode:
+            if parsed_model["config"].get("submission_method") is None:
+                parsed_model["config"]["submission_method"] = "session"
+                logger.debug("Auto-selected 'session' submission method for Python model")
+
         return super().submit_python_job(parsed_model, compiled_code)
 
     @available