fastflowtransform.executors.bigquery.bigframes

BigQueryBFExecutor

Bases: BigQueryBaseExecutor[BFDataFrame]

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

class BigQueryBFExecutor(BigQueryBaseExecutor[BFDataFrame]):
    ENGINE_NAME: str = "bigquery"
    runtime_contracts: BigQueryRuntimeContracts

    def __init__(
        self,
        project: str,
        dataset: str,
        location: str | None = None,
        allow_create_dataset: bool = False,
    ):
        if not project:
            raise RuntimeError("BigFrames executor requires FF_BQ_PROJECT to be set.")
        if not location:
            raise RuntimeError(
                "BigFrames executor requires FF_BQ_LOCATION to be set. "
                "Use the dataset's region (e.g., EU or US)."
            )
        super().__init__(
            project=project,
            dataset=dataset,
            location=location,
            allow_create_dataset=allow_create_dataset,
        )
        self.runtime_contracts = BigQueryRuntimeContracts(self)

        try:
            ctx = BigQueryOptions(
                project=project,
                location=location,
            )
            self.session = bigframes.Session(context=ctx)
        except Exception as exc:
            raise RuntimeError(
                "Failed to initialize BigFrames session. Verify FF_BQ_PROJECT, "
                "FF_BQ_DATASET, and FF_BQ_LOCATION are set for the active profile. "
                f"{exc}"
            ) from exc

    def run_python(self, node: Node) -> None:
        """
        Execute Python models with a session scoped to this executor.

        We avoid mutating the process-wide default session; instead we
        temporarily set the executor session as the active global session so
        model code using bpd.DataFrame(...) picks up the configured location,
        then restore afterward.
        """
        ctx = bf_global_session._GlobalSessionContext(self.session)
        with ctx:
            super().run_python(node)

    # ---------- Python (Frames) ----------
    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:
        self._ensure_dataset()
        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(...)."
        )

    # ---- Required-columns validation tuned for BigFrames ----
    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):
            # Single input frame case
            need = next(iter(requires.values()), set())
            miss = need - cols(inputs)
            if miss:
                errs.append(f"- missing columns: {sorted(miss)}")
        else:
            # Mapping {rel -> frame}
            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:
        if obj is None:
            return False
        return (
            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)"

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

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

        Even though this executor uses BigFrames for normal execution,
        utests compare pandas DataFrames, so we convert.
        """
        q = f"SELECT * FROM {self._qualified_identifier(relation)}"
        job = self.client.query(q, location=self.location)
        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).

        Implementation uses the raw BigQuery client with pandas, which is
        perfectly fine for test input setup.
        """
        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}"

        try:
            self.client.delete_table(table_id, not_found_ok=True)
        except NotFound:
            pass
        except TypeError:
            with suppress(NotFound):
                self.client.delete_table(table_id)
        except Exception:
            # Best-effort; don't make the whole test run fail because cleanup hiccupped.
            pass

run_python

run_python(node)

Execute Python models with a session scoped to this executor.

We avoid mutating the process-wide default session; instead we temporarily set the executor session as the active global session so model code using bpd.DataFrame(...) picks up the configured location, then restore afterward.

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

def run_python(self, node: Node) -> None:
    """
    Execute Python models with a session scoped to this executor.

    We avoid mutating the process-wide default session; instead we
    temporarily set the executor session as the active global session so
    model code using bpd.DataFrame(...) picks up the configured location,
    then restore afterward.
    """
    ctx = bf_global_session._GlobalSessionContext(self.session)
    with ctx:
        super().run_python(node)

utest_read_relation

utest_read_relation(relation)

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

Even though this executor uses BigFrames for normal execution, utests compare pandas DataFrames, so we convert.

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

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

    Even though this executor uses BigFrames for normal execution,
    utests compare pandas DataFrames, so we convert.
    """
    q = f"SELECT * FROM {self._qualified_identifier(relation)}"
    job = self.client.query(q, location=self.location)
    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).

Implementation uses the raw BigQuery client with pandas, which is perfectly fine for test input setup.

Source code in src/fastflowtransform/executors/bigquery/bigframes.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).

    Implementation uses the raw BigQuery client with pandas, which is
    perfectly fine for test input setup.
    """
    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/bigframes.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}"

    try:
        self.client.delete_table(table_id, not_found_ok=True)
    except NotFound:
        pass
    except TypeError:
        with suppress(NotFound):
            self.client.delete_table(table_id)
    except Exception:
        # Best-effort; don't make the whole test run fail because cleanup hiccupped.
        pass

configure_contracts

configure_contracts(contracts, project_contracts)

Inject parsed contracts into this executor instance. The run engine should call this once at startup.

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

