feat: add Vertica adapter support

.github/workflows/test-all-warehouses.yml

@@ -49,7 +49,16 @@ jobs:
           ${{ inputs.dbt-version && fromJSON(format('["{0}"]', inputs.dbt-version)) ||
           fromJSON('["latest_official", "latest_pre"]') }}
         warehouse-type:
-          [postgres, clickhouse, trino, dremio, spark, duckdb, sqlserver]
+          [
+            postgres,
+            clickhouse,
+            trino,
+            dremio,
+            spark,
+            duckdb,
+            sqlserver,
+            vertica,
+          ]
         exclude:
           # latest_pre is only tested on postgres
           - dbt-version: latest_pre
@@ -64,6 +73,8 @@ jobs:
             warehouse-type: duckdb
           - dbt-version: latest_pre
             warehouse-type: sqlserver
+          - dbt-version: latest_pre
+            warehouse-type: vertica
     uses: ./.github/workflows/test-warehouse.yml
     with:
       warehouse-type: ${{ matrix.warehouse-type }}

.github/workflows/test-warehouse.yml

@@ -20,6 +20,7 @@ on:
           - duckdb
           - sqlserver
           - fabric
+          - vertica
       elementary-ref:
         type: string
         required: false
@@ -151,8 +152,26 @@ jobs:
         if: startsWith(inputs.warehouse-type, 'databricks') && inputs.dbt-version < '1.7.0'
         run: pip install databricks-sql-connector==2.9.3
 
+      - name: Reject unsupported Vertica + Fusion combination
+        if: inputs.warehouse-type == 'vertica' && inputs.dbt-version == 'fusion'
+        run: |
+          echo "::error::dbt Fusion does not support third-party adapters such as dbt-vertica."
+          exit 1
+
+      - name: Install dbt-vertica
+        if: inputs.warehouse-type == 'vertica' && inputs.dbt-version != 'fusion'
+        run: |
+          # dbt-vertica pins dbt-core~=1.8 which lacks native support for the
+          # "arguments" test property used by the integration-test framework.
+          # Install dbt-vertica without deps, then install the requested
+          # dbt-core version separately (dbt-vertica works fine with newer
+          # dbt-core versions).
+          pip install dbt-vertica --no-deps
+          pip install vertica-python \
+            "dbt-core${{ (!startsWith(inputs.dbt-version, 'latest') && format('=={0}', inputs.dbt-version)) || '' }}"
+
       - name: Install dbt
-        if: ${{ inputs.dbt-version != 'fusion' }}
+        if: ${{ inputs.dbt-version != 'fusion' && inputs.warehouse-type != 'vertica' }}
         run:
           pip install${{ (inputs.dbt-version == 'latest_pre' && ' --pre') || '' }}
           "dbt-core${{ (!startsWith(inputs.dbt-version, 'latest') && format('=={0}', inputs.dbt-version)) || '' }}"
@@ -198,6 +217,18 @@ jobs:
           ln -sfn ${{ github.workspace }}/dbt-data-reliability dbt_project/dbt_packages/elementary
           pip install -r requirements.txt
 
+      - name: Start Vertica
+        if: inputs.warehouse-type == 'vertica'
+        working-directory: ${{ env.TESTS_DIR }}
+        run: docker compose -f docker-compose-vertica.yml up -d
+
+      - name: Wait for Vertica to be ready
+        if: inputs.warehouse-type == 'vertica'
+        run: |
+          echo "Waiting for Vertica to be healthy..."
+          timeout 60 bash -c 'until [ "$(docker inspect --format="{{.State.Health.Status}}" vertica)" == "healthy" ]; do echo "Waiting..."; sleep 5; done'
+          echo "Vertica is ready!"
+
       - name: Check DWH connection
         working-directory: ${{ env.TESTS_DIR }}
         run: |

.gitignore

@@ -4,6 +4,7 @@ dbt_internal_packages/
 logs/
 scripts/
 
