fastflowtransform.executors

BigQueryBFExecutor

Bases: BigQueryIdentifierMixin, BaseExecutor[DataFrame]

Source code in src/fastflowtransform/executors/bigquery_bf_exec.py

class BigQueryBFExecutor(BigQueryIdentifierMixin, BaseExecutor[BFDataFrame]):
    ENGINE_NAME = "bigquery_batch"

    def __init__(self, project: str, dataset: str, location: str | None = None):
        self.project = project
        self.dataset = dataset
        self.location = location
        self.client = bigquery.Client(project=project, location=location)

        try:
            ctx = BigQueryOptions(
                project=project,
                # default_dataset=dataset,
                location=location,
            )
            self.session = bigframes.Session(context=ctx)
        except Exception:
            # Fallback: session without explicit context (ADC/default project),
            # though you typically use fully qualified table IDs anyway.
            self.session = bigframes.Session()

        self.con = BigQueryConnShim(self.client, location=self.location)

    def _read_relation(self, relation: str, node: Node, deps: Iterable[str]) -> BFDataFrame:
        table_id = f"{self.project}.{self.dataset}.{relation}"
        try:
            return self.session.read_gbq(table_id)
        except NotFound as e:
            existing = [
                t.table_id for t in self.client.list_tables(f"{self.project}.{self.dataset}")
            ]
            raise RuntimeError(
                f"Dependency table not found: {table_id}\n"
                f"Deps: {list(deps)}\nExisting in dataset: {existing}\n"
                "Hinweis: Seeds/Upstream-Modelle erzeugt? DATASET korrekt?"
            ) from e

    def _materialize_relation(self, relation: str, df: BFDataFrame, node: Node) -> None:
        table_id = f"{self.project}.{self.dataset}.{relation}"

        to_gbq = getattr(df, "to_gbq", None)
        if callable(to_gbq):
            to_gbq(table_id, if_exists="replace")
            return

        # Fallback only when it is truly a method (not a column name!)
        mat = getattr(df, "materialize", None)
        if callable(mat):
            mat(table=table_id, mode="overwrite")
            return

        raise RuntimeError(
            "BigQuery DataFrames: Ergebnis nicht materialisierbar. "
            "Erwarte df.to_gbq(...) oder df.materialize(...)."
        )

    # ---- Meta hook ----
    def on_node_built(self, node: Node, relation: str, fingerprint: str) -> None:
        """Mirror DuckDB/PG: write/update _ff_meta after successful build."""
        try:
            ensure_meta_table(self)
            upsert_meta(self, node.name, relation, fingerprint, "bigquery")
        except Exception:
            # Best-effort: meta must not break the run
            pass

    def _validate_required(
        self, node_name: str, inputs: Any, requires: dict[str, set[str]]
    ) -> None:
        if not requires:
            return

        def cols(bf_df: BFDataFrame) -> set[str]:
            if hasattr(bf_df, "columns"):
                return set(map(str, list(bf_df.columns)))
            if hasattr(bf_df, "schema") and hasattr(bf_df.schema, "names"):
                return set(bf_df.schema.names)
            return set()

        errs: list[str] = []
        if self._is_frame(inputs):
            need = next(iter(requires.values()), set())
            miss = need - cols(inputs)
            if miss:
                errs.append(f"- missing columns: {sorted(miss)}")
        else:
            for rel, need in requires.items():
                if rel not in inputs:
                    errs.append(f"- missing dependency key '{rel}'")
                    continue
                miss = need - cols(inputs[rel])
                if miss:
                    errs.append(f"- [{rel}] missing: {sorted(miss)}")
        if errs:
            raise ValueError(
                f"Required columns check failed for BigQuery DataFrames model '{node_name}'.\n"
                + "\n".join(errs)
            )

    def _columns_of(self, frame: BFDataFrame) -> list[str]:
        if hasattr(frame, "columns"):
            return [str(c) for c in list(frame.columns)]
        if hasattr(frame, "schema") and hasattr(frame.schema, "names"):
            return list(frame.schema.names)
        return []

    def _is_frame(self, obj: Any) -> bool:
        return bool(obj) and (
            callable(getattr(obj, "to_gbq", None))
            or callable(getattr(obj, "materialize", None))
            or hasattr(obj, "columns")
        )

    def _frame_name(self) -> str:
        return "BigQuery DataFrame (BigFrames)"

    # ---- Helpers ----
    # ---- SQL hooks ----
    def _format_relation_for_ref(self, name: str) -> str:
        return self._qualified_identifier(relation_for(name))

    def _format_source_reference(
        self, cfg: dict[str, Any], source_name: str, table_name: str
    ) -> str:
        if cfg.get("location"):
            raise NotImplementedError("BigQuery executor does not support path-based sources.")

        ident = cfg.get("identifier")
        if not ident:
            raise KeyError(f"Source {source_name}.{table_name} missing identifier")

        proj = cfg.get("project") or cfg.get("database") or cfg.get("catalog") or self.project
        dset = cfg.get("dataset") or cfg.get("schema") or self.dataset
        return self._qualified_identifier(ident, project=proj, dataset=dset)

    def _apply_sql_materialization(
        self, node: Node, target_sql: str, select_body: str, materialization: str
    ) -> None:
        self._ensure_dataset()
        try:
            super()._apply_sql_materialization(node, target_sql, select_body, materialization)
        except BadRequest as e:
            raise RuntimeError(
                f"BigQuery SQL failed for {target_sql}:\n{select_body}\n\n{e}"
            ) from e

    def _create_or_replace_view(self, target_sql: str, select_body: str, node: Node) -> None:
        self.client.query(
            f"CREATE OR REPLACE VIEW {target_sql} AS {select_body}",
            location=self.location,
        ).result()

    def _create_or_replace_table(self, target_sql: str, select_body: str, node: Node) -> None:
        self.client.query(
            f"CREATE OR REPLACE TABLE {target_sql} AS {select_body}",
            location=self.location,
        ).result()

    def _create_or_replace_view_from_table(
        self, view_name: str, backing_table: str, node: Node
    ) -> None:
        view_id = self._qualified_identifier(view_name)
        back_id = self._qualified_identifier(backing_table)
        self.client.query(
            f"CREATE OR REPLACE VIEW {view_id} AS SELECT * FROM {back_id}",
            location=self.location,
        ).result()

    # ── Incremental API (feature parity with DuckDB/PG) ──────────────────
    def exists_relation(self, relation: str) -> bool:
        """Check presence in TABLES or VIEWS information schema."""
        proj = self.project
        dset = self.dataset
        rel = relation
        q = f"""
        SELECT 1
        FROM `{proj}.{dset}.INFORMATION_SCHEMA.TABLES`
        WHERE LOWER(table_name)=LOWER(@rel)
        UNION ALL
        SELECT 1
        FROM `{proj}.{dset}.INFORMATION_SCHEMA.VIEWS`
        WHERE LOWER(table_name)=LOWER(@rel)
        LIMIT 1
        """
        job = self.client.query(
            q,
            job_config=bigquery.QueryJobConfig(
                query_parameters=[bigquery.ScalarQueryParameter("rel", "STRING", rel)]
            ),
            location=self.location,
        )
        return bool(list(job.result()))

    def create_table_as(self, relation: str, select_sql: str) -> None:
        """CTAS with cleaned SELECT body (no trailing semicolons)."""
        self._ensure_dataset()
        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        target = self._qualified_identifier(relation, project=self.project, dataset=self.dataset)
        self.client.query(
            f"CREATE TABLE {target} AS {body}",
            location=self.location,
        ).result()

    def incremental_insert(self, relation: str, select_sql: str) -> None:
        """INSERT INTO with cleaned SELECT body."""
        self._ensure_dataset()
        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        target = self._qualified_identifier(relation, project=self.project, dataset=self.dataset)
        self.client.query(
            f"INSERT INTO {target} {body}",
            location=self.location,
        ).result()

    def incremental_merge(self, relation: str, select_sql: str, unique_key: list[str]) -> None:
        """
        Portable fallback in BigQuery (without full MERGE):
          - DELETE collisions via WHERE EXISTS against the cleaned SELECT body
          - INSERT all rows from the body
        Executed as two statements to keep error surfaces clean.
        """
        self._ensure_dataset()
        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        target = self._qualified_identifier(relation, project=self.project, dataset=self.dataset)
        pred = " AND ".join([f"t.{k}=s.{k}" for k in unique_key]) or "FALSE"

        # DELETE … WHERE EXISTS (SELECT 1 FROM (body) s WHERE pred)
        delete_sql = f"""
        DELETE FROM {target} t
        WHERE EXISTS (SELECT 1 FROM ({body}) s WHERE {pred})
        """
        self.client.query(delete_sql, location=self.location).result()

        # INSERT new rows
        insert_sql = f"INSERT INTO {target} SELECT * FROM ({body})"
        self.client.query(insert_sql, location=self.location).result()

    def alter_table_sync_schema(
        self, relation: str, select_sql: str, *, mode: str = "append_new_columns"
    ) -> None:
        """
        Best-effort additive schema sync:
          - infer select schema via dry-run (schema on QueryJob)
          - add missing columns as NULLABLE with inferred type
        """
        if mode not in {"append_new_columns", "sync_all_columns"}:
            return

        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        # Infer target schema from the query (no data read)
        probe_job = self.client.query(
            f"SELECT * FROM ({body}) WHERE 1=0",
            job_config=bigquery.QueryJobConfig(dry_run=False, use_query_cache=False),
            location=self.location,
        )
        probe_job.result()
        select_fields = {f.name: f for f in (probe_job.schema or [])}

        # Existing table schema
        table_ref = f"{self.project}.{self.dataset}.{relation}"
        try:
            tbl = self.client.get_table(table_ref)
        except NotFound:
            return
        existing_cols = {f.name for f in (tbl.schema or [])}

        to_add = [name for name in select_fields if name not in existing_cols]
        if not to_add:
            return

        target = self._qualified_identifier(relation, project=self.project, dataset=self.dataset)
        for col in to_add:
            bf = select_fields[col]
            # Use BigQuery standard SQL type string (e.g., STRING, INT64, BOOL, FLOAT64, …)
            typ = str(bf.field_type) if hasattr(bf, "field_type") else "STRING"
            # Nullable by default
            self.client.query(
                f"ALTER TABLE {target} ADD COLUMN {col} {typ}",
                location=self.location,
            ).result()

run_sql

run_sql(node, env)

Orchestrate SQL models

1) Render Jinja (ref/source/this) and strip leading {{ config(...) }}. 2) If the SQL is full DDL (CREATE …), execute it verbatim (passthrough). 3) Otherwise, normalize to CREATE OR REPLACE {TABLE|VIEW} AS . The body is CTE-aware (keeps WITH … SELECT … intact).

On failure, raise ModelExecutionError with a helpful snippet.

Source code in src/fastflowtransform/executors/base.py

def run_sql(self, node: Node, env: Environment) -> None:
    """
    Orchestrate SQL models:
      1) Render Jinja (ref/source/this) and strip leading {{ config(...) }}.
      2) If the SQL is full DDL (CREATE …), execute it verbatim (passthrough).
      3) Otherwise, normalize to CREATE OR REPLACE {TABLE|VIEW} AS <body>.
         The body is CTE-aware (keeps WITH … SELECT … intact).
    On failure, raise ModelExecutionError with a helpful snippet.
    """
    sql_rendered = self.render_sql(
        node,
        env,
        ref_resolver=lambda name: self._resolve_ref(name, env),
        source_resolver=self._resolve_source,
    )
    sql = self._strip_leading_config(sql_rendered).strip()

    materialization = (node.meta or {}).get("materialized", "table")
    if materialization == "ephemeral":
        return

    # 1) Direct DDL passthrough (CREATE [OR REPLACE] {TABLE|VIEW} …)
    if self._looks_like_direct_ddl(sql):
        try:
            self._execute_sql_direct(sql, node)
            return
        except NotImplementedError:
            # Engine doesn't implement direct DDL → fall back to normalized materialization.
            pass
        except Exception as e:
            raise ModelExecutionError(
                node_name=node.name,
                relation=relation_for(node.name),
                message=str(e),
                sql_snippet=sql,
            ) from e

    # 2) Normalized materialization path (CTE-safe body)
    body = self._selectable_body(sql).rstrip(" ;\n\t")
    target_sql = self._format_relation_for_ref(node.name)

    # Centralized SQL preview logging (applies to ALL engines)
    preview = (
        f"=== MATERIALIZE ===\n"
        f"-- model: {node.name}\n"
        f"-- materialized: {materialization}\n"
        f"-- target: {target_sql}\n"
        f"{body}\n"
    )
    echo_debug(preview)

    try:
        self._apply_sql_materialization(node, target_sql, body, materialization)
    except Exception as e:
        preview = f"-- materialized={materialization}\n-- target={target_sql}\n{body}"
        raise ModelExecutionError(
            node_name=node.name,
            relation=relation_for(node.name),
            message=str(e),
            sql_snippet=preview,
        ) from e

