perf: add SparkDirectSeeder (bypass dbt seed) and tune Spark config for -n8 parallelism

- Add SparkDirectSeeder that executes CREATE TABLE + INSERT VALUES directly
  via the dbt adapter, bypassing the ~4s dbt subprocess overhead per seed
- Add execute_sql() and schema_name property to AdapterQueryRunner
- DbtProject auto-selects SparkDirectSeeder when target is 'spark'
- Tune spark-defaults.conf: executor.cores=4, default.parallelism=4,
  thriftServer.async=true for better concurrent session handling
- Restore -n8 parallelism for Spark in CI (was -n4)

Co-Authored-By: Itamar Hartstein <haritamar@gmail.com>

Author: devin-ai-integration[bot]

.github/workflows/test-warehouse.yml

@@ -177,7 +177,7 @@ jobs:
 
       - name: Test
         working-directory: "${{ env.TESTS_DIR }}/tests"
-        run: py.test -n${{ (inputs.warehouse-type == 'spark' && '4') || '8' }} -vvv --target "${{ inputs.warehouse-type }}" --junit-xml=test-results.xml --html=detailed_report_${{ inputs.warehouse-type }}_dbt_${{ inputs.dbt-version }}.html --self-contained-html --clear-on-end ${{ (inputs.dbt-version == 'fusion' && '--runner-method fusion') || '' }}
+        run: py.test -n8 -vvv --target "${{ inputs.warehouse-type }}" --junit-xml=test-results.xml --html=detailed_report_${{ inputs.warehouse-type }}_dbt_${{ inputs.dbt-version }}.html --self-contained-html --clear-on-end ${{ (inputs.dbt-version == 'fusion' && '--runner-method fusion') || '' }}
 
       - name: Upload test results
         if: always()

integration_tests/docker/spark/spark-defaults.conf

@@ -1,5 +1,6 @@
 spark.driver.memory                    2g
 spark.executor.memory                  2g
+spark.executor.cores                   4
 spark.hadoop.datanucleus.autoCreateTables       true
 spark.hadoop.datanucleus.schema.autoCreateTables true
 spark.hadoop.datanucleus.fixedDatastore          false
@@ -8,6 +9,7 @@ spark.driver.userClassPathFirst        true
 spark.sql.extensions                   io.delta.sql.DeltaSparkSessionExtension
 spark.sql.catalog.spark_catalog        org.apache.spark.sql.delta.catalog.DeltaCatalog
 spark.sql.shuffle.partitions           2
-spark.default.parallelism              2
+spark.default.parallelism              4
 spark.ui.enabled                       false
 spark.sql.adaptive.enabled             true
+spark.sql.hive.thriftServer.async      true

integration_tests/tests/adapter_query_runner.py

@@ -239,6 +239,16 @@ def has_non_ref_jinja(query: str) -> bool:
         stripped = _SOURCE_PATTERN.sub("", stripped)
         return bool(_JINJA_EXPR_PATTERN.search(stripped))
 
