fastflowtransform.executors.bigquery.base

BigQueryBaseExecutor

Bases: BigQueryIdentifierMixin, SnapshotSqlMixin, BaseExecutor[TFrame]

Shared BigQuery executor logic (SQL, incremental, meta, DQ helpers).

Subclasses are responsible for

• frame type (pandas / BigFrames / ...)
• _read_relation()
• _materialize_relation()
• _is_frame()
• _frame_name()

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

class BigQueryBaseExecutor(BigQueryIdentifierMixin, SnapshotSqlMixin, BaseExecutor[TFrame]):
    """
    Shared BigQuery executor logic (SQL, incremental, meta, DQ helpers).

    Subclasses are responsible for:
      - frame type (pandas / BigFrames / ...)
      - _read_relation()
      - _materialize_relation()
      - _is_frame()
      - _frame_name()
    """

    # Subclasses override ENGINE_NAME ("bigquery", "bigquery_batch", ...)
    ENGINE_NAME = "bigquery_base"
    _BUDGET_GUARD = BudgetGuard(
        env_var="FF_BQ_MAX_BYTES",
        estimator_attr="_estimate_query_bytes",
        engine_label="BigQuery",
        what="query",
    )

    def __init__(
        self,
        project: str,
        dataset: str,
        location: str | None = None,
        client: Client | None = None,
        allow_create_dataset: bool = False,
    ):
        self.project = project
        self.dataset = dataset
        self.location = location
        self.allow_create_dataset = allow_create_dataset
        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,
        )

    def _execute_sql(self, sql: str) -> _TrackedQueryJob:
        """
        Central BigQuery query runner.

        - All 'real' SQL statements in this executor should go through here.
        - Returns the QueryJob so callers can call .result().
        """

        def _exec() -> _TrackedQueryJob:
            job_config = bigquery.QueryJobConfig()
            if self.dataset:
                # Let unqualified tables resolve to project.dataset.table
                job_config.default_dataset = bigquery.DatasetReference(self.project, self.dataset)

            job = self.client.query(
                sql,
                job_config=job_config,
                location=self.location,
            )
            return _TrackedQueryJob(job, on_complete=self._record_query_job_stats)

        return run_sql_with_budget(
            self,
            sql,
            guard=self._BUDGET_GUARD,
            exec_fn=_exec,
            estimate_fn=self._estimate_query_bytes,
            record_stats=False,
        )

    # --- Cost estimation for the shared BudgetGuard -----------------

    def _estimate_query_bytes(self, sql: str) -> int | None:
        """
        Estimate bytes for a BigQuery SQL statement using a dry-run.

        Returns the estimated bytes, or None if estimation is not possible.
        """
        cfg = bigquery.QueryJobConfig(
            dry_run=True,
            use_query_cache=False,
        )
        if self.dataset:
            # Let unqualified tables resolve to project.dataset.table
            cfg.default_dataset = bigquery.DatasetReference(self.project, self.dataset)

        job = self.client.query(
            sql,
            job_config=cfg,
            location=self.location,
        )
        # Dry-run is free; we just need the job metadata
        job.result()
        return int(getattr(job, "total_bytes_processed", 0) or 0)

    # ---- DQ test table formatting (fft test) ----
    def _format_test_table(self, table: str | None) -> str | None:
        """
        Ensure tests use fully-qualified BigQuery identifiers in fft test.
        """
        table = super()._format_test_table(table)
        if not isinstance(table, str) or not table.strip():
            return table
        return self._qualified_identifier(table.strip())

    # ---- SQL hooks ----
    def _this_identifier(self, node: Node) -> str:
        """
        Ensure {{ this }} renders as a fully-qualified identifier so BigQuery
        incremental SQL (e.g., subqueries against {{ this }}) includes project
        and dataset.
        """
        return self._qualify_identifier(relation_for(node.name))

    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._execute_sql(f"CREATE OR REPLACE VIEW {target_sql} AS {select_body}").result()

    def _create_or_replace_table(self, target_sql: str, select_body: str, node: Node) -> None:
        self._execute_sql(f"CREATE OR REPLACE TABLE {target_sql} AS {select_body}").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()
        self._execute_sql(f"CREATE OR REPLACE VIEW {view_id} AS SELECT * FROM {back_id}").result()

    # ---- Snapshot mixin hooks ----
    def _snapshot_prepare_target(self) -> None:
        self._ensure_dataset()

    def _snapshot_target_identifier(self, rel_name: str) -> str:
        return self._qualified_identifier(rel_name)

    def _snapshot_current_timestamp(self) -> str:
        return "CURRENT_TIMESTAMP()"

    def _snapshot_null_timestamp(self) -> str:
        return "CAST(NULL AS TIMESTAMP)"

    def _snapshot_null_hash(self) -> str:
        return "CAST(NULL AS STRING)"

    def _snapshot_hash_expr(self, check_cols: list[str], src_alias: str) -> str:
        concat_expr = self._snapshot_concat_expr(check_cols, src_alias)
        return f"TO_HEX(MD5({concat_expr}))"

    def _snapshot_cast_as_string(self, expr: str) -> str:
        return f"CAST({expr} AS STRING)"

    # ---- Meta hook ----
    def on_node_built(self, node: Node, relation: str, fingerprint: str) -> None:
        """
        Write/update dataset._ff_meta after a successful build.
        Both pandas + BigFrames executors use the logical engine key 'bigquery'.
        """
        ensure_meta_table(self)
        upsert_meta(self, node.name, relation, fingerprint, "bigquery")

    # ── Incremental API (shared across BigQuery executors) ───────────────
    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._selectable_body(select_sql).strip().rstrip(";\n\t ")
        target = self._qualified_identifier(
            relation,
            project=self.project,
            dataset=self.dataset,
        )
        self._execute_sql(f"CREATE TABLE {target} AS {body}").result()

    def incremental_insert(self, relation: str, select_sql: str) -> None:
        """
        INSERT INTO with cleaned SELECT body.
        """
        self._ensure_dataset()
        body = self._selectable_body(select_sql).strip().rstrip(";\n\t ")
        target = self._qualified_identifier(
            relation,
            project=self.project,
            dataset=self.dataset,
        )
        self._execute_sql(f"INSERT INTO {target} {body}").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._selectable_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._execute_sql(delete_sql).result()

        insert_sql = f"INSERT INTO {target} SELECT * FROM ({body})"
        self._execute_sql(insert_sql).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._selectable_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._execute_sql(f"ALTER TABLE {target} ADD COLUMN {col} {typ}").result()

    # ── Snapshots API (shared for pandas + BigFrames) ─────────────────────

    def execute_hook_sql(self, sql: str) -> None:
        """
        Execute one SQL statement for pre/post/on_run hooks.
        """
        self._execute_sql(sql).result()