on_node_built

on_node_built(node, relation, fingerprint)

Mirror DuckDB/PG: write/update _ff_meta after successful build.

Source code in src/fastflowtransform/executors/bigquery_bf_exec.py

def on_node_built(self, node: Node, relation: str, fingerprint: str) -> None:
    """Mirror DuckDB/PG: write/update _ff_meta after successful build."""
    try:
        ensure_meta_table(self)
        upsert_meta(self, node.name, relation, fingerprint, "bigquery")
    except Exception:
        # Best-effort: meta must not break the run
        pass

exists_relation

exists_relation(relation)

Check presence in TABLES or VIEWS information schema.

Source code in src/fastflowtransform/executors/bigquery_bf_exec.py

def exists_relation(self, relation: str) -> bool:
    """Check presence in TABLES or VIEWS information schema."""
    proj = self.project
    dset = self.dataset
    rel = relation
    q = f"""
    SELECT 1
    FROM `{proj}.{dset}.INFORMATION_SCHEMA.TABLES`
    WHERE LOWER(table_name)=LOWER(@rel)
    UNION ALL
    SELECT 1
    FROM `{proj}.{dset}.INFORMATION_SCHEMA.VIEWS`
    WHERE LOWER(table_name)=LOWER(@rel)
    LIMIT 1
    """
    job = self.client.query(
        q,
        job_config=bigquery.QueryJobConfig(
            query_parameters=[bigquery.ScalarQueryParameter("rel", "STRING", rel)]
        ),
        location=self.location,
    )
    return bool(list(job.result()))

create_table_as

create_table_as(relation, select_sql)

CTAS with cleaned SELECT body (no trailing semicolons).

Source code in src/fastflowtransform/executors/bigquery_bf_exec.py

def create_table_as(self, relation: str, select_sql: str) -> None:
    """CTAS with cleaned SELECT body (no trailing semicolons)."""
    self._ensure_dataset()
    body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
    target = self._qualified_identifier(relation, project=self.project, dataset=self.dataset)
    self.client.query(
        f"CREATE TABLE {target} AS {body}",
        location=self.location,
    ).result()

incremental_insert

incremental_insert(relation, select_sql)

INSERT INTO with cleaned SELECT body.

Source code in src/fastflowtransform/executors/bigquery_bf_exec.py

def incremental_insert(self, relation: str, select_sql: str) -> None:
    """INSERT INTO with cleaned SELECT body."""
    self._ensure_dataset()
    body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
    target = self._qualified_identifier(relation, project=self.project, dataset=self.dataset)
    self.client.query(
        f"INSERT INTO {target} {body}",
        location=self.location,
    ).result()

incremental_merge

incremental_merge(relation, select_sql, unique_key)

Portable fallback in BigQuery (without full MERGE): - DELETE collisions via WHERE EXISTS against the cleaned SELECT body - INSERT all rows from the body Executed as two statements to keep error surfaces clean.

Source code in src/fastflowtransform/executors/bigquery_bf_exec.py

def incremental_merge(self, relation: str, select_sql: str, unique_key: list[str]) -> None:
    """
    Portable fallback in BigQuery (without full MERGE):
      - DELETE collisions via WHERE EXISTS against the cleaned SELECT body
      - INSERT all rows from the body
    Executed as two statements to keep error surfaces clean.
    """
    self._ensure_dataset()
    body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
    target = self._qualified_identifier(relation, project=self.project, dataset=self.dataset)
    pred = " AND ".join([f"t.{k}=s.{k}" for k in unique_key]) or "FALSE"

    # DELETE … WHERE EXISTS (SELECT 1 FROM (body) s WHERE pred)
    delete_sql = f"""
    DELETE FROM {target} t
    WHERE EXISTS (SELECT 1 FROM ({body}) s WHERE {pred})
    """
    self.client.query(delete_sql, location=self.location).result()

    # INSERT new rows
    insert_sql = f"INSERT INTO {target} SELECT * FROM ({body})"
    self.client.query(insert_sql, location=self.location).result()

alter_table_sync_schema

alter_table_sync_schema(relation, select_sql, *, mode='append_new_columns')

Best-effort additive schema sync

• infer select schema via dry-run (schema on QueryJob)
• add missing columns as NULLABLE with inferred type

Source code in src/fastflowtransform/executors/bigquery_bf_exec.py

def alter_table_sync_schema(
    self, relation: str, select_sql: str, *, mode: str = "append_new_columns"
) -> None:
    """
    Best-effort additive schema sync:
      - infer select schema via dry-run (schema on QueryJob)
      - add missing columns as NULLABLE with inferred type
    """
    if mode not in {"append_new_columns", "sync_all_columns"}:
        return

    body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
    # Infer target schema from the query (no data read)
    probe_job = self.client.query(
        f"SELECT * FROM ({body}) WHERE 1=0",
        job_config=bigquery.QueryJobConfig(dry_run=False, use_query_cache=False),
        location=self.location,
    )
    probe_job.result()
    select_fields = {f.name: f for f in (probe_job.schema or [])}

    # Existing table schema
    table_ref = f"{self.project}.{self.dataset}.{relation}"
    try:
        tbl = self.client.get_table(table_ref)
    except NotFound:
        return
    existing_cols = {f.name for f in (tbl.schema or [])}

    to_add = [name for name in select_fields if name not in existing_cols]
    if not to_add:
        return

    target = self._qualified_identifier(relation, project=self.project, dataset=self.dataset)
    for col in to_add:
        bf = select_fields[col]
        # Use BigQuery standard SQL type string (e.g., STRING, INT64, BOOL, FLOAT64, …)
        typ = str(bf.field_type) if hasattr(bf, "field_type") else "STRING"
        # Nullable by default
        self.client.query(
            f"ALTER TABLE {target} ADD COLUMN {col} {typ}",
            location=self.location,
        ).result()

BigQueryExecutor

Bases: BigQueryIdentifierMixin, BaseExecutor[DataFrame]

Source code in src/fastflowtransform/executors/bigquery_exec.py

class BigQueryExecutor(BigQueryIdentifierMixin, BaseExecutor[pd.DataFrame]):
    ENGINE_NAME = "bigquery"
    """
    BigQuery executor (pandas DataFrames).
    ENV/Profiles typically use:
      - FF_BQ_PROJECT
      - FF_BQ_DATASET
      - FF_BQ_LOCATION (optional)
    """

    def __init__(
        self,
        project: str,
        dataset: str,
        location: str | None = None,
        client: Client | None = None,
    ):
        self.project = project
        self.dataset = dataset
        self.location = location
        self.client: Client = client or bigquery.Client(
            project=self.project, location=self.location
        )
        # Testing-API: con.execute(...)
        self.con = BigQueryConnShim(
            self.client, location=self.location, project=self.project, dataset=self.dataset
        )

    # ---------- Helpers ----------
    # ---------- Python (Frames) ----------
    def _read_relation(self, relation: str, node: Node, deps: Iterable[str]) -> pd.DataFrame:
        q = f"SELECT * FROM {self._qualified_identifier(relation)}"
        try:
            job = self.client.query(q, location=self.location)
            return job.result().to_dataframe(create_bqstorage_client=True)
        except NotFound as e:
            # list existing tables to aid debugging
            tables = list(self.client.list_tables(f"{self.project}.{self.dataset}"))
            existing = [t.table_id for t in tables]
            raise RuntimeError(
                f"Dependency table not found: {self.project}.{self.dataset}.{relation}\n"
                f"Deps: {list(deps)}\nExisting in dataset: {existing}\n"
                "Hinweis: Seeds/Upstream-Modelle erzeugt? DATASET korrekt?"
            ) from e

    def _materialize_relation(self, relation: str, df: pd.DataFrame, node: Node) -> None:
        self._ensure_dataset()
        table_id = f"{self.project}.{self.dataset}.{relation}"
        job_config = LoadJobConfig(write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE)
        # Optionally extend dtype mapping here (NUMERIC/STRING etc.)
        try:
            job = self.client.load_table_from_dataframe(
                df, table_id, job_config=job_config, location=self.location
            )
            job.result()
        except BadRequest as e:
            raise RuntimeError(f"BigQuery write failed: {table_id}\n{e}") from e

    def _create_view_over_table(self, view_name: str, backing_table: str, node: Node) -> None:
        view_id = self._qualified_identifier(view_name)
        back_id = self._qualified_identifier(backing_table)
        self._ensure_dataset()
        job = self.client.query(
            f"CREATE OR REPLACE VIEW {view_id} AS SELECT * FROM {back_id}",
            location=self.location,
        )
        job.result()

    def _frame_name(self) -> str:
        return "pandas"

    # ---- SQL hooks ----
    def _format_relation_for_ref(self, name: str) -> str:
        return self._qualified_identifier(relation_for(name))

    def _format_source_reference(
        self, cfg: dict[str, Any], source_name: str, table_name: str
    ) -> str:
        if cfg.get("location"):
            raise NotImplementedError("BigQuery executor does not support path-based sources.")

        ident = cfg.get("identifier")
        if not ident:
            raise KeyError(f"Source {source_name}.{table_name} missing identifier")

        proj = cfg.get("project") or cfg.get("database") or cfg.get("catalog") or self.project
        dset = cfg.get("dataset") or cfg.get("schema") or self.dataset
        return self._qualified_identifier(ident, project=proj, dataset=dset)

    def _apply_sql_materialization(
        self, node: Node, target_sql: str, select_body: str, materialization: str
    ) -> None:
        self._ensure_dataset()
        try:
            super()._apply_sql_materialization(node, target_sql, select_body, materialization)
        except BadRequest as e:
            raise RuntimeError(
                f"BigQuery SQL failed for {target_sql}:\n{select_body}\n\n{e}"
            ) from e

    def _create_or_replace_view(self, target_sql: str, select_body: str, node: Node) -> None:
        job = self.client.query(
            f"CREATE OR REPLACE VIEW {target_sql} AS {select_body}",
            location=self.location,
        )
        job.result()

    def _create_or_replace_table(self, target_sql: str, select_body: str, node: Node) -> None:
        job = self.client.query(
            f"CREATE OR REPLACE TABLE {target_sql} AS {select_body}",
            location=self.location,
        )
        job.result()

    def _create_or_replace_view_from_table(
        self, view_name: str, backing_table: str, node: Node
    ) -> None:
        view_id = self._qualified_identifier(view_name)
        back_id = self._qualified_identifier(backing_table)
        self._ensure_dataset()
        job = self.client.query(
            f"CREATE OR REPLACE VIEW {view_id} AS SELECT * FROM {back_id}",
            location=self.location,
        )
        job.result()

    def on_node_built(self, node: Node, relation: str, fingerprint: str) -> None:
        """
        Write/update dataset._ff_meta after a successful build.
        """
        try:
            ensure_meta_table(self)
            upsert_meta(self, node.name, relation, fingerprint, "bigquery")
        except Exception:
            pass

    # ── Incremental API (parity with DuckDB/PG) ───────────────────────────
    def exists_relation(self, relation: str) -> bool:
        """
        Check presence in INFORMATION_SCHEMA for tables/views.
        """
        proj = self.project
        dset = self.dataset
        rel = relation
        q = f"""
        SELECT 1
        FROM `{proj}.{dset}.INFORMATION_SCHEMA.TABLES`
        WHERE LOWER(table_name)=LOWER(@rel)
        UNION ALL
        SELECT 1
        FROM `{proj}.{dset}.INFORMATION_SCHEMA.VIEWS`
        WHERE LOWER(table_name)=LOWER(@rel)
        LIMIT 1
        """
        job = self.client.query(
            q,
            job_config=bigquery.QueryJobConfig(
                query_parameters=[bigquery.ScalarQueryParameter("rel", "STRING", rel)]
            ),
            location=self.location,
        )
        return bool(list(job.result()))

    def create_table_as(self, relation: str, select_sql: str) -> None:
        """
        CREATE TABLE AS with cleaned SELECT body (no trailing semicolons).
        """
        self._ensure_dataset()
        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        target = self._qualified_identifier(relation, project=self.project, dataset=self.dataset)
        self.client.query(
            f"CREATE TABLE {target} AS {body}",
            location=self.location,
        ).result()

    def incremental_insert(self, relation: str, select_sql: str) -> None:
        """
        INSERT INTO with cleaned SELECT body.
        """
        self._ensure_dataset()
        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        target = self._qualified_identifier(relation, project=self.project, dataset=self.dataset)
        self.client.query(
            f"INSERT INTO {target} {body}",
            location=self.location,
        ).result()

    def incremental_merge(self, relation: str, select_sql: str, unique_key: list[str]) -> None:
        """
        Portable fallback without native MERGE:
          - DELETE collisions via WHERE EXISTS against the cleaned SELECT body
          - INSERT new rows from the same body
        """
        self._ensure_dataset()
        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        target = self._qualified_identifier(relation, project=self.project, dataset=self.dataset)
        pred = " AND ".join([f"t.{k}=s.{k}" for k in unique_key]) or "FALSE"

        delete_sql = f"""
        DELETE FROM {target} t
        WHERE EXISTS (SELECT 1 FROM ({body}) s WHERE {pred})
        """
        self.client.query(delete_sql, location=self.location).result()

        insert_sql = f"INSERT INTO {target} SELECT * FROM ({body})"
        self.client.query(insert_sql, location=self.location).result()

    def alter_table_sync_schema(
        self, relation: str, select_sql: str, *, mode: str = "append_new_columns"
    ) -> None:
        """
        Best-effort additive schema sync:
          - infer select schema via LIMIT 0 query
          - add missing columns as NULLABLE using inferred BigQuery types
        """
        if mode not in {"append_new_columns", "sync_all_columns"}:
            return
        self._ensure_dataset()

        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        # Infer schema using a no-row query (lets BigQuery type the expressions)
        probe = self.client.query(
            f"SELECT * FROM ({body}) WHERE 1=0",
            job_config=bigquery.QueryJobConfig(dry_run=False, use_query_cache=False),
            location=self.location,
        )
        probe.result()
        out_fields = {f.name: f for f in (probe.schema or [])}

        # Existing table schema
        table_ref = f"{self.project}.{self.dataset}.{relation}"
        try:
            tbl = self.client.get_table(table_ref)
        except NotFound:
            return
        existing_cols = {f.name for f in (tbl.schema or [])}

        to_add = [name for name in out_fields if name not in existing_cols]
        if not to_add:
            return
        target = self._qualified_identifier(relation, project=self.project, dataset=self.dataset)
        for col in to_add:
            f = out_fields[col]
            typ = str(f.field_type) if hasattr(f, "field_type") else "STRING"
            self.client.query(
                f"ALTER TABLE {target} ADD COLUMN {col} {typ}",
                location=self.location,
            ).result()

