fastflowtransform.executors.bigquery.pandas¶

BigQueryExecutor ¶

Bases: BigQueryBaseExecutor[DataFrame]

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

class BigQueryExecutor(BigQueryBaseExecutor[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,
        allow_create_dataset: bool = False,
    ):
        super().__init__(
            project=project,
            dataset=dataset,
            location=location,
            client=client,
            allow_create_dataset=allow_create_dataset,
        )

    # ---------- 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.)
        start = perf_counter()
        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
        else:
            duration_ms = int((perf_counter() - start) * 1000)
            self._record_dataframe_stats(df, duration_ms)

    def _create_view_over_table(self, view_name: str, backing_table: str, node: Node) -> None:
        """
        Convenience helper for a simple view on top of a backing table.
        """
        # Delegate to the shared base implementation
        self._create_or_replace_view_from_table(view_name, backing_table, node)

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

    def _record_dataframe_stats(self, df: pd.DataFrame, duration_ms: int) -> None:
        rows = len(df)
        bytes_val = int(df.memory_usage(deep=True).sum()) if rows > 0 else 0
        self._record_query_stats(
            QueryStats(
                bytes_processed=bytes_val if bytes_val > 0 else None,
                rows=rows if rows > 0 else None,
                duration_ms=duration_ms,
            )
        )

        # ---- Unit-test helpers (pandas) ---------------------------------------

    def utest_read_relation(self, relation: str) -> pd.DataFrame:
        """
        Read a relation into a pandas DataFrame for unit-test assertions.
        """
        q = f"SELECT * FROM {self._qualified_identifier(relation)}"
        job = self.client.query(q, location=self.location)
        # Same convention as _read_relation: use BigQuery Storage if available
        return job.result().to_dataframe(create_bqstorage_client=True)

    def utest_load_relation_from_rows(self, relation: str, rows: list[dict]) -> None:
        """
        Load rows into a BigQuery table for unit tests (replace if exists).
        """
        self._ensure_dataset()
        table_id = f"{self.project}.{self.dataset}.{relation}"
        df = pd.DataFrame(rows)

        job_config = LoadJobConfig(write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE)

        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 utest write failed: {table_id}\n{e}") from e

    def utest_clean_target(self, relation: str) -> None:
        """
        For unit tests: drop any table/view with this name in the configured dataset.
        """
        table_id = f"{self.project}.{self.dataset}.{relation}"
        # BigQuery treats views & tables both as "tables" for deletion.
        try:
            # not_found_ok=True is available on the real client; our typing alias
            # should be compatible - if not, just ignore NotFound below.
            self.client.delete_table(table_id, not_found_ok=True)
        except NotFound:
            pass
        except TypeError:
            # For older client versions without not_found_ok, fall back:
            with suppress(NotFound):
                self.client.delete_table(table_id)
        except Exception:
            # Cleanup is best-effort in utests.
            pass

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)

utest_read_relation ¶

utest_read_relation(relation)

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

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

def utest_read_relation(self, relation: str) -> pd.DataFrame:
    """
    Read a relation into a pandas DataFrame for unit-test assertions.
    """
    q = f"SELECT * FROM {self._qualified_identifier(relation)}"
    job = self.client.query(q, location=self.location)
    # Same convention as _read_relation: use BigQuery Storage if available
    return job.result().to_dataframe(create_bqstorage_client=True)

utest_load_relation_from_rows ¶

utest_load_relation_from_rows(relation, rows)

Load rows into a BigQuery table for unit tests (replace if exists).

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

def utest_load_relation_from_rows(self, relation: str, rows: list[dict]) -> None:
    """
    Load rows into a BigQuery table for unit tests (replace if exists).
    """
    self._ensure_dataset()
    table_id = f"{self.project}.{self.dataset}.{relation}"
    df = pd.DataFrame(rows)

    job_config = LoadJobConfig(write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE)

    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 utest write failed: {table_id}\n{e}") from e

utest_clean_target ¶

utest_clean_target(relation)

For unit tests: drop any table/view with this name in the configured dataset.

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

def utest_clean_target(self, relation: str) -> None:
    """
    For unit tests: drop any table/view with this name in the configured dataset.
    """
    table_id = f"{self.project}.{self.dataset}.{relation}"
    # BigQuery treats views & tables both as "tables" for deletion.
    try:
        # not_found_ok=True is available on the real client; our typing alias
        # should be compatible - if not, just ignore NotFound below.
        self.client.delete_table(table_id, not_found_ok=True)
    except NotFound:
        pass
    except TypeError:
        # For older client versions without not_found_ok, fall back:
        with suppress(NotFound):
            self.client.delete_table(table_id)
    except Exception:
        # Cleanup is best-effort in utests.
        pass

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)