on_node_built

on_node_built(node, relation, fingerprint)

Write/update dataset._ff_meta after a successful build. Both pandas + BigFrames executors use the logical engine key 'bigquery'.

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

def on_node_built(self, node: Node, relation: str, fingerprint: str) -> None:
    """
    Write/update dataset._ff_meta after a successful build.
    Both pandas + BigFrames executors use the logical engine key 'bigquery'.
    """
    ensure_meta_table(self)
    upsert_meta(self, node.name, relation, fingerprint, "bigquery")

exists_relation

exists_relation(relation)

Check presence in INFORMATION_SCHEMA for tables/views.

Source code in src/fastflowtransform/executors/bigquery/base.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/base.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._selectable_body(select_sql).strip().rstrip(";\n\t ")
    target = self._qualified_identifier(
        relation,
        project=self.project,
        dataset=self.dataset,
    )
    self._execute_sql(f"CREATE TABLE {target} AS {body}").result()

incremental_insert

incremental_insert(relation, select_sql)

INSERT INTO with cleaned SELECT body.

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

def incremental_insert(self, relation: str, select_sql: str) -> None:
    """
    INSERT INTO with cleaned SELECT body.
    """
    self._ensure_dataset()
    body = self._selectable_body(select_sql).strip().rstrip(";\n\t ")
    target = self._qualified_identifier(
        relation,
        project=self.project,
        dataset=self.dataset,
    )
    self._execute_sql(f"INSERT INTO {target} {body}").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/base.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._selectable_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._execute_sql(delete_sql).result()

    insert_sql = f"INSERT INTO {target} SELECT * FROM ({body})"
    self._execute_sql(insert_sql).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/base.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._selectable_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._execute_sql(f"ALTER TABLE {target} ADD COLUMN {col} {typ}").result()