ENGINE_NAME class-attribute instance-attribute

ENGINE_NAME = 'bigquery'

BigQuery executor (pandas DataFrames). ENV/Profiles typically use: - FF_BQ_PROJECT - FF_BQ_DATASET - FF_BQ_LOCATION (optional)

run_sql

run_sql(node, env)

Orchestrate SQL models

1) Render Jinja (ref/source/this) and strip leading {{ config(...) }}. 2) If the SQL is full DDL (CREATE …), execute it verbatim (passthrough). 3) Otherwise, normalize to CREATE OR REPLACE {TABLE|VIEW} AS . The body is CTE-aware (keeps WITH … SELECT … intact).

On failure, raise ModelExecutionError with a helpful snippet.

Source code in src/fastflowtransform/executors/base.py

def run_sql(self, node: Node, env: Environment) -> None:
    """
    Orchestrate SQL models:
      1) Render Jinja (ref/source/this) and strip leading {{ config(...) }}.
      2) If the SQL is full DDL (CREATE …), execute it verbatim (passthrough).
      3) Otherwise, normalize to CREATE OR REPLACE {TABLE|VIEW} AS <body>.
         The body is CTE-aware (keeps WITH … SELECT … intact).
    On failure, raise ModelExecutionError with a helpful snippet.
    """
    sql_rendered = self.render_sql(
        node,
        env,
        ref_resolver=lambda name: self._resolve_ref(name, env),
        source_resolver=self._resolve_source,
    )
    sql = self._strip_leading_config(sql_rendered).strip()

    materialization = (node.meta or {}).get("materialized", "table")
    if materialization == "ephemeral":
        return

    # 1) Direct DDL passthrough (CREATE [OR REPLACE] {TABLE|VIEW} …)
    if self._looks_like_direct_ddl(sql):
        try:
            self._execute_sql_direct(sql, node)
            return
        except NotImplementedError:
            # Engine doesn't implement direct DDL → fall back to normalized materialization.
            pass
        except Exception as e:
            raise ModelExecutionError(
                node_name=node.name,
                relation=relation_for(node.name),
                message=str(e),
                sql_snippet=sql,
            ) from e

    # 2) Normalized materialization path (CTE-safe body)
    body = self._selectable_body(sql).rstrip(" ;\n\t")
    target_sql = self._format_relation_for_ref(node.name)

    # Centralized SQL preview logging (applies to ALL engines)
    preview = (
        f"=== MATERIALIZE ===\n"
        f"-- model: {node.name}\n"
        f"-- materialized: {materialization}\n"
        f"-- target: {target_sql}\n"
        f"{body}\n"
    )
    echo_debug(preview)

    try:
        self._apply_sql_materialization(node, target_sql, body, materialization)
    except Exception as e:
        preview = f"-- materialized={materialization}\n-- target={target_sql}\n{body}"
        raise ModelExecutionError(
            node_name=node.name,
            relation=relation_for(node.name),
            message=str(e),
            sql_snippet=preview,
        ) from e

on_node_built

on_node_built(node, relation, fingerprint)

Write/update dataset._ff_meta after a successful build.

Source code in src/fastflowtransform/executors/bigquery_exec.py

def on_node_built(self, node: Node, relation: str, fingerprint: str) -> None:
    """
    Write/update dataset._ff_meta after a successful build.
    """
    try:
        ensure_meta_table(self)
        upsert_meta(self, node.name, relation, fingerprint, "bigquery")
    except Exception:
        pass

exists_relation

exists_relation(relation)

Check presence in INFORMATION_SCHEMA for tables/views.

Source code in src/fastflowtransform/executors/bigquery_exec.py

def exists_relation(self, relation: str) -> bool:
    """
    Check presence in INFORMATION_SCHEMA for tables/views.
    """
    proj = self.project
    dset = self.dataset
    rel = relation
    q = f"""
    SELECT 1
    FROM `{proj}.{dset}.INFORMATION_SCHEMA.TABLES`
    WHERE LOWER(table_name)=LOWER(@rel)
    UNION ALL
    SELECT 1
    FROM `{proj}.{dset}.INFORMATION_SCHEMA.VIEWS`
    WHERE LOWER(table_name)=LOWER(@rel)
    LIMIT 1
    """
    job = self.client.query(
        q,
        job_config=bigquery.QueryJobConfig(
            query_parameters=[bigquery.ScalarQueryParameter("rel", "STRING", rel)]
        ),
        location=self.location,
    )
    return bool(list(job.result()))

create_table_as

create_table_as(relation, select_sql)

CREATE TABLE AS with cleaned SELECT body (no trailing semicolons).

Source code in src/fastflowtransform/executors/bigquery_exec.py

def create_table_as(self, relation: str, select_sql: str) -> None:
    """
    CREATE TABLE AS with cleaned SELECT body (no trailing semicolons).
    """
    self._ensure_dataset()
    body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
    target = self._qualified_identifier(relation, project=self.project, dataset=self.dataset)
    self.client.query(
        f"CREATE TABLE {target} AS {body}",
        location=self.location,
    ).result()

incremental_insert

incremental_insert(relation, select_sql)

INSERT INTO with cleaned SELECT body.

Source code in src/fastflowtransform/executors/bigquery_exec.py

def incremental_insert(self, relation: str, select_sql: str) -> None:
    """
    INSERT INTO with cleaned SELECT body.
    """
    self._ensure_dataset()
    body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
    target = self._qualified_identifier(relation, project=self.project, dataset=self.dataset)
    self.client.query(
        f"INSERT INTO {target} {body}",
        location=self.location,
    ).result()

incremental_merge

incremental_merge(relation, select_sql, unique_key)

Portable fallback without native MERGE

• DELETE collisions via WHERE EXISTS against the cleaned SELECT body
• INSERT new rows from the same body

Source code in src/fastflowtransform/executors/bigquery_exec.py

def incremental_merge(self, relation: str, select_sql: str, unique_key: list[str]) -> None:
    """
    Portable fallback without native MERGE:
      - DELETE collisions via WHERE EXISTS against the cleaned SELECT body
      - INSERT new rows from the same body
    """
    self._ensure_dataset()
    body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
    target = self._qualified_identifier(relation, project=self.project, dataset=self.dataset)
    pred = " AND ".join([f"t.{k}=s.{k}" for k in unique_key]) or "FALSE"

    delete_sql = f"""
    DELETE FROM {target} t
    WHERE EXISTS (SELECT 1 FROM ({body}) s WHERE {pred})
    """
    self.client.query(delete_sql, location=self.location).result()

    insert_sql = f"INSERT INTO {target} SELECT * FROM ({body})"
    self.client.query(insert_sql, location=self.location).result()

alter_table_sync_schema

alter_table_sync_schema(relation, select_sql, *, mode='append_new_columns')

Best-effort additive schema sync

• infer select schema via LIMIT 0 query
• add missing columns as NULLABLE using inferred BigQuery types

Source code in src/fastflowtransform/executors/bigquery_exec.py

def alter_table_sync_schema(
    self, relation: str, select_sql: str, *, mode: str = "append_new_columns"
) -> None:
    """
    Best-effort additive schema sync:
      - infer select schema via LIMIT 0 query
      - add missing columns as NULLABLE using inferred BigQuery types
    """
    if mode not in {"append_new_columns", "sync_all_columns"}:
        return
    self._ensure_dataset()

    body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
    # Infer schema using a no-row query (lets BigQuery type the expressions)
    probe = self.client.query(
        f"SELECT * FROM ({body}) WHERE 1=0",
        job_config=bigquery.QueryJobConfig(dry_run=False, use_query_cache=False),
        location=self.location,
    )
    probe.result()
    out_fields = {f.name: f for f in (probe.schema or [])}

    # Existing table schema
    table_ref = f"{self.project}.{self.dataset}.{relation}"
    try:
        tbl = self.client.get_table(table_ref)
    except NotFound:
        return
    existing_cols = {f.name for f in (tbl.schema or [])}

    to_add = [name for name in out_fields if name not in existing_cols]
    if not to_add:
        return
    target = self._qualified_identifier(relation, project=self.project, dataset=self.dataset)
    for col in to_add:
        f = out_fields[col]
        typ = str(f.field_type) if hasattr(f, "field_type") else "STRING"
        self.client.query(
            f"ALTER TABLE {target} ADD COLUMN {col} {typ}",
            location=self.location,
        ).result()

DatabricksSparkExecutor

Bases: BaseExecutor[DataFrame]

Source code in src/fastflowtransform/executors/databricks_spark_exec.py

