fix: address CodeRabbit review round 3 - cleanup and hardening

- test-warehouse.yml: replace for-loop with case statement for Docker adapter check
- dateadd.sql: use bare TIMESTAMPADD keywords instead of SQL_TSI_* constants, add case-insensitive datepart matching
- spark.py: harden connection cleanup with None-init + conditional close, escape single quotes in container_path
- dremio.py: switch from PyYAML to ruamel.yaml for project consistency, log non-file parsing failures, make seeding idempotent with DROP TABLE before CREATE TABLE

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

Author: devin-ai-integration[bot]

.github/workflows/test-warehouse.yml

@@ -186,14 +186,10 @@ jobs:
           # Docker-based adapters use ephemeral containers, so a fixed schema
           # name is safe (the concurrency group prevents parallel collisions).
           # This enables caching the seeded database state between runs.
-          DOCKER_ADAPTERS="postgres clickhouse trino dremio duckdb spark"
           IS_DOCKER=false
-          for adapter in $DOCKER_ADAPTERS; do
-            if [ "$adapter" = "${{ inputs.warehouse-type }}" ]; then
-              IS_DOCKER=true
-              break
-            fi
-          done
+          case "${{ inputs.warehouse-type }}" in
+            postgres|clickhouse|trino|dremio|duckdb|spark) IS_DOCKER=true ;;
+          esac
 
           if [ "$IS_DOCKER" = "true" ]; then
             SCHEMA_NAME="elementary_tests"

elementary/monitor/dbt_project/macros/overrides/dateadd.sql

@@ -10,27 +10,28 @@
 #}
 
 {% macro dremio__dateadd(datepart, interval, from_date_or_timestamp) %}