execute_hook_sql

execute_hook_sql(sql)

Execute one SQL statement for pre/post/on_run hooks.

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

def execute_hook_sql(self, sql: str) -> None:
    """
    Execute one SQL statement for pre/post/on_run hooks.
    """
    self._execute_sql(sql).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.
    """
    meta = getattr(node, "meta", {}) or {}
    if self._meta_is_incremental(meta):
        # Delegates to incremental engine: render, schema sync, merge/insert, etc.
        return _ff_incremental.run_or_dispatch(self, node, env)

    if self._meta_is_snapshot(meta):
        # Snapshots are executed via the dedicated CLI: `fft snapshot run`.
        raise ModelExecutionError(
            node_name=node.name,
            relation=relation_for(node.name),
            message=(
                "Snapshot models cannot be executed via 'fft run'. "
                "Use 'fft snapshot run' instead."
            ),
            sql_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

configure_query_budget_limit

configure_query_budget_limit(limit)

Inject a configured per-query byte limit (e.g. from budgets.yml).

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

def configure_query_budget_limit(self, limit: int | None) -> None:
    """
    Inject a configured per-query byte limit (e.g. from budgets.yml).
    """
    if limit is None:
        self._ff_configured_query_limit = None
        return
    try:
        iv = int(limit)
    except Exception:
        self._ff_configured_query_limit = None
        return
    self._ff_configured_query_limit = iv if iv > 0 else None

reset_node_stats

reset_node_stats()

Reset per-node statistics buffer.

The run engine calls this before executing a model so that all stats recorded via _record_query_stats(...) belong to that node.

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

def reset_node_stats(self) -> None:
    """
    Reset per-node statistics buffer.

    The run engine calls this before executing a model so that all
    stats recorded via `_record_query_stats(...)` belong to that node.
    """
    # just clear the buffer; next recording will re-create it
    self._ff_query_stats_buffer = []

get_node_stats

get_node_stats()

Aggregate buffered QueryStats into a simple dict:

{
  "bytes_scanned": <sum>,
  "rows": <sum>,
  "query_duration_ms": <sum>,
}

Called by the run engine after a node finishes.

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

def get_node_stats(self) -> dict[str, int]:
    """
    Aggregate buffered QueryStats into a simple dict:

        {
          "bytes_scanned": <sum>,
          "rows": <sum>,
          "query_duration_ms": <sum>,
        }

    Called by the run engine after a node finishes.
    """
    stats_list = self._drain_query_stats()
    if not stats_list:
        return {}

    total_bytes = 0
    total_rows = 0
    total_duration = 0

    for s in stats_list:
        if s.bytes_processed is not None:
            total_bytes += int(s.bytes_processed)
        if s.rows is not None:
            total_rows += int(s.rows)
        if s.duration_ms is not None:
            total_duration += int(s.duration_ms)

    return {
        "bytes_scanned": total_bytes,
        "rows": total_rows,
        "query_duration_ms": total_duration,
    }

run_python

run_python(node)

Execute the Python model for a given node and materialize its result.

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

def run_python(self, node: Node) -> None:
    """Execute the Python model for a given node and materialize its result."""
    func = REGISTRY.py_funcs[node.name]
    deps = REGISTRY.nodes[node.name].deps or []

    self._reset_http_ctx(node)

    # arg = self._build_python_args(node, deps)
    args, argmap = self._build_python_inputs(node, deps)
    requires = REGISTRY.py_requires.get(node.name, {})
    # if deps:
    #     self._validate_required(node.name, arg, requires)
    if deps:
        # Required-columns check works against the mapping
        self._validate_required(node.name, argmap, requires)

    # out = self._execute_python_func(func, arg, node)
    out = self._execute_python_func(func, args, node)

    target = relation_for(node.name)
    meta = getattr(node, "meta", {}) or {}
    mat = self._resolve_materialization_strategy(meta)

    if mat == "incremental":
        self._materialize_incremental(target, out, node, meta)
    elif mat == "view":
        self._materialize_view(target, out, node)
    else:
        self._materialize_relation(target, out, node)

    self._snapshot_http_ctx(node)

snapshot_prune

snapshot_prune(relation, unique_key, keep_last, *, dry_run=False)

Delete older snapshot versions while keeping the most recent keep_last rows per business key (including the current row).

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

    def snapshot_prune(
        self,
        relation: str,
        unique_key: list[str],
        keep_last: int,
        *,
        dry_run: bool = False,
    ) -> None:
        """
        Delete older snapshot versions while keeping the most recent `keep_last`
        rows per business key (including the current row).
        """
        ex = cast("BaseExecutor[Any]", self)

        if keep_last <= 0:
            return

        keys = [k for k in unique_key if k]
        if not keys:
            return

        target = self._snapshot_target_identifier(relation)
        vf = self.SNAPSHOT_VALID_FROM_COL  # type: ignore[attr-defined]

        key_select = ", ".join(keys)
        part_by = ", ".join(keys)

        ranked_sql = f"""