class DatabricksSparkExecutor(BaseExecutor[SDF]):
    ENGINE_NAME = "databricks_spark"
    """Spark/Databricks executor without pandas: Python models operate on Spark DataFrames."""

    def __init__(
        self,
        master: str = "local[*]",
        app_name: str = "fastflowtransform",
        *,
        extra_conf: dict[str, Any] | None = None,
        warehouse_dir: str | None = None,
        use_hive_metastore: bool = False,
        catalog: str | None = None,
        database: str | None = None,
        table_format: str | None = "parquet",
        table_options: dict[str, Any] | None = None,
    ):
        builder = SparkSession.builder.master(master).appName(app_name)

        warehouse_path: Path | None = None
        if warehouse_dir:
            warehouse_path = Path(warehouse_dir).expanduser()
            if not warehouse_path.is_absolute():
                warehouse_path = Path.cwd() / warehouse_path
            warehouse_path.mkdir(parents=True, exist_ok=True)
            builder = builder.config("spark.sql.warehouse.dir", str(warehouse_path))

        if catalog:
            builder = builder.config("spark.sql.catalog.spark_catalog", catalog)

        if extra_conf:
            for key, value in extra_conf.items():
                if value is not None:
                    builder = builder.config(str(key), str(value))

        if use_hive_metastore:
            builder = builder.config("spark.sql.catalogImplementation", "hive")
            builder = builder.enableHiveSupport()

        self.spark = builder.getOrCreate()
        # Lightweight testing shim so tests can call executor.con.execute("SQL")
        self.con = _SparkConnShim(self.spark)
        self._registered_path_sources: dict[str, dict[str, Any]] = {}
        self.warehouse_dir = warehouse_path
        self.catalog = catalog
        self.database = database
        self.schema = database
        if database:
            self.spark.sql(f"CREATE DATABASE IF NOT EXISTS `{database}`")
            with suppress(Exception):
                self.spark.catalog.setCurrentDatabase(database)

        fmt = (table_format or "").strip().lower()
        self.spark_table_format: str | None = fmt or None
        if table_options:
            self.spark_table_options = {str(k): str(v) for k, v in table_options.items()}
        else:
            self.spark_table_options = {}

    # ---------- Frame hooks (required) ----------
    def _read_relation(self, relation: str, node: Node, deps: Iterable[str]) -> SDF:
        # relation may optionally be "db.table" (via source()/ref())
        return self.spark.table(relation)

    def _materialize_relation(self, relation: str, df: SDF, node: Node) -> None:
        if not self._is_frame(df):
            raise TypeError("Spark model must return a Spark DataFrame")
        storage_meta = self._storage_meta(node, relation)
        if storage_meta.get("path"):
            self._write_to_storage_path(relation, df, storage_meta)
            return
        # write as a table in Hive/Unity/Delta environments
        self._save_df_as_table(relation, df, storage=storage_meta)

    def _create_view_over_table(self, view_name: str, backing_table: str, node: Node) -> None:
        self.spark.sql(f"CREATE OR REPLACE VIEW `{view_name}` AS SELECT * FROM `{backing_table}`")

    def _validate_required(
        self, node_name: str, inputs: Any, requires: dict[str, set[str]]
    ) -> None:
        if not requires:
            return

        def cols(df: SDF) -> set[str]:
            return set(df.schema.fieldNames())

        errors: list[str] = []
        # Single dependency: requires typically contains exactly one entry (ignore the key)
        if isinstance(inputs, SDF):
            need = next(iter(requires.values()), set())
            missing = need - cols(inputs)
            if missing:
                errors.append(f"- missing columns: {sorted(missing)} | have={sorted(cols(inputs))}")
        else:
            # Multiple dependencies: keys in requires = physical relations (relation_for(dep))
            for rel, need in requires.items():
                if rel not in inputs:
                    errors.append(f"- missing dependency key '{rel}'")
                    continue
                missing = need - cols(inputs[rel])
                if missing:
                    errors.append(
                        f"- [{rel}] missing: {sorted(missing)} | have={sorted(cols(inputs[rel]))}"
                    )

        if errors:
            raise ValueError(
                "Required columns check failed for Spark model "
                f"'{node_name}'.\n" + "\n".join(errors)
            )

    def _columns_of(self, frame: SDF) -> list[str]:  # pragma: no cover
        return frame.schema.fieldNames()

    def _is_frame(self, obj: Any) -> bool:  # pragma: no cover
        return isinstance(obj, SDF)

    def _frame_name(self) -> str:  # pragma: no cover
        return "Spark"

    # ---- Helpers ----
    @staticmethod
    def _q_ident(value: str | None) -> str:
        if value is None:
            return ""
        return f"`{value.replace('`', '``')}`"

    def _storage_meta(self, node: Node | None, relation: str) -> dict[str, Any]:
        """
        Retrieve configured storage overrides for the logical node backing `relation`.
        """
        rel_clean = self._strip_quotes(relation)
        if node is not None:
            meta = dict((node.meta or {}).get("storage") or {})
            if meta:
                return meta
            lookup = storage.get_model_storage(node.name)
            if lookup:
                return lookup
        for cand in getattr(REGISTRY, "nodes", {}).values():
            try:
                if self._strip_quotes(relation_for(cand.name)) == rel_clean:
                    meta = dict((cand.meta or {}).get("storage") or {})
                    if meta:
                        return meta
                    lookup = storage.get_model_storage(cand.name)
                    if lookup:
                        return lookup
            except Exception:
                continue
        return storage.get_model_storage(rel_clean)

    def _write_to_storage_path(
        self, relation: str, df: SDF, storage_meta: dict[str, Any]
    ) -> None:  # pragma: no cover
        parts = self._identifier_parts(relation)
        identifier = ".".join(parts)
        storage.spark_write_to_path(
            self.spark,
            identifier,
            df,
            storage=storage_meta,
            default_format=self.spark_table_format,
            default_options=self.spark_table_options,
        )

    # ---- SQL hooks ----
    def _format_relation_for_ref(self, name: str) -> str:
        return self._q_ident(relation_for(name))

    def _format_source_reference(
        self, cfg: dict[str, Any], source_name: str, table_name: str
    ) -> str:
        location = cfg.get("location")
        identifier = cfg.get("identifier")

        if location:
            alias = identifier or f"__ff_src_{source_name}_{table_name}"
            fmt = cfg.get("format")
            if not fmt:
                raise KeyError(
                    f"Source {source_name}.{table_name} requires 'format' when using a location"
                )

            options = dict(cfg.get("options") or {})
            descriptor = {
                "location": location,
                "format": fmt,
                "options": options,
            }
            existing = self._registered_path_sources.get(alias)
            if existing != descriptor:
                reader = self.spark.read.format(fmt)
                if options:
                    reader = reader.options(**options)
                df = reader.load(location)
                df.createOrReplaceTempView(alias)
                self._registered_path_sources[alias] = descriptor
            return self._q_ident(alias)

        if not identifier:
            raise KeyError(f"Source {source_name}.{table_name} missing identifier")

        catalog = cfg.get("catalog") or cfg.get("database")
        schema = cfg.get("schema")
        parts = [p for p in (catalog, schema, identifier) if p]
        if not parts:
            parts = [identifier]
        return ".".join(self._q_ident(str(part)) for part in parts)

    # ---- Spark table helpers ----
    @staticmethod
    def _strip_quotes(identifier: str) -> str:
        return identifier.replace("`", "").replace('"', "")

    def _identifier_parts(self, identifier: str) -> list[str]:
        cleaned = self._strip_quotes(identifier)
        return [part for part in cleaned.split(".") if part]

    def _warehouse_base(self) -> Path | None:
        try:
            conf_val = self.spark.conf.get("spark.sql.warehouse.dir", "spark-warehouse")
        except Exception:
            conf_val = "spark-warehouse"

        if not isinstance(conf_val, str):
            conf_val = str(conf_val)
        parsed = urlparse(conf_val)
        scheme = (parsed.scheme or "").lower()

        if scheme and scheme != "file":
            return None

        if scheme == "file":
            if parsed.netloc and parsed.netloc not in {"", "localhost"}:
                return None
            raw_path = unquote(parsed.path or "")
            if not raw_path:
                return None
            base = Path(raw_path)
        else:
            base = Path(conf_val)

        if not base.is_absolute():
            base = Path.cwd() / base
        return base

    def _table_location(self, parts: list[str]) -> Path | None:
        base = self._warehouse_base()
        if base is None or not parts:
            return None

        filtered = [p for p in parts if p]
        if not filtered:
            return None

        catalog_cutoff = 3
        if len(filtered) >= catalog_cutoff and filtered[0].lower() in {"spark_catalog", "spark"}:
            filtered = filtered[1:]

        table = filtered[-1]
        schema_cutoff = 2
        schema = filtered[-2] if len(filtered) >= schema_cutoff else None

        location = base
        if schema:
            location = location / f"{schema}.db"
        return location / table

    def _save_df_as_table(
        self, identifier: str, df: SDF, *, storage: dict[str, Any] | None = None
    ) -> None:
        parts = self._identifier_parts(identifier)
        if not parts:
            raise ValueError(f"Invalid Spark table identifier: {identifier}")

        storage_meta = storage or self._storage_meta(None, identifier)
        if storage_meta.get("path"):
            self._write_to_storage_path(identifier, df, storage_meta)
            return

        table_name = ".".join(parts)
        target_location = self._table_location(parts)

        def _write() -> None:
            writer = df.write.mode("overwrite")
            if self.spark_table_format:
                writer = writer.format(self.spark_table_format)
            if self.spark_table_options:
                writer = writer.options(**self.spark_table_options)
            writer.saveAsTable(table_name)

        target_sql = ".".join(self._q_ident(p) for p in parts)
        with suppress(Exception):
            self.spark.sql(f"DROP TABLE IF EXISTS {target_sql}")
        if target_location and target_location.exists():
            with suppress(Exception):
                shutil.rmtree(target_location, ignore_errors=True)

        try:
            _write()
        except AnalysisException as exc:  # pragma: no cover - requires real Spark/Delta error
            message = str(exc)
            if target_location and "LOCATION_ALREADY_EXISTS" in message.upper():
                with suppress(Exception):
                    shutil.rmtree(target_location, ignore_errors=True)
                _write()
            else:
                raise

    def _create_or_replace_view(self, target_sql: str, select_body: str, node: Node) -> None:
        self.spark.sql(f"CREATE OR REPLACE VIEW {target_sql} AS {select_body}")

    def _create_or_replace_table(self, target_sql: str, select_body: str, node: Node) -> None:
        preview = f"-- target={target_sql}\n{select_body}"
        try:
            df = self.spark.sql(select_body)
            storage_meta = self._storage_meta(node, target_sql)
            self._save_df_as_table(target_sql, df, storage=storage_meta)
        except Exception as exc:
            raise ModelExecutionError(node.name, target_sql, str(exc), sql_snippet=preview) from exc

    def _create_or_replace_view_from_table(
        self, view_name: str, backing_table: str, node: Node
    ) -> None:
        self.spark.sql(f"CREATE OR REPLACE VIEW `{view_name}` AS SELECT * FROM `{backing_table}`")

    # ---- Meta hook ----
    def on_node_built(self, node: Node, relation: str, fingerprint: str) -> None:
        """After successful materialization, upsert _ff_meta (best-effort)."""
        try:
            ensure_meta_table(self)
            upsert_meta(self, node.name, relation, fingerprint, "databricks_spark")
        except Exception:
            pass

    # ── Incremental API (parity) ─────────────────────────────────────────
    def exists_relation(self, relation: str) -> bool:
        """Check whether a table/view exists (optionally qualified with database)."""
        db, tbl = _split_db_table(relation)
        if db:
            return bool(self.spark.catalog._jcatalog.tableExists(db, tbl))
        return self.spark.catalog.tableExists(tbl)

    def create_table_as(self, relation: str, select_sql: str) -> None:
        """CREATE TABLE AS with cleaned SELECT body."""
        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        df = self.spark.sql(body)
        self._save_df_as_table(relation, df)

    def incremental_insert(self, relation: str, select_sql: str) -> None:
        """INSERT INTO with cleaned SELECT body."""
        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        self.spark.sql(f"INSERT INTO {relation} {body}")

    def incremental_merge(self, relation: str, select_sql: str, unique_key: list[str]) -> None:
        """
        Try Delta MERGE (Databricks typical). If MERGE fails (non-Delta), fallback to full replace.
        """
        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        pred = " AND ".join([f"t.{k}=s.{k}" for k in unique_key]) or "FALSE"
        try:
            # Use inline subquery as source; SET * / INSERT * requires Delta ≥ 1.2 / Spark ≥ 3.4.
            self.spark.sql(
                f"""
                MERGE INTO {relation} AS t
                USING ({body}) AS s
                ON {pred}
                WHEN MATCHED THEN UPDATE SET *
                WHEN NOT MATCHED THEN INSERT *
                """
            )
        except Exception:
            # Fallback: Full replace is safer across lake formats
            df = self.spark.sql(body)
            self._save_df_as_table(relation, df)

    def alter_table_sync_schema(
        self, relation: str, select_sql: str, *, mode: str = "append_new_columns"
    ) -> None:
        """
        Best-effort additive schema sync:
          - infer select schema via LIMIT 0
          - add missing columns as STRING (safe default)
        """
        if mode not in {"append_new_columns", "sync_all_columns"}:
            return
        # Target schema
        try:
            target_df = self.spark.table(relation)
        except Exception:
            return
        existing = {f.name for f in target_df.schema.fields}
        # Output schema from the SELECT
        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        probe = self.spark.sql(f"SELECT * FROM ({body}) q LIMIT 0")
        to_add = [f for f in probe.schema.fields if f.name not in existing]
        if not to_add:
            return

        # Map types best-effort (Spark SQL types); default STRING
        def _spark_sql_type(dt: DataType) -> str:
            # Use simple, portable mapping for documentation UIs & broad compatibility
            return (
                getattr(dt, "simpleString", lambda: "string")().upper()
                if hasattr(dt, "simpleString")
                else "STRING"
            )

        cols_sql = ", ".join([f"`{f.name}` {_spark_sql_type(f.dataType)}" for f in to_add])
        self.spark.sql(f"ALTER TABLE {relation} ADD COLUMNS ({cols_sql})")

