fastflowtransform.meta

Engine-aware metadata store and relation-existence helpers.

This module persists a per-engine _ff_meta table with the following columns: - node_name (PK where supported) - relation - fp - engine - built_at (server-side timestamp)

APIs

ensure_meta_table(executor) upsert_meta(executor, node_name, relation, fp, engine) get_meta(executor, node_name) -> tuple[str, str, object, str] | None relation_exists(executor, relation) -> bool

Supported engines

• DuckDB (executor.con)
• Postgres (executor.engine, optional .schema)
• BigQuery (executor.client, .dataset, optional .project)

ensure_meta_table

ensure_meta_table(executor)

Create the _ff_meta table if it does not exist for the active engine.

Source code in src/fastflowtransform/meta.py

def ensure_meta_table(executor: Any) -> None:
    """
    Create the _ff_meta table if it does not exist for the active engine.
    """
    if _is_duckdb(executor):
        sql = (
            'create table if not exists "_ff_meta" ('
            "  node_name text primary key,"
            "  relation text,"
            "  fp text,"
            "  engine text,"
            "  built_at timestamp default current_timestamp"
            ")"
        )
        executor.con.execute(sql)
        return

    if _is_postgres(executor):
        qual = _pg_qual_meta(executor)
        ddl = (
            f"create table if not exists {qual} ("
            "  node_name text primary key,"
            "  relation text,"
            "  fp text,"
            "  engine text,"
            "  built_at timestamptz default now()"
            ")"
        )
        with executor.engine.begin() as conn:
            conn.execute(text(ddl))
        return

    if _is_bigquery(executor):
        # BigQuery supports IF NOT EXISTS in standard SQL DDL
        qual = _bq_qual_meta(executor)
        ddl = (
            f"create table if not exists {qual} ("
            "  node_name string,"
            "  relation string,"
            "  fp string,"
            "  engine string,"
            "  built_at timestamp"
            ")"
        )
        executor.client.query(ddl).result()
        return

upsert_meta

upsert_meta(executor, node_name, relation, fp, engine)

Insert or update _ff_meta for a given node.

Source code in src/fastflowtransform/meta.py

def upsert_meta(executor: Any, node_name: str, relation: str, fp: str, engine: str) -> None:
    """
    Insert or update `_ff_meta` for a given node.
    """
    ensure_meta_table(executor)

    if _is_duckdb(executor):
        # DuckDB: emulate upsert via delete + insert inside the same connection.
        executor.con.execute('delete from "_ff_meta" where node_name = ?', [node_name])
        executor.con.execute(
            'insert into "_ff_meta"(node_name, relation, fp, engine, built_at) '
            "values (?, ?, ?, ?, current_timestamp)",
            [node_name, relation, fp, engine],
        )
        return

    if _is_postgres(executor):
        qual = _pg_qual_meta(executor)
        sql = (
            f"insert into {qual}(node_name, relation, fp, engine, built_at) "
            "values (:n, :r, :f, :e, now()) "
            "on conflict (node_name) do update set "
            "  relation = excluded.relation, "
            "  fp = excluded.fp, "
            "  engine = excluded.engine, "
            "  built_at = now()"
        )
        with executor.engine.begin() as conn:
            conn.execute(text(sql), {"n": node_name, "r": relation, "f": fp, "e": engine})
        return

    if _is_bigquery(executor):
        qual = _bq_qual_meta(executor)

        # Use MERGE to emulate upsert
        # Parameterization with BigQuery QueryJobConfig is optional; build a safe literal instead.
        def _q(s: str) -> str:
            return s.replace("\\", "\\\\").replace("`", "\\`").replace("'", "\\'")

        sql = f"""merge {qual} T
        using (
          select '{_q(node_name)}' as node_name,
                 '{_q(relation)}'  as relation,
                 '{_q(fp)}'        as fp,
                 '{_q(engine)}'    as engine
        ) S
        on T.node_name = S.node_name
        when matched then update set
          relation = S.relation,
          fp       = S.fp,
          engine   = S.engine,
          built_at = current_timestamp()
        when not matched then insert (node_name, relation, fp, engine, built_at)
          values (S.node_name, S.relation, S.fp, S.engine, current_timestamp())
        """
        executor.client.query(sql).result()
        return

get_meta

get_meta(executor, node_name)

Return (fp, relation, built_at, engine) for the node, or None if not found.