+    {% set datepart = datepart | lower %}
     {% set interval = interval | string %}
     {# dbt-dremio's original macro wraps the result in a scalar subquery
        ("select TIMESTAMPADD(...) order by 1"), so when we receive the
        interval from upstream it may carry a trailing "order by 1". #}
     {% set interval = interval.replace('order by 1', '') %}
     {% if datepart == 'year' %}
-        TIMESTAMPADD(SQL_TSI_YEAR, CAST({{interval}} as int), CAST({{from_date_or_timestamp}} as TIMESTAMP))
+        TIMESTAMPADD(YEAR, CAST({{interval}} as int), CAST({{from_date_or_timestamp}} as TIMESTAMP))
     {% elif datepart == 'quarter' %}
-        TIMESTAMPADD(SQL_TSI_QUARTER, CAST({{interval}} as int), CAST({{from_date_or_timestamp}} as TIMESTAMP))
+        TIMESTAMPADD(QUARTER, CAST({{interval}} as int), CAST({{from_date_or_timestamp}} as TIMESTAMP))
     {% elif datepart == 'month' %}
-        TIMESTAMPADD(SQL_TSI_MONTH, CAST({{interval}} as int), CAST({{from_date_or_timestamp}} as TIMESTAMP))
+        TIMESTAMPADD(MONTH, CAST({{interval}} as int), CAST({{from_date_or_timestamp}} as TIMESTAMP))
     {% elif datepart == 'week' %}
-        TIMESTAMPADD(SQL_TSI_WEEK, CAST({{interval}} as int), CAST({{from_date_or_timestamp}} as TIMESTAMP))
+        TIMESTAMPADD(WEEK, CAST({{interval}} as int), CAST({{from_date_or_timestamp}} as TIMESTAMP))
     {% elif datepart == 'hour' %}
-        TIMESTAMPADD(SQL_TSI_HOUR, CAST({{interval}} as int), CAST({{from_date_or_timestamp}} as TIMESTAMP))
+        TIMESTAMPADD(HOUR, CAST({{interval}} as int), CAST({{from_date_or_timestamp}} as TIMESTAMP))
     {% elif datepart == 'minute' %}
-        TIMESTAMPADD(SQL_TSI_MINUTE, CAST({{interval}} as int), CAST({{from_date_or_timestamp}} as TIMESTAMP))
+        TIMESTAMPADD(MINUTE, CAST({{interval}} as int), CAST({{from_date_or_timestamp}} as TIMESTAMP))
     {% elif datepart == 'second' %}
-        TIMESTAMPADD(SQL_TSI_SECOND, CAST({{interval}} as int), CAST({{from_date_or_timestamp}} as TIMESTAMP))
+        TIMESTAMPADD(SECOND, CAST({{interval}} as int), CAST({{from_date_or_timestamp}} as TIMESTAMP))
     {% elif datepart == 'day' %}
-        TIMESTAMPADD(SQL_TSI_DAY, CAST({{interval}} as int), CAST({{from_date_or_timestamp}} as TIMESTAMP))
+        TIMESTAMPADD(DAY, CAST({{interval}} as int), CAST({{from_date_or_timestamp}} as TIMESTAMP))
     {% else %}
         {{ exceptions.raise_compiler_error("dremio__dateadd: unrecognized datepart '" ~ datepart ~ "'. Supported: year, quarter, month, week, day, hour, minute, second.") }}
     {% endif %}

tests/e2e_dbt_project/external_seeders/dremio.py

@@ -6,7 +6,8 @@
 import re
 import time
 
-import yaml
+from ruamel.yaml import YAML
+
 from external_seeders.base import ExternalSeeder
 
 
@@ -21,8 +22,9 @@ def _docker_defaults() -> dict[str, str]:
     # --- docker-compose.yml: MinIO credentials ---
     compose_path = os.path.join(project_dir, "docker-compose.yml")
     try:
+        _yaml = YAML()
         with open(compose_path) as fh:
-            cfg = yaml.safe_load(fh)
+            cfg = _yaml.load(fh)
         services = cfg.get("services", {})
         for item in services.get("dremio-minio", {}).get("environment", []):
             if isinstance(item, str) and "=" in item:
@@ -35,8 +37,10 @@ def _docker_defaults() -> dict[str, str]:
                     defaults["MINIO_ACCESS_KEY"] = v
                 elif k == "MINIO_ROOT_PASSWORD":
                     defaults["MINIO_SECRET_KEY"] = v
-    except Exception:
+    except FileNotFoundError:
         pass
+    except Exception as e:
+        print(f"  Warning: failed parsing docker defaults from {compose_path}: {e}")
 
     # --- dremio-setup.sh: Dremio login credentials ---
     # Extract default values from bash variable assignments like:
@@ -51,8 +55,10 @@ def _docker_defaults() -> dict[str, str]:
         m = re.search(r'DREMIO_USER="\$\{DREMIO_USER:-([^}]+)\}"', content)
         if m:
             defaults["DREMIO_USER"] = m.group(1)
-    except Exception:
+    except FileNotFoundError:
         pass
+    except Exception as e:
+        print(f"  Warning: failed parsing dremio defaults from {setup_path}: {e}")
 
     return defaults
 
@@ -364,9 +370,13 @@ def load(self) -> None:
 
             fqn = f"{nessie_ns}.{qi(table_name)}"
 
-            # Create empty Iceberg table with VARCHAR columns
+            # Drop + recreate to ensure idempotent seeding (no stale data)
             col_defs = ", ".join(f"{qi(c)} VARCHAR" for c in cols)
-            create_sql = f"CREATE TABLE IF NOT EXISTS {fqn} ({col_defs})"
+            try:
+                self._sql(token, f"DROP TABLE IF EXISTS {fqn}", timeout=30)
+            except Exception:
+                pass  # Table may not exist yet
+            create_sql = f"CREATE TABLE {fqn} ({col_defs})"
             try:
                 self._sql(token, create_sql, timeout=60)
                 created_tables.append(table_name)

tests/e2e_dbt_project/external_seeders/spark.py

@@ -40,9 +40,11 @@ def load(self) -> None:
         port = int(os.environ.get("SPARK_PORT", "10000"))
 
         print(f"Connecting to Spark Thrift at {host}:{port}...")
-        conn = hive.Connection(host=host, port=port, username="dbt")
-        cursor = conn.cursor()
+        conn = None
+        cursor = None
         try:
+            conn = hive.Connection(host=host, port=port, username="dbt")
+            cursor = conn.cursor()
             print(f"Creating schema '{seed_schema}'...")
             cursor.execute(f"CREATE DATABASE IF NOT EXISTS `{seed_schema}`")
 
@@ -69,13 +71,15 @@ def load(self) -> None:
                     continue
 
                 container_path = f"/seed-data/{subdir}/{fname}"
+                # Escape single quotes in path to prevent SQL injection
+                safe_path = container_path.replace("'", "''")
                 tmp_view = f"_tmp_csv_{table_name}"
                 print(f"  Loading: {table_name}")
                 try:
                     cursor.execute(
                         f"CREATE OR REPLACE TEMPORARY VIEW {q(tmp_view)} "
                         f"USING csv "
-                        f"OPTIONS (path '{container_path}', header 'true', "
+                        f"OPTIONS (path '{safe_path}', header 'true', "
                         f"inferSchema 'true')"
                     )
                     cursor.execute(
@@ -88,8 +92,10 @@ def load(self) -> None:
                 except Exception as e:
                     failures.append(f"{table_name}: {e}")
         finally:
-            cursor.close()
-            conn.close()
+            if cursor is not None:
+                cursor.close()
+            if conn is not None:
+                conn.close()
         if failures:
             raise RuntimeError(
                 "Spark seed loading failed:\n - " + "\n - ".join(failures)