ENGINE_NAME class-attribute instance-attribute

ENGINE_NAME = 'databricks_spark'

Spark/Databricks executor without pandas: Python models operate on Spark DataFrames.

run_sql

run_sql(node, env)

Orchestrate SQL models

1) Render Jinja (ref/source/this) and strip leading {{ config(...) }}. 2) If the SQL is full DDL (CREATE …), execute it verbatim (passthrough). 3) Otherwise, normalize to CREATE OR REPLACE {TABLE|VIEW} AS . The body is CTE-aware (keeps WITH … SELECT … intact).

On failure, raise ModelExecutionError with a helpful snippet.

Source code in src/fastflowtransform/executors/base.py

def run_sql(self, node: Node, env: Environment) -> None:
    """
    Orchestrate SQL models:
      1) Render Jinja (ref/source/this) and strip leading {{ config(...) }}.
      2) If the SQL is full DDL (CREATE …), execute it verbatim (passthrough).
      3) Otherwise, normalize to CREATE OR REPLACE {TABLE|VIEW} AS <body>.
         The body is CTE-aware (keeps WITH … SELECT … intact).
    On failure, raise ModelExecutionError with a helpful snippet.
    """
    sql_rendered = self.render_sql(
        node,
        env,
        ref_resolver=lambda name: self._resolve_ref(name, env),
        source_resolver=self._resolve_source,
    )
    sql = self._strip_leading_config(sql_rendered).strip()

    materialization = (node.meta or {}).get("materialized", "table")
    if materialization == "ephemeral":
        return

    # 1) Direct DDL passthrough (CREATE [OR REPLACE] {TABLE|VIEW} …)
    if self._looks_like_direct_ddl(sql):
        try:
            self._execute_sql_direct(sql, node)
            return
        except NotImplementedError:
            # Engine doesn't implement direct DDL → fall back to normalized materialization.
            pass
        except Exception as e:
            raise ModelExecutionError(
                node_name=node.name,
                relation=relation_for(node.name),
                message=str(e),
                sql_snippet=sql,
            ) from e

    # 2) Normalized materialization path (CTE-safe body)
    body = self._selectable_body(sql).rstrip(" ;\n\t")
    target_sql = self._format_relation_for_ref(node.name)

    # Centralized SQL preview logging (applies to ALL engines)
    preview = (
        f"=== MATERIALIZE ===\n"
        f"-- model: {node.name}\n"
        f"-- materialized: {materialization}\n"
        f"-- target: {target_sql}\n"
        f"{body}\n"
    )
    echo_debug(preview)

    try:
        self._apply_sql_materialization(node, target_sql, body, materialization)
    except Exception as e:
        preview = f"-- materialized={materialization}\n-- target={target_sql}\n{body}"
        raise ModelExecutionError(
            node_name=node.name,
            relation=relation_for(node.name),
            message=str(e),
            sql_snippet=preview,
        ) from e

on_node_built

on_node_built(node, relation, fingerprint)

After successful materialization, upsert _ff_meta (best-effort).

Source code in src/fastflowtransform/executors/databricks_spark_exec.py

def on_node_built(self, node: Node, relation: str, fingerprint: str) -> None:
    """After successful materialization, upsert _ff_meta (best-effort)."""
    try:
        ensure_meta_table(self)
        upsert_meta(self, node.name, relation, fingerprint, "databricks_spark")
    except Exception:
        pass

exists_relation

exists_relation(relation)

Check whether a table/view exists (optionally qualified with database).

Source code in src/fastflowtransform/executors/databricks_spark_exec.py

def exists_relation(self, relation: str) -> bool:
    """Check whether a table/view exists (optionally qualified with database)."""
    db, tbl = _split_db_table(relation)
    if db:
        return bool(self.spark.catalog._jcatalog.tableExists(db, tbl))
    return self.spark.catalog.tableExists(tbl)

create_table_as

create_table_as(relation, select_sql)

CREATE TABLE AS with cleaned SELECT body.

Source code in src/fastflowtransform/executors/databricks_spark_exec.py

def create_table_as(self, relation: str, select_sql: str) -> None:
    """CREATE TABLE AS with cleaned SELECT body."""
    body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
    df = self.spark.sql(body)
    self._save_df_as_table(relation, df)

incremental_insert

incremental_insert(relation, select_sql)

INSERT INTO with cleaned SELECT body.

Source code in src/fastflowtransform/executors/databricks_spark_exec.py

def incremental_insert(self, relation: str, select_sql: str) -> None:
    """INSERT INTO with cleaned SELECT body."""
    body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
    self.spark.sql(f"INSERT INTO {relation} {body}")

incremental_merge

incremental_merge(relation, select_sql, unique_key)

Try Delta MERGE (Databricks typical). If MERGE fails (non-Delta), fallback to full replace.

Source code in src/fastflowtransform/executors/databricks_spark_exec.py

def incremental_merge(self, relation: str, select_sql: str, unique_key: list[str]) -> None:
    """
    Try Delta MERGE (Databricks typical). If MERGE fails (non-Delta), fallback to full replace.
    """
    body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
    pred = " AND ".join([f"t.{k}=s.{k}" for k in unique_key]) or "FALSE"
    try:
        # Use inline subquery as source; SET * / INSERT * requires Delta ≥ 1.2 / Spark ≥ 3.4.
        self.spark.sql(
            f"""
            MERGE INTO {relation} AS t
            USING ({body}) AS s
            ON {pred}
            WHEN MATCHED THEN UPDATE SET *
            WHEN NOT MATCHED THEN INSERT *
            """
        )
    except Exception:
        # Fallback: Full replace is safer across lake formats
        df = self.spark.sql(body)
        self._save_df_as_table(relation, df)

alter_table_sync_schema

alter_table_sync_schema(relation, select_sql, *, mode='append_new_columns')

Best-effort additive schema sync

• infer select schema via LIMIT 0
• add missing columns as STRING (safe default)

Source code in src/fastflowtransform/executors/databricks_spark_exec.py

def alter_table_sync_schema(
    self, relation: str, select_sql: str, *, mode: str = "append_new_columns"
) -> None:
    """
    Best-effort additive schema sync:
      - infer select schema via LIMIT 0
      - add missing columns as STRING (safe default)
    """
    if mode not in {"append_new_columns", "sync_all_columns"}:
        return
    # Target schema
    try:
        target_df = self.spark.table(relation)
    except Exception:
        return
    existing = {f.name for f in target_df.schema.fields}
    # Output schema from the SELECT
    body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
    probe = self.spark.sql(f"SELECT * FROM ({body}) q LIMIT 0")
    to_add = [f for f in probe.schema.fields if f.name not in existing]
    if not to_add:
        return

    # Map types best-effort (Spark SQL types); default STRING
    def _spark_sql_type(dt: DataType) -> str:
        # Use simple, portable mapping for documentation UIs & broad compatibility
        return (
            getattr(dt, "simpleString", lambda: "string")().upper()
            if hasattr(dt, "simpleString")
            else "STRING"
        )

    cols_sql = ", ".join([f"`{f.name}` {_spark_sql_type(f.dataType)}" for f in to_add])
    self.spark.sql(f"ALTER TABLE {relation} ADD COLUMNS ({cols_sql})")

DuckExecutor

Bases: BaseExecutor[DataFrame]

Source code in src/fastflowtransform/executors/duckdb_exec.py

class DuckExecutor(BaseExecutor[pd.DataFrame]):
    ENGINE_NAME = "duckdb"

    def __init__(self, db_path: str = ":memory:"):
        if db_path and db_path != ":memory:" and "://" not in db_path:
            with suppress(Exception):
                Path(db_path).parent.mkdir(parents=True, exist_ok=True)
        self.db_path = db_path
        self.con = duckdb.connect(db_path)

    def clone(self) -> DuckExecutor:
        """
        Generates a new Executor instance with own connection for Thread-Worker.
        """
        return DuckExecutor(self.db_path)

    def _exec_many(self, sql: str) -> None:
        """
        Execute multiple SQL statements separated by ';' on the same connection.
        DuckDB normally accepts one statement per execute(), so we split here.
        """
        # very simple splitter - good enough for what we emit in the executor
        for stmt in (part.strip() for part in sql.split(";")):
            if not stmt:
                continue
            self.con.execute(stmt)

    # ---- Frame hooks ----
    def _read_relation(self, relation: str, node: Node, deps: Iterable[str]) -> pd.DataFrame:
        try:
            return self.con.table(relation).df()
        except CatalogException as e:
            existing = [
                r[0]
                for r in self.con.execute(
                    "select table_name from information_schema.tables "
                    "where table_schema in ('main','temp')"
                ).fetchall()
            ]
            raise RuntimeError(
                f"Dependency table not found: '{relation}'\n"
                f"Deps: {list(deps)}\nExisting tables: {existing}\n"
                "Hinweis: gleiche Datei-DB/Connection für Seeding & Run verwenden."
            ) from e

    def _materialize_relation(self, relation: str, df: pd.DataFrame, node: Node) -> None:
        tmp = "_ff_py_out"
        try:
            self.con.register(tmp, df)
            self.con.execute(f'create or replace table "{relation}" as select * from "{tmp}"')
        finally:
            try:
                self.con.unregister(tmp)
            except Exception:
                self.con.execute(f'drop view if exists "{tmp}"')

    def _create_or_replace_view_from_table(
        self, view_name: str, backing_table: str, node: Node
    ) -> None:
        self.con.execute(f'create or replace view "{view_name}" as select * from "{backing_table}"')

    def _frame_name(self) -> str:
        return "pandas"

    # ---- SQL hooks ----
    def _format_relation_for_ref(self, name: str) -> str:
        return _q(relation_for(name))

    def _format_source_reference(
        self, cfg: dict[str, Any], source_name: str, table_name: str
    ) -> str:
        location = cfg.get("location")
        if location:
            raise NotImplementedError("DuckDB executor does not support path-based sources yet.")

        identifier = cfg.get("identifier")
        if not identifier:
            raise KeyError(f"Source {source_name}.{table_name} missing identifier")

        parts = [
            p
            for p in (
                cfg.get("catalog") or cfg.get("database"),
                cfg.get("schema"),
                identifier,
            )
            if p
        ]
        if not parts:
            parts = [identifier]

        return ".".join(_q(str(part)) for part in parts)

    def _create_or_replace_view(self, target_sql: str, select_body: str, node: Node) -> None:
        self.con.execute(f"create or replace view {target_sql} as {select_body}")

    def _create_or_replace_table(self, target_sql: str, select_body: str, node: Node) -> None:
        self.con.execute(f"create or replace table {target_sql} as {select_body}")

    # ---- Meta hook ----
    def on_node_built(self, node: Node, relation: str, fingerprint: str) -> None:
        """
        After successful materialization, ensure the meta table exists and upsert the row.
        """
        # Best-effort: do not let meta errors break the run
        try:
            ensure_meta_table(self)
            upsert_meta(self, node.name, relation, fingerprint, "duckdb")
        except Exception:
            pass

    # ── Incremental API ────────────────────────────────────────────────────
    def exists_relation(self, relation: str) -> bool:
        sql = """
          select 1
          from information_schema.tables
          where table_schema in ('main','temp')
            and lower(table_name) = lower(?)
          limit 1
        """
        res = self.con.execute(sql, [relation]).fetchone()
        return bool(res)

    def create_table_as(self, relation: str, select_sql: str) -> None:
        # Use only the SELECT body and strip trailing semicolons for safety.
        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        self.con.execute(f"create table {relation} as {body}")

    def incremental_insert(self, relation: str, select_sql: str) -> None:
        # Ensure the inner SELECT is clean (no trailing semicolon; SELECT body only).
        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        self.con.execute(f"insert into {relation} {body}")

    def incremental_merge(self, relation: str, select_sql: str, unique_key: list[str]) -> None:
        """
        Fallback strategy for DuckDB:
        - DELETE collisions via DELETE ... USING (<select>) s
        - INSERT all rows via INSERT ... SELECT * FROM (<select>)
        We intentionally do NOT use a CTE here, because we execute two separate
        statements and DuckDB won't see the CTE from the previous statement.
        """
        # 1) clean inner SELECT
        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")

        # 2) predicate for DELETE
        keys_pred = " AND ".join([f"t.{k}=s.{k}" for k in unique_key]) or "FALSE"

        # 3) first: delete collisions
        delete_sql = f"delete from {relation} t using ({body}) s where {keys_pred}"
        self.con.execute(delete_sql)

        # 4) then: insert fresh rows
        insert_sql = f"insert into {relation} select * from ({body}) src"
        self.con.execute(insert_sql)

    def alter_table_sync_schema(
        self, relation: str, select_sql: str, *, mode: str = "append_new_columns"
    ) -> None:
        """
        Best-effort: add new columns with inferred type.
        """
        # Probe: empty projection from the SELECT (cleaned to avoid parser issues).
        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        probe = self.con.execute(f"select * from ({body}) as q limit 0")
        cols = [c[0] for c in probe.description or []]
        # vorhandene Spalten
        existing = {
            r[0]
            for r in self.con.execute(
                "select column_name from information_schema.columns "
                + "where lower(table_name)=lower(?)",
                [relation],
            ).fetchall()
        }
        add = [c for c in cols if c not in existing]
        for c in add:
            # Typ heuristisch: typeof aus einer CAST-Probe; fallback VARCHAR
            try:
                # Versuche Typ aus Expression abzuleiten (best effort)
                self.con.execute(f"alter table {relation} add column {c} varchar")
            except Exception:
                self.con.execute(f"alter table {relation} add column {c} varchar")