Source code in src/fastflowtransform/meta.py

def get_meta(executor: Any, node_name: str) -> tuple[str, str, Any, str] | None:
    """
    Return (fp, relation, built_at, engine) for the node, or None if not found.
    """
    if _is_duckdb(executor):
        row = executor.con.execute(
            'select fp, relation, built_at, engine from "_ff_meta" where node_name = ? limit 1',
            [node_name],
        ).fetchone()
        return (row[0], row[1], row[2], row[3]) if row else None

    if _is_postgres(executor):
        qual = _pg_qual_meta(executor)
        with executor.engine.begin() as conn:
            row = conn.execute(
                text(
                    f"select fp, relation, built_at, engine from {qual} "
                    "where node_name = :n limit 1"
                ),
                {"n": node_name},
            ).fetchone()
        return (row[0], row[1], row[2], row[3]) if row else None

    if _is_bigquery(executor):
        qual = _bq_qual_meta(executor)
        # Parameterized query would need google.cloud.bigquery; keep it dependency-light.
        node = node_name.replace("\\", "\\\\").replace("`", "\\`").replace("'", "\\'")
        sql = (
            f"select fp, relation, built_at, engine from {qual} where node_name = '{node}' limit 1"
        )
        rows = list(executor.client.query(sql).result())
        if not rows:
            return None
        r = rows[0]
        # Access by field name if available, else positional
        try:
            return (r["fp"], r["relation"], r["built_at"], r["engine"])
        except Exception:
            return (r[0], r[1], r[2], r[3])

    return None

relation_exists

relation_exists(executor, relation)

Check whether a materialized relation exists on the active engine.

Source code in src/fastflowtransform/meta.py

def relation_exists(executor: Any, relation: str) -> bool:
    """
    Check whether a materialized relation exists on the active engine.
    """
    if _is_duckdb(executor):
        try:
            rows = executor.con.execute(
                "select 1 from information_schema.tables "
                + "where table_schema in ('main','temp') and table_name = ?",
                [relation],
            ).fetchall()
            return bool(rows)
        except Exception:
            return True  # be permissive on unexpected errors

    if _is_postgres(executor):
        try:
            with executor.engine.begin() as conn:
                rows = conn.execute(
                    text(
                        "select 1 from information_schema.tables "
                        + "where table_schema = current_schema() and table_name = :t"
                    ),
                    {"t": relation},
                ).fetchall()
            return bool(rows)
        except Exception:
            return True

    if _is_bigquery(executor):
        try:
            dataset = getattr(executor, "dataset", None)
            project = getattr(executor, "project", None)
            if not dataset:
                return True
            qual = f"`{project}.{dataset}`" if project else f"`{dataset}`"
            rel = relation.replace("`", "\\`").replace("'", "\\'")
            sql = (
                f"select 1 from {qual}.INFORMATION_SCHEMA.TABLES where table_name = '{rel}' limit 1"
            )
            rows = list(executor.client.query(sql).result())
            return bool(rows)
        except Exception:
            return True

    return True

delete_meta_for_node

delete_meta_for_node(executor, node_name)

Remove meta row(s) for a given logical node. Best-effort, silent on absence.

Source code in src/fastflowtransform/meta.py

def delete_meta_for_node(executor: Any, node_name: str) -> None:
    """Remove meta row(s) for a given logical node. Best-effort, silent on absence."""
    try:
        ensure_meta_table(executor)
    except Exception:
        return

    # DuckDB
    if hasattr(executor, "con"):
        with suppress(Exception):
            executor.con.execute("delete from _ff_meta where node_name = ?", [node_name])
        return

    # Postgres
    if hasattr(executor, "engine"):
        schema = getattr(executor, "schema", None)
        tbl = f'"{schema}"._ff_meta' if schema else "_ff_meta"
        try:
            with executor.engine.begin() as conn:
                conn.execute(
                    text(f"delete from {tbl} where node_name = :node"), {"node": node_name}
                )
        except Exception:
            pass
        return

    # BigQuery (simple fallback; works for Fake in tests)
    if hasattr(executor, "client") and hasattr(executor, "dataset"):
        dataset = executor.dataset
        with suppress(Exception):
            # Best-effort string literal delete (tests use simple node names)
            executor.client.query(
                f'DELETE FROM `{dataset}._ff_meta` WHERE node_name = "{node_name}"'
            )
        return