+.github/fixtures/.user.yml
 .idea
 .DS_Store
 

integration_tests/dbt_project/macros/ci_schemas_cleanup/test_drop_stale_ci_schemas.sql

@@ -98,3 +98,9 @@
     {% set safe_schema = schema_name | replace("`", "``") %}
     {% do run_query("CREATE DATABASE IF NOT EXISTS `" ~ safe_schema ~ "`") %}
 {% endmacro %}
+
+{% macro vertica__edr_create_schema(database, schema_name) %}
+    {#- Vertica DDL is auto-committed; an explicit adapter.commit() would
+        fail with "no transaction in progress". -#}
+    {% do run_query("CREATE SCHEMA IF NOT EXISTS " ~ schema_name) %}
+{% endmacro %}

integration_tests/dbt_project/macros/clear_env.sql

@@ -82,3 +82,9 @@
     {% do run_query("DROP SCHEMA IF EXISTS " ~ schema_name ~ " CASCADE") %}
     {% do adapter.commit() %}
 {% endmacro %}
+
+{% macro vertica__edr_drop_schema(database_name, schema_name) %}
+    {#- Vertica DDL is auto-committed; an explicit adapter.commit() would
+        fail with "no transaction in progress". -#}
+    {% do run_query("DROP SCHEMA IF EXISTS " ~ schema_name ~ " CASCADE") %}
+{% endmacro %}

integration_tests/dbt_project/macros/schema_utils/list_schemas.sql

@@ -54,3 +54,12 @@
     {% for row in results %} {% do schemas.append(row[0]) %} {% endfor %}
     {% do return(schemas) %}
 {% endmacro %}
+
+{% macro vertica__edr_list_schemas(database) %}
+    {#- Vertica's v_catalog.schemata is scoped to the current database and
+        does not have a database_name filter column. -#}
+    {% set results = run_query("SELECT schema_name FROM v_catalog.schemata") %}
+    {% set schemas = [] %}
+    {% for row in results %} {% do schemas.append(row[0]) %} {% endfor %}
+    {% do return(schemas) %}
+{% endmacro %}

integration_tests/dbt_project/macros/schema_utils/schema_exists.sql

@@ -64,3 +64,14 @@
     {% set result = run_query("SHOW DATABASES LIKE '" ~ safe_schema ~ "'") %}
     {% do return(result | length > 0) %}
 {% endmacro %}
+
+{% macro vertica__edr_schema_exists(database, schema_name) %}
+    {#- Vertica's v_catalog.schemata is scoped to the current database. -#}
+    {% set safe_schema = schema_name | replace("'", "''") %}
+    {% set result = run_query(
+        "SELECT schema_name FROM v_catalog.schemata WHERE lower(schema_name) = lower('"
+        ~ safe_schema
+        ~ "')"
+    ) %}
+    {% do return(result | length > 0) %}
+{% endmacro %}

integration_tests/dbt_project/macros/vertica_seed_override.sql

@@ -0,0 +1,23 @@
+{#- Override the dbt-vertica seed helper so that each seed file uses a
+    unique reject-table name.  The upstream macro hardcodes
+    ``seed_rejects`` for every seed, which causes "Object already exists"
+    errors when ``dbt seed`` processes more than one file. -#}
+{% macro copy_local_load_csv_rows(model, agate_table) %}
+    {% set cols_sql = get_seed_column_quoted_csv(model, agate_table.column_names) %}
+
+    {#- Build a per-seed reject table name so concurrent seeds don't clash. -#}
+    {% set reject_table = model["alias"] ~ "_rejects" %}
+
+    {% set sql %}
+        copy {{ this.render() }}
+        ({{ cols_sql }})
+        from local '{{ agate_table.original_abspath }}'
+        delimiter ','
+        enclosed by '"'
+        skip 1
+        abort on error
+        rejected data as table {{ this.without_identifier() }}.{{ reject_table }};
+    {% endset %}
+
+    {{ return(sql) }}
+{% endmacro %}

integration_tests/docker-compose-vertica.yml

@@ -0,0 +1,40 @@
+services:
+  vertica:
+    environment:
+      VERTICA_USER: dbadmin
+      VERTICA_PASS: vertica
+      VERTICA_HOST: localhost
+      VERTICA_PORT: 5433
+      VERTICA_DATABASE: elementary_tests
+      VERTICA_SCHEMA: ${SCHEMA_NAME}
+      APP_DB_USER: dbadmin
+      APP_DB_PASSWORD: vertica
+      TZ: "America/Los_Angeles"
+      VERTICA_DB_NAME: elementary_tests
+      VMART_ETL_SCRIPT: ""
+    container_name: vertica
+    image: ghcr.io/ratiopbc/vertica-ce
+    ports:
+      - "5433:5433"
+      - "5444:5444"
+    deploy:
+      mode: global
+    ulimits:
+      nofile:
+        soft: 65536
+        hard: 65536
+    volumes:
+      - type: volume
+        source: vertica-data
+        target: /data
+    healthcheck:
+      test:
+        [
+          "CMD-SHELL",
+          "/opt/vertica/bin/vsql -U dbadmin -w vertica -c 'SELECT 1;'",
+        ]
+      interval: 5s
+      timeout: 5s
+      retries: 10
+volumes:
+  vertica-data:

integration_tests/profiles/profiles.yml.j2

@@ -75,6 +75,18 @@ elementary_tests:
       trust_cert: true
       threads: 4
 
+    vertica: &vertica
+      type: vertica
+      host: localhost
+      port: 5433
+      username: dbadmin
+      password: vertica
+      database: elementary_tests
+      schema: {{ schema_name }}
+      connection_load_balance: false
+      retries: 2
+      threads: 4
+
     # ── Cloud targets (secrets substituted at CI time) ─────────────────
 
     snowflake: &snowflake
@@ -150,7 +162,7 @@ elementary_tests:
 elementary:
   target: postgres
   outputs:
-{%- set targets = ['postgres', 'clickhouse', 'trino', 'dremio', 'spark', 'duckdb', 'sqlserver', 'snowflake', 'bigquery', 'redshift', 'databricks_catalog', 'athena', 'fabric'] %}
+{%- set targets = ['postgres', 'clickhouse', 'trino', 'dremio', 'spark', 'duckdb', 'sqlserver', 'vertica', 'snowflake', 'bigquery', 'redshift', 'databricks_catalog', 'athena', 'fabric'] %}
 {%- for t in targets %}
     {{ t }}:
       <<: *{{ t }}

integration_tests/tests/data_seeder.py

@@ -4,7 +4,7 @@
 from contextlib import contextmanager
 from pathlib import Path
 from types import MappingProxyType
-from typing import TYPE_CHECKING, ClassVar, Dict, Generator, List, Mapping
+from typing import TYPE_CHECKING, ClassVar, Dict, Generator, List, Mapping, Optional
 
 from elementary.clients.dbt.base_dbt_runner import BaseDbtRunner
 from logger import get_logger
@@ -121,7 +121,7 @@ class BaseSqlInsertSeeder(ABC):
 
     def __init__(
         self,
-        query_runner: "AdapterQueryRunner",
+        query_runner: Optional["AdapterQueryRunner"],
         schema: str,
         seeds_dir_path: Path,
     ) -> None:
@@ -454,3 +454,119 @@ def _create_table_sql(self, fq_table: str, col_defs: str) -> str:
             f"CREATE TABLE {fq_table} ({col_defs}) "
             f"ENGINE = MergeTree() ORDER BY tuple()"
         )
+
+
+class VerticaDirectSeeder(BaseSqlInsertSeeder):
+    """Fast seeder for Vertica: executes CREATE TABLE + INSERT directly.
+
+    Bypasses ``dbt seed`` (which uses Vertica's COPY command) because COPY
+    rejects empty CSV fields for non-string columns instead of treating them
+    as NULL.  Direct INSERT statements handle NULL correctly.
+
+    Uses a *direct* ``vertica_python`` connection (rather than dbt's adapter
+    connection pool) so that all DDL + DML runs in a single session and can
+    be committed atomically.  dbt's ``connection_named`` context manager
+    releases (and effectively rolls back) the connection after each
+    ``execute_sql`` call, which caused INSERT data to be invisible to
+    subsequent ``dbt test`` sessions.
+
+    Vertica uses double-quote identifiers (not backticks), so this class
+    overrides the ``seed`` method to use ``"col"`` quoting.
+    """
+
+    def _type_string(self) -> str:
+        # Must match edr_type_string (varchar(16000)) so that schema-change
+        # detection sees a consistent type between seeded tables and
+        # elementary metadata columns.
+        return "VARCHAR(16000)"
+
+    def _type_boolean(self) -> str:
+        return "BOOLEAN"
+
+    def _type_integer(self) -> str:
+        return "INTEGER"
+
+    def _type_float(self) -> str:
+        return "FLOAT"
+
+    def _format_value(self, value: object, col_type: str) -> str:
+        if value is None or (isinstance(value, str) and value == ""):
+            return "NULL"
+        if isinstance(value, bool):
+            return "true" if value else "false"
+        if isinstance(value, (int, float)):
+            return str(value)
+        text = str(value)
+        text = text.replace("'", "''")
+        return f"'{text}'"
+
+    def _create_table_sql(self, fq_table: str, col_defs: str) -> str:
+        return f"CREATE TABLE {fq_table} ({col_defs})"
+
+    @staticmethod
+    def _vertica_connection():
+        """Open a direct vertica_python connection from env / defaults."""
+        import vertica_python  # available in the test venv
+
+        conn_info = {
+            "host": os.environ.get("VERTICA_HOST", "localhost"),
+            "port": int(os.environ.get("VERTICA_PORT", "5433")),
+            "user": os.environ.get("VERTICA_USER", "dbadmin"),
+            "password": os.environ.get("VERTICA_PASSWORD", "vertica"),
+            "database": os.environ.get("VERTICA_DATABASE", "elementary_tests"),
+        }
+        return vertica_python.connect(**conn_info)
+
+    @contextmanager
+    def seed(self, data: List[dict], table_name: str) -> Generator[None, None, None]:
+        """Override base seed to use double-quote identifiers for Vertica."""
+        if not data:
+            raise ValueError(f"Seed data for '{table_name}' must not be empty")
+        columns = list(data[0].keys())
+        col_types: Dict[str, str] = {
+            col: self._infer_column_type([row.get(col) for row in data])
+            for col in columns
+        }
+        # Vertica uses double-quote identifiers, not backticks.
+        col_defs = ", ".join(f'"{col}" {col_types[col]}' for col in columns)
+        fq_table = f'"{self._schema}"."{table_name}"'
+
+        seed_path = self._write_csv(data, table_name)
+
+        try:
+            # Use a direct connection so DDL + DML share the same session
+            # and the COMMIT is guaranteed to persist the data.
+            conn = self._vertica_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute(f"DROP TABLE IF EXISTS {fq_table}")
+                cur.execute(self._create_table_sql(fq_table, col_defs))
+
+                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._format_value(row.get(c), col_types[c])
+                            for c in columns
+                        )
+                        + ")"
+                        for row in batch
+                    )
+                    cur.execute(f"INSERT INTO {fq_table} VALUES {rows_sql}")
+
+                conn.commit()
+            finally:
+                conn.close()
+
+            logger.info(
+                "%s: loaded %d rows into %s (%s)",
+                type(self).__name__,
+                len(data),
+                fq_table,
+                ", ".join(f"{c}: {t}" for c, t in col_types.items()),
+            )
+
+            yield
+        finally:
+            seed_path.unlink(missing_ok=True)