run_sql

run_sql(node, env)

Orchestrate SQL models

1) Render Jinja (ref/source/this) and strip leading {{ config(...) }}. 2) If the SQL is full DDL (CREATE …), execute it verbatim (passthrough). 3) Otherwise, normalize to CREATE OR REPLACE {TABLE|VIEW} AS . The body is CTE-aware (keeps WITH … SELECT … intact).

On failure, raise ModelExecutionError with a helpful snippet.

Source code in src/fastflowtransform/executors/base.py

def run_sql(self, node: Node, env: Environment) -> None:
    """
    Orchestrate SQL models:
      1) Render Jinja (ref/source/this) and strip leading {{ config(...) }}.
      2) If the SQL is full DDL (CREATE …), execute it verbatim (passthrough).
      3) Otherwise, normalize to CREATE OR REPLACE {TABLE|VIEW} AS <body>.
         The body is CTE-aware (keeps WITH … SELECT … intact).
    On failure, raise ModelExecutionError with a helpful snippet.
    """
    sql_rendered = self.render_sql(
        node,
        env,
        ref_resolver=lambda name: self._resolve_ref(name, env),
        source_resolver=self._resolve_source,
    )
    sql = self._strip_leading_config(sql_rendered).strip()

    materialization = (node.meta or {}).get("materialized", "table")
    if materialization == "ephemeral":
        return

    # 1) Direct DDL passthrough (CREATE [OR REPLACE] {TABLE|VIEW} …)
    if self._looks_like_direct_ddl(sql):
        try:
            self._execute_sql_direct(sql, node)
            return
        except NotImplementedError:
            # Engine doesn't implement direct DDL → fall back to normalized materialization.
            pass
        except Exception as e:
            raise ModelExecutionError(
                node_name=node.name,
                relation=relation_for(node.name),
                message=str(e),
                sql_snippet=sql,
            ) from e

    # 2) Normalized materialization path (CTE-safe body)
    body = self._selectable_body(sql).rstrip(" ;\n\t")
    target_sql = self._format_relation_for_ref(node.name)

    # Centralized SQL preview logging (applies to ALL engines)
    preview = (
        f"=== MATERIALIZE ===\n"
        f"-- model: {node.name}\n"
        f"-- materialized: {materialization}\n"
        f"-- target: {target_sql}\n"
        f"{body}\n"
    )
    echo_debug(preview)

    try:
        self._apply_sql_materialization(node, target_sql, body, materialization)
    except Exception as e:
        preview = f"-- materialized={materialization}\n-- target={target_sql}\n{body}"
        raise ModelExecutionError(
            node_name=node.name,
            relation=relation_for(node.name),
            message=str(e),
            sql_snippet=preview,
        ) from e

clone

clone()

Generates a new Executor instance with own connection for Thread-Worker.

Source code in src/fastflowtransform/executors/duckdb_exec.py

def clone(self) -> DuckExecutor:
    """
    Generates a new Executor instance with own connection for Thread-Worker.
    """
    return DuckExecutor(self.db_path)

on_node_built

on_node_built(node, relation, fingerprint)

After successful materialization, ensure the meta table exists and upsert the row.

Source code in src/fastflowtransform/executors/duckdb_exec.py

def on_node_built(self, node: Node, relation: str, fingerprint: str) -> None:
    """
    After successful materialization, ensure the meta table exists and upsert the row.
    """
    # Best-effort: do not let meta errors break the run
    try:
        ensure_meta_table(self)
        upsert_meta(self, node.name, relation, fingerprint, "duckdb")
    except Exception:
        pass

incremental_merge

incremental_merge(relation, select_sql, unique_key)

Fallback strategy for DuckDB: - DELETE collisions via DELETE ... USING () s - INSERT all rows via INSERT ... SELECT * FROM () We intentionally do NOT use a CTE here, because we execute two separate statements and DuckDB won't see the CTE from the previous statement.

Source code in src/fastflowtransform/executors/duckdb_exec.py

def incremental_merge(self, relation: str, select_sql: str, unique_key: list[str]) -> None:
    """
    Fallback strategy for DuckDB:
    - DELETE collisions via DELETE ... USING (<select>) s
    - INSERT all rows via INSERT ... SELECT * FROM (<select>)
    We intentionally do NOT use a CTE here, because we execute two separate
    statements and DuckDB won't see the CTE from the previous statement.
    """
    # 1) clean inner SELECT
    body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")

    # 2) predicate for DELETE
    keys_pred = " AND ".join([f"t.{k}=s.{k}" for k in unique_key]) or "FALSE"

    # 3) first: delete collisions
    delete_sql = f"delete from {relation} t using ({body}) s where {keys_pred}"
    self.con.execute(delete_sql)

    # 4) then: insert fresh rows
    insert_sql = f"insert into {relation} select * from ({body}) src"
    self.con.execute(insert_sql)

alter_table_sync_schema

alter_table_sync_schema(relation, select_sql, *, mode='append_new_columns')

Best-effort: add new columns with inferred type.

Source code in src/fastflowtransform/executors/duckdb_exec.py

def alter_table_sync_schema(
    self, relation: str, select_sql: str, *, mode: str = "append_new_columns"
) -> None:
    """
    Best-effort: add new columns with inferred type.
    """
    # Probe: empty projection from the SELECT (cleaned to avoid parser issues).
    body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
    probe = self.con.execute(f"select * from ({body}) as q limit 0")
    cols = [c[0] for c in probe.description or []]
    # vorhandene Spalten
    existing = {
        r[0]
        for r in self.con.execute(
            "select column_name from information_schema.columns "
            + "where lower(table_name)=lower(?)",
            [relation],
        ).fetchall()
    }
    add = [c for c in cols if c not in existing]
    for c in add:
        # Typ heuristisch: typeof aus einer CAST-Probe; fallback VARCHAR
        try:
            # Versuche Typ aus Expression abzuleiten (best effort)
            self.con.execute(f"alter table {relation} add column {c} varchar")
        except Exception:
            self.con.execute(f"alter table {relation} add column {c} varchar")

PostgresExecutor

Bases: BaseExecutor[DataFrame]

Source code in src/fastflowtransform/executors/postgres_exec.py