+    def execute_sql(self, sql: str) -> None:
+        """Execute a SQL statement that does not return results (DDL/DML)."""
+        with self._adapter.connection_named("execute_sql"):
+            self._adapter.execute(sql, fetch=False)
+
+    @property
+    def schema_name(self) -> str:
+        """Return the base schema name from the adapter credentials."""
+        return self._adapter.config.credentials.schema
+
     def run_query(self, prerendered_query: str) -> List[Dict[str, Any]]:
         """Render Jinja refs/sources and execute a query, returning rows as dicts.
 

integration_tests/tests/data_seeder.py

@@ -1,17 +1,20 @@
 import csv
 from contextlib import contextmanager
 from pathlib import Path
-from typing import Generator, List
+from typing import TYPE_CHECKING, Generator, List
 
 from elementary.clients.dbt.base_dbt_runner import BaseDbtRunner
 from logger import get_logger
 
-# TODO: Write more performant data seeders per adapter.
+if TYPE_CHECKING:
+    from adapter_query_runner import AdapterQueryRunner
 
 logger = get_logger(__name__)
 
 
 class DbtDataSeeder:
+    """Default seeder: writes a CSV and calls ``dbt seed``."""
+
     def __init__(
         self, dbt_runner: BaseDbtRunner, dbt_project_path: Path, seeds_dir_path: Path
     ):
@@ -48,3 +51,67 @@ def seed(self, data: List[dict], table_name: str) -> Generator[None, None, None]
                 yield
         finally:
             seed_path.unlink()
+
+
+# Maximum number of rows per INSERT VALUES statement.  Spark's Thrift
+# protocol can choke on very large statements, so we batch inserts.
+_INSERT_BATCH_SIZE = 500
+
+
+class SparkDirectSeeder:
+    """Fast seeder for Spark: executes CREATE TABLE + INSERT directly.
+
+    Bypasses the ``dbt seed`` subprocess entirely, avoiding the ~4 s
+    Python/manifest-parsing overhead per invocation.  All columns are
+    created as STRING, which matches ``dbt seed`` behaviour.
+    """
+
+    def __init__(self, query_runner: "AdapterQueryRunner", schema: str) -> None:
+        self._query_runner = query_runner
+        self._schema = schema
+
+    # ------------------------------------------------------------------
+    # helpers
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _escape(value: object) -> str:
+        """Escape a value for a Spark SQL string literal."""
+        if value is None or (isinstance(value, str) and value == ""):
+            return "NULL"
+        text = str(value)
+        # Replace backslashes first, then single-quotes.
+        text = text.replace("\\", "\\\\")
+        text = text.replace("'", "\\'")
+        # Spark INSERT VALUES doesn't support embedded newlines.
+        text = text.replace("\n", " ").replace("\r", " ")
+        return f"'{text}'"
+
+    # ------------------------------------------------------------------
+    # public API (same shape as DbtDataSeeder)
+    # ------------------------------------------------------------------
+
+    @contextmanager
+    def seed(self, data: List[dict], table_name: str) -> Generator[None, None, None]:
+        columns = list(data[0].keys())
+        col_defs = ", ".join(f"`{col}` STRING" for col in columns)
+        fq_table = f"`{self._schema}`.`{table_name}`"
+
+        # DROP + CREATE is the fastest way to get a clean table.
+        self._query_runner.execute_sql(f"DROP TABLE IF EXISTS {fq_table}")
+        self._query_runner.execute_sql(
+            f"CREATE TABLE {fq_table} ({col_defs}) USING DELTA"
+        )
+
+        # Insert in batches.
+        for batch_start in range(0, len(data), _INSERT_BATCH_SIZE):
+            batch = data[batch_start : batch_start + _INSERT_BATCH_SIZE]
+            rows_sql = ", ".join(
+                "(" + ", ".join(self._escape(row.get(c)) for c in columns) + ")"
+                for row in batch
+            )
+            self._query_runner.execute_sql(f"INSERT INTO {fq_table} VALUES {rows_sql}")
+
+        logger.info("SparkDirectSeeder: loaded %d rows into %s", len(data), fq_table)
+
+        yield

integration_tests/tests/dbt_project.py

@@ -7,7 +7,7 @@
 from uuid import uuid4
 
 from adapter_query_runner import AdapterQueryRunner, UnsupportedJinjaError
-from data_seeder import DbtDataSeeder
+from data_seeder import DbtDataSeeder, SparkDirectSeeder
 from dbt_utils import get_database_and_schema_properties
 from elementary.clients.dbt.base_dbt_runner import BaseDbtRunner
 from elementary.clients.dbt.factory import RunnerMethod, create_dbt_runner
@@ -326,10 +326,18 @@ def test(
             }
             return [test_result] if multiple_results else test_result
 
-    def seed(self, data: List[dict], table_name: str):
-        with DbtDataSeeder(
+    def _create_seeder(self) -> Union[DbtDataSeeder, SparkDirectSeeder]:
+        """Return the fastest available seeder for the current target."""
+        if self.target == "spark":
+            runner = self._get_query_runner()
+            schema = runner.schema_name + SCHEMA_NAME_SUFFIX
+            return SparkDirectSeeder(runner, schema)
+        return DbtDataSeeder(
             self.dbt_runner, self.project_dir_path, self.seeds_dir_path
-        ).seed(data, table_name):
+        )
+
+    def seed(self, data: List[dict], table_name: str):
+        with self._create_seeder().seed(data, table_name):
             self._fix_seed_if_needed(table_name)
 
     def _fix_seed_if_needed(self, table_name: str):
@@ -345,9 +353,7 @@ def _fix_seed_if_needed(self, table_name: str):
     def seed_context(
         self, data: List[dict], table_name: str
     ) -> Generator[None, None, None]:
-        with DbtDataSeeder(
-            self.dbt_runner, self.project_dir_path, self.seeds_dir_path
-        ).seed(data, table_name):
+        with self._create_seeder().seed(data, table_name):
             yield
 
     @contextmanager