def configure_contracts(
    self,
    contracts: Mapping[str, ContractsFileModel] | None,
    project_contracts: ProjectContractsModel | None,
) -> None:
    """
    Inject parsed contracts into this executor instance.
    The run engine should call this once at startup.
    """
    self._ff_contracts = contracts or {}
    self._ff_project_contracts = project_contracts

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:
        runtime = getattr(self, "runtime_contracts", None)
        # contracts only for TABLE materialization for now
        if runtime is not None and materialization == "table":
            contracts = getattr(self, "_ff_contracts", {}) or {}
            project_contracts = getattr(self, "_ff_project_contracts", None)

            # keying: prefer the logical table name (contracts.table),
            # but node.name or relation_for(node.name) is usually what you want.
            logical_name = relation_for(node.name)
            contract = contracts.get(logical_name) or contracts.get(node.name)

            ctx = runtime.build_context(
                node=node,
                relation=logical_name,
                physical_table=target_sql,
                contract=contract,
                project_contracts=project_contracts,
                is_incremental=self._meta_is_incremental(meta),
            )
            # Engine-specific enforcement (verify/cast/off)
            runtime.apply_sql_contracts(ctx=ctx, select_body=body)
        else:
            # Old behavior
            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

execute_test_sql

execute_test_sql(stmt)

Execute lightweight SQL for DQ tests using the BigQuery client.

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

def execute_test_sql(self, stmt: Any) -> Any:
    """
    Execute lightweight SQL for DQ tests using the BigQuery client.
    """

    def _infer_param_type(value: Any) -> str:
        if isinstance(value, bool):
            return "BOOL"
        if isinstance(value, int) and not isinstance(value, bool):
            return "INT64"
        if isinstance(value, float):
            return "FLOAT64"
        return "STRING"

    def _run_job(sql: str, params: dict[str, Any] | None = None) -> Any:
        job_config = bigquery.QueryJobConfig()
        if self.dataset:
            job_config.default_dataset = bigquery.DatasetReference(self.project, self.dataset)
        if params:
            job_config.query_parameters = [
                bigquery.ScalarQueryParameter(k, _infer_param_type(v), v)
                for k, v in params.items()
            ]
        return self.client.query(sql, job_config=job_config, location=self.location)

    def _run_one(s: Any) -> Any:
        statement_len = 2
        if (
            isinstance(s, tuple)
            and len(s) == statement_len
            and isinstance(s[0], str)
            and isinstance(s[1], dict)
        ):
            return _run_job(s[0], s[1]).result()
        if isinstance(s, str):
            # Use guarded execution path for simple statements
            return self._execute_sql(s).result()
        if isinstance(s, Iterable) and not isinstance(s, (bytes, bytearray, str)):
            res = None
            for item in s:
                res = _run_one(item)
            return res
        return _run_job(str(s)).result()

    return make_fetchable(_run_one(stmt))

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,
    }

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()

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_basic(f"ALTER TABLE {target} ADD COLUMN {col} {typ}").result()

normalize_physical_type

normalize_physical_type(t)

Canonicalize a physical type string for comparisons (DQ + contracts).

Default: just strip + lower. Engines may override to account for dialect quirks in information_schema (e.g. Postgres timestamp variants, Snowflake VARCHAR(…) / NUMBER(…)).

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

def normalize_physical_type(self, t: str | None) -> str:
    """
    Canonicalize a physical type string for comparisons (DQ + contracts).

    Default: just strip + lower.
    Engines may override to account for dialect quirks in information_schema
    (e.g. Postgres timestamp variants, Snowflake VARCHAR(…) / NUMBER(…)).
    """
    return (t or "").strip().lower()

collect_docs_columns

collect_docs_columns()

Column metadata for docs (project+dataset scoped).

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

def collect_docs_columns(self) -> dict[str, list[ColumnInfo]]:
    """
    Column metadata for docs (project+dataset scoped).
    """
    sql = f"""
    select table_name, column_name, data_type, is_nullable
    from `{self.project}.{self.dataset}.INFORMATION_SCHEMA.COLUMNS`
    order by table_name, ordinal_position
    """
    try:
        job = self.client.query(
            sql,
            job_config=bigquery.QueryJobConfig(
                default_dataset=bigquery.DatasetReference(self.project, self.dataset)
            ),
            location=self.location,
        )
        rows = list(job.result())
    except Exception:
        return {}

    out: dict[str, list[ColumnInfo]] = {}
    for row in rows:
        table = str(row["table_name"])
        col = str(row["column_name"])
        dtype = str(row["data_type"])
        nullable = str(row["is_nullable"]).upper() == "YES"
        out.setdefault(table, []).append(ColumnInfo(col, dtype, nullable))
    return out