class PostgresExecutor(BaseExecutor[pd.DataFrame]):
    ENGINE_NAME = "postgres"

    def __init__(self, dsn: str, schema: str | None = None):
        """
        Initialize Postgres executor.

        dsn     e.g.: postgresql+psycopg://user:pass@localhost:5432/dbname
        schema  default schema for reads/writes (also used for search_path)
        """
        if not dsn:
            raise ProfileConfigError(
                "Postgres DSN not set. Hint: profiles.yml → postgres.dsn or env FF_PG_DSN."
            )
        self.engine: Engine = create_engine(dsn, future=True)
        self.schema = schema

        if self.schema:
            try:
                with self.engine.begin() as conn:
                    conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {self._q_ident(self.schema)}"))
            except SQLAlchemyError as exc:
                raise ProfileConfigError(
                    f"Failed to ensure schema '{self.schema}' exists: {exc}"
                ) from exc

        # ⇣ fastflowtransform.testing expects executor.con.execute("SQL")
        self.con = SAConnShim(self.engine, schema=self.schema)

    # --- Helpers ---------------------------------------------------------
    def _q_ident(self, ident: str) -> str:
        # Simple, safe quoting for identifiers
        return '"' + ident.replace('"', '""') + '"'

    def _qualified(self, relname: str, schema: str | None = None) -> str:
        sch = schema or self.schema
        if sch:
            return f"{self._q_ident(sch)}.{self._q_ident(relname)}"
        return self._q_ident(relname)

    def _set_search_path(self, conn: Connection | SAConnShim) -> None:
        if self.schema:
            conn.execute(text(f"SET LOCAL search_path = {self._q_ident(self.schema)}"))

    def _extract_select_like(self, sql_or_body: str) -> str:
        """
        Normalize a SELECT/CTE body:
        - Accept full statements and strip everything before the first WITH/SELECT.
        - Strip trailing semicolons/whitespace.
        """
        s = (sql_or_body or "").lstrip()
        lower = s.lower()
        pos_with = lower.find("with")
        pos_select = lower.find("select")
        if pos_with == -1 and pos_select == -1:
            return s.rstrip(";\n\t ")
        start = min([p for p in (pos_with, pos_select) if p != -1])
        return s[start:].rstrip(";\n\t ")

    # ---------- IO ----------
    def _read_relation(self, relation: str, node: Node, deps: Iterable[str]) -> pd.DataFrame:
        qualified = self._qualified(relation)
        try:
            with self.engine.begin() as conn:
                if self.schema:
                    conn.execute(text(f'SET LOCAL search_path = "{self.schema}"'))
                return pd.read_sql_query(text(f"select * from {qualified}"), conn)
        except ProgrammingError as e:
            raise e

    def _materialize_relation(self, relation: str, df: pd.DataFrame, node: Node) -> None:
        try:
            df.to_sql(
                relation,
                self.engine,
                if_exists="replace",
                index=False,
                schema=self.schema,
                method="multi",
            )
        except SQLAlchemyError as e:
            raise ModelExecutionError(
                node_name=node.name, relation=self._qualified(relation), message=str(e)
            ) from e

    # ---------- Python view helper ----------
    def _create_or_replace_view_from_table(
        self, view_name: str, backing_table: str, node: Node
    ) -> None:
        q_view = self._qualified(view_name)
        q_back = self._qualified(backing_table)
        try:
            with self.engine.begin() as c:
                self._set_search_path(c)
                c.execute(text(f"DROP VIEW IF EXISTS {q_view} CASCADE"))
                c.execute(text(f"CREATE OR REPLACE VIEW {q_view} AS SELECT * FROM {q_back}"))
        except Exception as e:
            raise ModelExecutionError(node.name, q_view, str(e)) from e

    def _frame_name(self) -> str:
        return "pandas"

    # ---- SQL hooks ----
    def _format_relation_for_ref(self, name: str) -> str:
        return self._qualified(relation_for(name))

    def _format_source_reference(
        self, cfg: dict[str, Any], source_name: str, table_name: str
    ) -> str:
        if cfg.get("location"):
            raise NotImplementedError("Postgres executor does not support path-based sources.")

        ident = cfg.get("identifier")
        if not ident:
            raise KeyError(f"Source {source_name}.{table_name} missing identifier")

        relation = self._qualified(ident, schema=cfg.get("schema"))
        database = cfg.get("database") or cfg.get("catalog")
        if database:
            return f"{self._q_ident(database)}.{relation}"
        return relation

    def _create_or_replace_view(self, target_sql: str, select_body: str, node: Node) -> None:
        try:
            with self.engine.begin() as conn:
                self._set_search_path(conn)
                conn.execute(text(f"DROP VIEW IF EXISTS {target_sql} CASCADE"))
                conn.execute(text(f"CREATE OR REPLACE VIEW {target_sql} AS {select_body}"))
        except Exception as e:
            preview = f"-- target={target_sql}\n{select_body}"
            raise ModelExecutionError(node.name, target_sql, str(e), sql_snippet=preview) from e

    def _create_or_replace_table(self, target_sql: str, select_body: str, node: Node) -> None:
        """
        Postgres does NOT support 'CREATE OR REPLACE TABLE'.
        Use DROP TABLE IF EXISTS + CREATE TABLE AS, and accept CTE bodies.
        """
        try:
            with self.engine.begin() as conn:
                self._set_search_path(conn)
                conn.execute(text(f"DROP TABLE IF EXISTS {target_sql} CASCADE"))
                conn.execute(text(f"CREATE TABLE {target_sql} AS {select_body}"))
        except Exception as e:
            preview = f"-- target={target_sql}\n{select_body}"
            raise ModelExecutionError(node.name, target_sql, str(e), sql_snippet=preview) from e

    # ---------- meta ----------
    def on_node_built(self, node: Node, relation: str, fingerprint: str) -> None:
        """
        Write/update _ff_meta in the current schema after a successful build.
        """
        try:
            ensure_meta_table(self)
            upsert_meta(self, node.name, relation, fingerprint, "postgres")
        except Exception:
            pass

    # ── Incremental API ────────────────────────────────────────────────────
    def exists_relation(self, relation: str) -> bool:
        """
        Return True if a table OR view exists for 'relation' in current schema.
        """
        sql = text(
            """
            select 1
            from information_schema.tables
            where table_schema = current_schema()
              and lower(table_name) = lower(:t)
            union all
            select 1
            from information_schema.views
            where table_schema = current_schema()
              and lower(table_name) = lower(:t)
            limit 1
            """
        )
        with self.engine.begin() as con:
            return bool(con.execute(sql, {"t": relation}).fetchone())

    def create_table_as(self, relation: str, select_sql: str) -> None:
        body = self._extract_select_like(select_sql)
        qrel = self._qualified(relation)
        with self.engine.begin() as con:
            self._set_search_path(con)
            con.execute(text(f"create table {qrel} as {body}"))

    def incremental_insert(self, relation: str, select_sql: str) -> None:
        body = self._extract_select_like(select_sql)
        qrel = self._qualified(relation)
        with self.engine.begin() as con:
            self._set_search_path(con)
            con.execute(text(f"insert into {qrel} {body}"))

    def incremental_merge(self, relation: str, select_sql: str, unique_key: list[str]) -> None:
        """
        Portable fallback: staging + delete + insert.
        """
        body = self._extract_select_like(select_sql)
        qrel = self._qualified(relation)
        pred = " AND ".join([f"t.{k}=s.{k}" for k in unique_key])
        with self.engine.begin() as con:
            self._set_search_path(con)
            con.execute(text(f"create temporary table ff_stg as {body}"))
            try:
                con.execute(text(f"delete from {qrel} t using ff_stg s where {pred}"))
                con.execute(text(f"insert into {qrel} select * from ff_stg"))
            finally:
                con.execute(text("drop table if exists ff_stg"))

    def alter_table_sync_schema(
        self, relation: str, select_sql: str, *, mode: str = "append_new_columns"
    ) -> None:
        """
        Add new columns present in SELECT but missing on target (as text).
        """
        body = self._extract_select_like(select_sql)
        qrel = self._qualified(relation)
        with self.engine.begin() as con:
            self._set_search_path(con)
            cols = [r[0] for r in con.execute(text(f"select * from ({body}) q limit 0"))]
            existing = {
                r[0]
                for r in con.execute(
                    text(
                        "select column_name from information_schema.columns "
                        "where table_schema = current_schema() and lower(table_name)=lower(:t)"
                    ),
                    {"t": relation},
                ).fetchall()
            }
            add = [c for c in cols if c not in existing]
            for c in add:
                con.execute(text(f'alter table {qrel} add column "{c}" text'))

run_sql

run_sql(node, env)

Orchestrate SQL models

1) Render Jinja (ref/source/this) and strip leading {{ config(...) }}. 2) If the SQL is full DDL (CREATE …), execute it verbatim (passthrough). 3) Otherwise, normalize to CREATE OR REPLACE {TABLE|VIEW} AS . The body is CTE-aware (keeps WITH … SELECT … intact).

On failure, raise ModelExecutionError with a helpful snippet.

Source code in src/fastflowtransform/executors/base.py

def run_sql(self, node: Node, env: Environment) -> None:
    """
    Orchestrate SQL models:
      1) Render Jinja (ref/source/this) and strip leading {{ config(...) }}.
      2) If the SQL is full DDL (CREATE …), execute it verbatim (passthrough).
      3) Otherwise, normalize to CREATE OR REPLACE {TABLE|VIEW} AS <body>.
         The body is CTE-aware (keeps WITH … SELECT … intact).
    On failure, raise ModelExecutionError with a helpful snippet.
    """
    sql_rendered = self.render_sql(
        node,
        env,
        ref_resolver=lambda name: self._resolve_ref(name, env),
        source_resolver=self._resolve_source,
    )
    sql = self._strip_leading_config(sql_rendered).strip()

    materialization = (node.meta or {}).get("materialized", "table")
    if materialization == "ephemeral":
        return

    # 1) Direct DDL passthrough (CREATE [OR REPLACE] {TABLE|VIEW} …)
    if self._looks_like_direct_ddl(sql):
        try:
            self._execute_sql_direct(sql, node)
            return
        except NotImplementedError:
            # Engine doesn't implement direct DDL → fall back to normalized materialization.
            pass
        except Exception as e:
            raise ModelExecutionError(
                node_name=node.name,
                relation=relation_for(node.name),
                message=str(e),
                sql_snippet=sql,
            ) from e

    # 2) Normalized materialization path (CTE-safe body)
    body = self._selectable_body(sql).rstrip(" ;\n\t")
    target_sql = self._format_relation_for_ref(node.name)

    # Centralized SQL preview logging (applies to ALL engines)
    preview = (
        f"=== MATERIALIZE ===\n"
        f"-- model: {node.name}\n"
        f"-- materialized: {materialization}\n"
        f"-- target: {target_sql}\n"
        f"{body}\n"
    )
    echo_debug(preview)

    try:
        self._apply_sql_materialization(node, target_sql, body, materialization)
    except Exception as e:
        preview = f"-- materialized={materialization}\n-- target={target_sql}\n{body}"
        raise ModelExecutionError(
            node_name=node.name,
            relation=relation_for(node.name),
            message=str(e),
            sql_snippet=preview,
        ) from e

on_node_built

on_node_built(node, relation, fingerprint)

Write/update _ff_meta in the current schema after a successful build.

Source code in src/fastflowtransform/executors/postgres_exec.py

def on_node_built(self, node: Node, relation: str, fingerprint: str) -> None:
    """
    Write/update _ff_meta in the current schema after a successful build.
    """
    try:
        ensure_meta_table(self)
        upsert_meta(self, node.name, relation, fingerprint, "postgres")
    except Exception:
        pass

exists_relation

exists_relation(relation)

Return True if a table OR view exists for 'relation' in current schema.

Source code in src/fastflowtransform/executors/postgres_exec.py

def exists_relation(self, relation: str) -> bool:
    """
    Return True if a table OR view exists for 'relation' in current schema.
    """
    sql = text(
        """
        select 1
        from information_schema.tables
        where table_schema = current_schema()
          and lower(table_name) = lower(:t)
        union all
        select 1
        from information_schema.views
        where table_schema = current_schema()
          and lower(table_name) = lower(:t)
        limit 1
        """
    )
    with self.engine.begin() as con:
        return bool(con.execute(sql, {"t": relation}).fetchone())

incremental_merge

incremental_merge(relation, select_sql, unique_key)

Portable fallback: staging + delete + insert.

Source code in src/fastflowtransform/executors/postgres_exec.py

def incremental_merge(self, relation: str, select_sql: str, unique_key: list[str]) -> None:
    """
    Portable fallback: staging + delete + insert.
    """
    body = self._extract_select_like(select_sql)
    qrel = self._qualified(relation)
    pred = " AND ".join([f"t.{k}=s.{k}" for k in unique_key])
    with self.engine.begin() as con:
        self._set_search_path(con)
        con.execute(text(f"create temporary table ff_stg as {body}"))
        try:
            con.execute(text(f"delete from {qrel} t using ff_stg s where {pred}"))
            con.execute(text(f"insert into {qrel} select * from ff_stg"))
        finally:
            con.execute(text("drop table if exists ff_stg"))

alter_table_sync_schema

alter_table_sync_schema(relation, select_sql, *, mode='append_new_columns')

Add new columns present in SELECT but missing on target (as text).

Source code in src/fastflowtransform/executors/postgres_exec.py

def alter_table_sync_schema(
    self, relation: str, select_sql: str, *, mode: str = "append_new_columns"
) -> None:
    """
    Add new columns present in SELECT but missing on target (as text).
    """
    body = self._extract_select_like(select_sql)
    qrel = self._qualified(relation)
    with self.engine.begin() as con:
        self._set_search_path(con)
        cols = [r[0] for r in con.execute(text(f"select * from ({body}) q limit 0"))]
        existing = {
            r[0]
            for r in con.execute(
                text(
                    "select column_name from information_schema.columns "
                    "where table_schema = current_schema() and lower(table_name)=lower(:t)"
                ),
                {"t": relation},
            ).fetchall()
        }
        add = [c for c in cols if c not in existing]
        for c in add:
            con.execute(text(f'alter table {qrel} add column "{c}" text'))

SnowflakeSnowparkExecutor

Bases: BaseExecutor[DataFrame]

Source code in src/fastflowtransform/executors/snowflake_snowpark_exec.py