SELECT
  {key_select},
  {vf},
  ROW_NUMBER() OVER (
    PARTITION BY {part_by}
    ORDER BY {vf} DESC
  ) AS rn
FROM {target}
"""

        if dry_run:
            sql = f"""
WITH ranked AS (
  {ranked_sql}
)
SELECT COUNT(*) AS rows_to_delete
FROM ranked
WHERE rn > {int(keep_last)}
"""
            res = ex._execute_sql(sql)
            count = self._snapshot_fetch_count(res)
            echo(
                f"[DRY-RUN] snapshot_prune({relation}): would delete {count} row(s) "
                f"(keep_last={keep_last})"
            )
            return

        join_pred = " AND ".join([f"t.{k} = r.{k}" for k in keys])
        delete_sql = f"""
DELETE FROM {target} t
USING (
  {ranked_sql}
) r
WHERE
  r.rn > {int(keep_last)}
  AND {join_pred}
  AND t.{vf} = r.{vf}
"""
        ex._execute_sql(delete_sql)

utest_load_relation_from_rows

utest_load_relation_from_rows(relation, rows)

Load test input rows into a physical relation for unit tests.

Default: not implemented. Engines that support fft utest should override.

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

def utest_load_relation_from_rows(self, relation: str, rows: list[dict]) -> None:
    """
    Load test input rows into a physical relation for unit tests.

    Default: not implemented. Engines that support `fft utest` should override.
    """
    raise NotImplementedError(
        f"utest_load_relation_from_rows not implemented for engine '{self.engine_name}'."
    )

utest_read_relation

utest_read_relation(relation)

Read a physical relation into a pandas.DataFrame for unit-test assertions.

Default: not implemented. Engines that support fft utest should override.

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

def utest_read_relation(self, relation: str) -> _PDDataFrame:
    """
    Read a physical relation into a pandas.DataFrame for unit-test assertions.

    Default: not implemented. Engines that support `fft utest` should override.
    """
    raise NotImplementedError(
        f"utest_read_relation not implemented for engine '{self.engine_name}'."
    )

utest_clean_target

utest_clean_target(relation)

Best-effort cleanup hook before executing a unit-test model:

• Drop tables/views with the target name so view<->table flips cannot fail (DuckDB, Postgres, ...).
• This runs only in fft utest, and we already enforce that utest profiles use isolated DBs/schemas.

Default: no-op.

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

def utest_clean_target(self, relation: str) -> None:
    """
    Best-effort cleanup hook before executing a unit-test model:

    - Drop tables/views with the target name so view<->table flips
      cannot fail (DuckDB, Postgres, ...).
    - This runs *only* in `fft utest`, and we already enforce that
      utest profiles use isolated DBs/schemas.

    Default: no-op.
    """
    return