class SnowflakeSnowparkExecutor(BaseExecutor[SNDF]):
    ENGINE_NAME = "snowflake_snowpark"
    """Snowflake executor operating on Snowpark DataFrames (no pandas)."""

    def __init__(self, cfg: dict):
        # cfg: {account, user, password, warehouse, database, schema, role?}
        self.session = Session.builder.configs(cfg).create()
        self.database = cfg["database"]
        self.schema = cfg["schema"]
        # Provide a tiny testing shim so tests can call executor.con.execute("SQL")
        self.con = _SFCursorShim(self.session)

    # ---------- Helpers ----------
    def _q(self, s: str) -> str:
        return '"' + s.replace('"', '""') + '"'

    def _qualified(self, rel: str) -> str:
        # "DB"."SCHEMA"."TABLE"
        return f"{self._q(self.database)}.{self._q(self.schema)}.{self._q(rel)}"

    # ---------- Frame-Hooks ----------
    def _read_relation(self, relation: str, node: Node, deps: Iterable[str]) -> SNDF:
        return self.session.table(self._qualified(relation))

    def _materialize_relation(self, relation: str, df: SNDF, node: Node) -> None:
        if not self._is_frame(df):
            raise TypeError("Snowpark model must return a Snowpark DataFrame")
        df.write.save_as_table(self._qualified(relation), mode="overwrite")

    def _create_view_over_table(self, view_name: str, backing_table: str, node: Node) -> None:
        qv = self._qualified(view_name)
        qb = self._qualified(backing_table)
        self.session.sql(f"CREATE OR REPLACE VIEW {qv} AS SELECT * FROM {qb}").collect()

    def _validate_required(
        self, node_name: str, inputs: Any, requires: dict[str, set[str]]
    ) -> None:
        if not requires:
            return

        def cols(df: SNDF) -> set[str]:
            # Snowpark: schema names
            return set(df.schema.names)

        errors: list[str] = []
        # Single dependency
        if isinstance(inputs, SNDF):
            need = next(iter(requires.values()), set())
            missing = need - cols(inputs)
            if missing:
                errors.append(f"- missing columns: {sorted(missing)} | have={sorted(cols(inputs))}")
        else:
            # Multiple dependencies
            for rel, need in requires.items():
                if rel not in inputs:
                    errors.append(f"- missing dependency key '{rel}'")
                    continue
                missing = need - cols(inputs[rel])
                if missing:
                    errors.append(
                        f"- [{rel}] missing: {sorted(missing)} | have={sorted(cols(inputs[rel]))}"
                    )

        if errors:
            raise ValueError(
                "Required columns check failed for Snowpark model "
                f"'{node_name}'.\n" + "\n".join(errors)
            )

    def _columns_of(self, frame: SNDF) -> list[str]:
        return list(frame.schema.names)

    def _is_frame(self, obj: Any) -> bool:
        return isinstance(obj, SNDF)

    def _frame_name(self) -> str:
        return "Snowpark"

    # ---- SQL hooks ----
    def _format_relation_for_ref(self, name: str) -> str:
        return self._qualified(relation_for(name))

    def _format_source_reference(
        self, cfg: dict[str, Any], source_name: str, table_name: str
    ) -> str:
        if cfg.get("location"):
            raise NotImplementedError("Snowflake executor does not support path-based sources.")

        ident = cfg.get("identifier")
        if not ident:
            raise KeyError(f"Source {source_name}.{table_name} missing identifier")

        db = cfg.get("database") or cfg.get("catalog") or self.database
        sch = cfg.get("schema") or self.schema
        if not db or not sch:
            raise KeyError(
                f"Source {source_name}.{table_name} missing database/schema for Snowflake"
            )
        return f"{self._q(db)}.{self._q(sch)}.{self._q(ident)}"

    def _create_or_replace_view(self, target_sql: str, select_body: str, node: Node) -> None:
        self.session.sql(f"CREATE OR REPLACE VIEW {target_sql} AS {select_body}").collect()

    def _create_or_replace_table(self, target_sql: str, select_body: str, node: Node) -> None:
        self.session.sql(f"CREATE OR REPLACE TABLE {target_sql} AS {select_body}").collect()

    def _create_or_replace_view_from_table(
        self, view_name: str, backing_table: str, node: Node
    ) -> None:
        view_id = self._qualified(view_name)
        back_id = self._qualified(backing_table)
        self.session.sql(f"CREATE OR REPLACE VIEW {view_id} AS SELECT * FROM {back_id}").collect()

    # ---- Meta hook ----
    def on_node_built(self, node: Node, relation: str, fingerprint: str) -> None:
        """After successful materialization, upsert _ff_meta (best-effort)."""
        try:
            ensure_meta_table(self)
            upsert_meta(self, node.name, relation, fingerprint, "snowflake_snowpark")
        except Exception:
            pass

    # ── Incremental API (parity with DuckDB/PG) ──────────────────────────
    def exists_relation(self, relation: str) -> bool:
        """Check existence via information_schema.tables."""
        db = self._q(self.database)
        q = f"""
          select 1
          from {db}.information_schema.tables
          where table_schema = {self._q(self.schema)}
            and lower(table_name) = lower({self._q(relation)})
          limit 1
        """
        try:
            return bool(self.session.sql(q).collect())
        except Exception:
            return False

    def create_table_as(self, relation: str, select_sql: str) -> None:
        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        self.session.sql(f"CREATE OR REPLACE TABLE {self._qualified(relation)} AS {body}").collect()

    def incremental_insert(self, relation: str, select_sql: str) -> None:
        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        self.session.sql(f"INSERT INTO {self._qualified(relation)} {body}").collect()

    def incremental_merge(self, relation: str, select_sql: str, unique_key: list[str]) -> None:
        """
        Portable fallback without explicit column list:
          - WITH src AS (<body>)
          - DELETE ... USING src ...
          - INSERT ... SELECT * FROM src
        This avoids Snowflake MERGE column listing complexity.
        """
        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        pred = " AND ".join([f"t.{k}=s.{k}" for k in unique_key]) or "FALSE"
        qrel = self._qualified(relation)
        sql = f"""
        WITH src AS ({body})
        DELETE FROM {qrel} AS t USING src AS s WHERE {pred};
        INSERT INTO {qrel} SELECT * FROM src;
        """
        self.session.sql(sql).collect()

    def alter_table_sync_schema(
        self, relation: str, select_sql: str, *, mode: str = "append_new_columns"
    ) -> None:
        """
        Best-effort additive schema sync:
          - infer SELECT schema via LIMIT 0
          - add missing columns as STRING
        """
        if mode not in {"append_new_columns", "sync_all_columns"}:
            return
        qrel = self._qualified(relation)
        try:
            existing = {
                r[0]
                for r in self.session.sql(
                    f"""
                select column_name
                from {self._q(self.database)}.information_schema.columns
                where table_schema={self._q(self.schema)}
                  and lower(table_name)=lower({self._q(relation)})
                """
                ).collect()
            }
        except Exception:
            existing = set()
        # Probe SELECT columns
        body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
        probe = self.session.sql(f"SELECT * FROM ({body}) q WHERE 1=0")
        probe_cols = list(probe.schema.names)
        to_add = [c for c in probe_cols if c not in existing]
        if not to_add:
            return
        cols_sql = ", ".join(f"{self._q(c)} STRING" for c in to_add)
        self.session.sql(f"ALTER TABLE {qrel} ADD COLUMN {cols_sql}").collect()

ENGINE_NAME class-attribute instance-attribute

ENGINE_NAME = 'snowflake_snowpark'

Snowflake executor operating on Snowpark DataFrames (no pandas).

run_sql

run_sql(node, env)

Orchestrate SQL models

1) Render Jinja (ref/source/this) and strip leading {{ config(...) }}. 2) If the SQL is full DDL (CREATE …), execute it verbatim (passthrough). 3) Otherwise, normalize to CREATE OR REPLACE {TABLE|VIEW} AS . The body is CTE-aware (keeps WITH … SELECT … intact).

On failure, raise ModelExecutionError with a helpful snippet.

Source code in src/fastflowtransform/executors/base.py

def run_sql(self, node: Node, env: Environment) -> None:
    """
    Orchestrate SQL models:
      1) Render Jinja (ref/source/this) and strip leading {{ config(...) }}.
      2) If the SQL is full DDL (CREATE …), execute it verbatim (passthrough).
      3) Otherwise, normalize to CREATE OR REPLACE {TABLE|VIEW} AS <body>.
         The body is CTE-aware (keeps WITH … SELECT … intact).
    On failure, raise ModelExecutionError with a helpful snippet.
    """
    sql_rendered = self.render_sql(
        node,
        env,
        ref_resolver=lambda name: self._resolve_ref(name, env),
        source_resolver=self._resolve_source,
    )
    sql = self._strip_leading_config(sql_rendered).strip()

    materialization = (node.meta or {}).get("materialized", "table")
    if materialization == "ephemeral":
        return

    # 1) Direct DDL passthrough (CREATE [OR REPLACE] {TABLE|VIEW} …)
    if self._looks_like_direct_ddl(sql):
        try:
            self._execute_sql_direct(sql, node)
            return
        except NotImplementedError:
            # Engine doesn't implement direct DDL → fall back to normalized materialization.
            pass
        except Exception as e:
            raise ModelExecutionError(
                node_name=node.name,
                relation=relation_for(node.name),
                message=str(e),
                sql_snippet=sql,
            ) from e

    # 2) Normalized materialization path (CTE-safe body)
    body = self._selectable_body(sql).rstrip(" ;\n\t")
    target_sql = self._format_relation_for_ref(node.name)

    # Centralized SQL preview logging (applies to ALL engines)
    preview = (
        f"=== MATERIALIZE ===\n"
        f"-- model: {node.name}\n"
        f"-- materialized: {materialization}\n"
        f"-- target: {target_sql}\n"
        f"{body}\n"
    )
    echo_debug(preview)

    try:
        self._apply_sql_materialization(node, target_sql, body, materialization)
    except Exception as e:
        preview = f"-- materialized={materialization}\n-- target={target_sql}\n{body}"
        raise ModelExecutionError(
            node_name=node.name,
            relation=relation_for(node.name),
            message=str(e),
            sql_snippet=preview,
        ) from e

on_node_built

on_node_built(node, relation, fingerprint)

After successful materialization, upsert _ff_meta (best-effort).

Source code in src/fastflowtransform/executors/snowflake_snowpark_exec.py

def on_node_built(self, node: Node, relation: str, fingerprint: str) -> None:
    """After successful materialization, upsert _ff_meta (best-effort)."""
    try:
        ensure_meta_table(self)
        upsert_meta(self, node.name, relation, fingerprint, "snowflake_snowpark")
    except Exception:
        pass

exists_relation

exists_relation(relation)

Check existence via information_schema.tables.

Source code in src/fastflowtransform/executors/snowflake_snowpark_exec.py

def exists_relation(self, relation: str) -> bool:
    """Check existence via information_schema.tables."""
    db = self._q(self.database)
    q = f"""
      select 1
      from {db}.information_schema.tables
      where table_schema = {self._q(self.schema)}
        and lower(table_name) = lower({self._q(relation)})
      limit 1
    """
    try:
        return bool(self.session.sql(q).collect())
    except Exception:
        return False

incremental_merge

incremental_merge(relation, select_sql, unique_key)

Portable fallback without explicit column list

• WITH src AS ()
• DELETE ... USING src ...
• INSERT ... SELECT * FROM src

This avoids Snowflake MERGE column listing complexity.

Source code in src/fastflowtransform/executors/snowflake_snowpark_exec.py

def incremental_merge(self, relation: str, select_sql: str, unique_key: list[str]) -> None:
    """
    Portable fallback without explicit column list:
      - WITH src AS (<body>)
      - DELETE ... USING src ...
      - INSERT ... SELECT * FROM src
    This avoids Snowflake MERGE column listing complexity.
    """
    body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
    pred = " AND ".join([f"t.{k}=s.{k}" for k in unique_key]) or "FALSE"
    qrel = self._qualified(relation)
    sql = f"""
    WITH src AS ({body})
    DELETE FROM {qrel} AS t USING src AS s WHERE {pred};
    INSERT INTO {qrel} SELECT * FROM src;
    """
    self.session.sql(sql).collect()

alter_table_sync_schema

alter_table_sync_schema(relation, select_sql, *, mode='append_new_columns')

Best-effort additive schema sync

• infer SELECT schema via LIMIT 0
• add missing columns as STRING

Source code in src/fastflowtransform/executors/snowflake_snowpark_exec.py

def alter_table_sync_schema(
    self, relation: str, select_sql: str, *, mode: str = "append_new_columns"
) -> None:
    """
    Best-effort additive schema sync:
      - infer SELECT schema via LIMIT 0
      - add missing columns as STRING
    """
    if mode not in {"append_new_columns", "sync_all_columns"}:
        return
    qrel = self._qualified(relation)
    try:
        existing = {
            r[0]
            for r in self.session.sql(
                f"""
            select column_name
            from {self._q(self.database)}.information_schema.columns
            where table_schema={self._q(self.schema)}
              and lower(table_name)=lower({self._q(relation)})
            """
            ).collect()
        }
    except Exception:
        existing = set()
    # Probe SELECT columns
    body = self._first_select_body(select_sql).strip().rstrip(";\n\t ")
    probe = self.session.sql(f"SELECT * FROM ({body}) q WHERE 1=0")
    probe_cols = list(probe.schema.names)
    to_add = [c for c in probe_cols if c not in existing]
    if not to_add:
        return
    cols_sql = ", ".join(f"{self._q(c)} STRING" for c in to_add)
    self.session.sql(f"ALTER TABLE {qrel} ADD COLUMN {cols_sql}").collect()