BigQuery Federated Queries Explained

Some of the data an analysis needs is not in the warehouse. The current list of customers, today's open orders or a product catalogue may live in an operational database and change by the minute. The usual answer is a pipeline that copies it. A federated query is the other answer: BigQuery sends a statement to the live database, takes the rows that come back, and joins them to its own tables. Nothing is copied or scheduled. It is the right tool for a specific set of cases and a poor one outside them.

How a federated query works

You call EXTERNAL_QUERY with a connection and a statement. BigQuery runs the statement in the external database, converts the result to GoogleSQL types and treats it as a temporary table (Google Cloud: federated queries):

SELECT c.customer_id, c.name, o.first_order_date
FROM warehouse.customers AS c
LEFT JOIN EXTERNAL_QUERY(
  'us.orders-db',
  '''SELECT customer_id, MIN(order_date) AS first_order_date
     FROM orders
     GROUP BY customer_id''') AS o
  ON o.customer_id = c.customer_id;

Three details to notice (Google Cloud: federated query functions):

• The inner statement is in the source's dialect, MySQL or PostgreSQL here, not GoogleSQL.
• It is read-only. DML and DDL are not allowed in the external query.
• Ordering is not preserved. An ORDER BY inside the external query is not honoured; sort in the outer query.

The supported sources are Cloud SQL, AlloyDB and Spanner.

The connection

EXTERNAL_QUERY never sees a password. An administrator creates a connection that holds the database credentials, encrypted, and grants people permission to use it (Google Cloud: connections).

bq mk --connection --display_name='Orders database' --connection_type='CLOUD_SQL' \
  --properties='{"instanceId":"analytics-prod:us-central1:orders","database":"orders","type":"POSTGRES"}' \
  --connection_credential='{"username":"bq_reader", "password":"PLACEHOLDER"}' \
  --project_id=analytics-prod --location=us orders-db

gcloud projects add-iam-policy-binding analytics-prod \
  --member=serviceAccount:CONNECTION_SERVICE_ACCOUNT \
  --role=roles/cloudsql.client

The commands follow the documented form (Google Cloud: connect to Cloud SQL). Two habits make this safe. Use a database user that can only read, and only the tables that analysts should see. And point the connection at a read replica, which Google recommends, so that an analyst's query cannot slow the application.

People who run federated queries then need the BigQuery Connection User role, not the database password (Google Cloud: Cloud SQL federated queries).

Pushdown: what crosses the wire

The cost and speed of a federated query depend on how many rows travel from the database to BigQuery. BigQuery tries to reduce that through SQL pushdown: it rewrites the external statement so the database returns only the columns and rows the outer query needs.

The rules are narrow. Pushdown applies to external queries of the form SELECT * FROM table. It covers column pruning and filters. It does not cover joins, limits, ordering or aggregation.

The script shows what that means on the Chinook sample database (Chinook on GitHub). SQLite stands in for the operational database and DuckDB for the warehouse. It is a local reproduction, not BigQuery or Cloud SQL. It runs the statements that the documentation says are sent, and counts what comes back.

# A local reproduction of what a federated query moves. SQLite holding the Chinook sample database
# stands in for the operational database, and DuckDB for the warehouse. It is not BigQuery or
# Cloud SQL; it runs the statements the documentation says are sent, and counts what comes back.
import sqlite3
import duckdb

source = sqlite3.connect("../../sqlds/Chinook_Sqlite.sqlite")          # the "operational database"


def external_query(sql):
    """Run a statement in the source database, as EXTERNAL_QUERY would, and report what it returns."""
    cursor = source.execute(sql)
    rows = cursor.fetchall()
    return rows, len(rows), len(rows) * len(cursor.description)


QUESTION = "invoices billed to India"
cases = [
    ("no pushdown: the whole table",        "SELECT * FROM Invoice"),
    ("pushdown: columns and filter",        "SELECT BillingCountry FROM (SELECT * FROM Invoice) t WHERE (BillingCountry = 'India')"),
]
print(f"{QUESTION}\n{'statement run in the source database':38} {'rows':>5} {'values':>7}")
for label, sql in cases:
    rows, n, values = external_query(sql)
    answer = sum(1 for row in rows if "India" in row)
    print(f"{label:38} {n:5} {values:7}   answer {answer}")

QUESTION = "first invoice date of each customer"
cases = [
    ("aggregate in the warehouse",           "SELECT CustomerId, InvoiceDate FROM (SELECT * FROM Invoice) t"),
    ("aggregate inside the external query",  "SELECT CustomerId, MIN(InvoiceDate) FROM Invoice GROUP BY CustomerId"),
]
print(f"\n{QUESTION}\n{'statement run in the source database':38} {'rows':>5} {'values':>7}")
for label, sql in cases:
    rows, n, values = external_query(sql)
    first = {}
    for customer, invoice_date in rows:
        first[customer] = min(first.get(customer, invoice_date), invoice_date)
    print(f"{label:38} {n:5} {values:7}   customers {len(first)}, earliest {min(first.values())[:10]}")

# Joining the small result to a table that lives in the warehouse.
warehouse = duckdb.connect()
customers = source.execute("SELECT CustomerId, FirstName, Country FROM Customer").fetchall()
warehouse.execute("CREATE TABLE customers (customer_id INTEGER, name VARCHAR, country VARCHAR)")
warehouse.executemany("INSERT INTO customers VALUES (?, ?, ?)", customers)
result, _, _ = external_query("SELECT CustomerId, MIN(InvoiceDate) FROM Invoice GROUP BY CustomerId")
warehouse.execute("CREATE TABLE first_invoice (customer_id INTEGER, first_invoice VARCHAR)")
warehouse.executemany("INSERT INTO first_invoice VALUES (?, ?)", result)
print("\nwarehouse table joined to the external result, by country")
for country, n, earliest in warehouse.execute("""
        SELECT c.country, COUNT(*), MIN(f.first_invoice)
        FROM customers AS c LEFT JOIN first_invoice AS f USING (customer_id)
        GROUP BY c.country ORDER BY COUNT(*) DESC, c.country LIMIT 4""").fetchall():
    print(f"  {country:10} {n:3} customers, first invoice {earliest[:10]}")

invoices billed to India
statement run in the source database    rows  values
no pushdown: the whole table             412    3708   answer 13
pushdown: columns and filter              13      13   answer 13

first invoice date of each customer
statement run in the source database    rows  values
aggregate in the warehouse               412     824   customers 59, earliest 2009-01-01
aggregate inside the external query       59     118   customers 59, earliest 2009-01-01

warehouse table joined to the external result, by country
  USA         13 customers, first invoice 2009-01-11
  Canada       8 customers, first invoice 2009-01-06
  Brazil       5 customers, first invoice 2009-04-09
  France       5 customers, first invoice 2009-02-01

A filter is pushed down. To count invoices billed to India, the unrewritten statement returns the whole table: 412 rows and 3,708 values. The rewritten one returns 13 rows of one column. The answer is 13 either way.

An aggregate is not. To find each customer's first invoice with the GROUP BY in the outer query, pushdown can drop the unused columns, and all 412 rows still travel. Writing the aggregate inside the external query returns 59 rows, one per customer. On a table of 400 million rows that difference decides whether the query is practical.

The rule to take away: let pushdown handle simple filters, and do the heavy reduction yourself, inside the external statement, in the source's own SQL. Google's own example does exactly that.

The last block joins the small result to a table that lives in the warehouse, which is the purpose of the whole exercise.

What it costs and where it runs

• Pricing. With on-demand pricing you are charged for the bytes returned by the external query. With an edition, you are charged for slots as usual. The source database, of course, does its part of the work on its own resources.
• No estimate in advance. The bytes cannot be known before the external query runs, so a dry run cannot price it and the maximum bytes billed setting is not supported. This is one more reason to keep external results small.
• Regions. For Cloud SQL and AlloyDB, a dataset in a single region can only query a database in the same region. The US and EU multi-regions can query sources in their geographic area. A cross-region federated query counts against a limit of 1 TB per project per day.
• Connections. One query can use at most 10 connections.

When to use one, and when not

Use a federated query when:

• The data is small and changes often, such as reference or status tables.
• You need the current value, not last night's.
• The need is occasional, and a pipeline would cost more to build than the question is worth.
• You are exploring an operational database before deciding what to bring across.

Do not use one when:

• The table is large and the analysis is heavy. The source database is not built for analytical scans, and Google warns that federated queries are likely to be slower than queries on BigQuery storage.
• Dashboards refresh it constantly. Every refresh is a query on the operational database.
• You need history. The source holds the present; yesterday's value is gone unless you kept it.

In those cases, replicate. Datastream is a change data capture service that keeps a copy of the tables in BigQuery up to date (Google Cloud: Datastream), and a scheduled load is enough when daily freshness will do. See loading data into BigQuery.

A short checklist

1. Read-only database user, limited to the needed tables.
2. Connection to a read replica.
3. Reduction written inside the external statement.
4. Outer query sorts and joins; inner query filters and aggregates.
5. Result sizes checked before the query is put behind a dashboard.

For files in Cloud Storage, the equivalent is BigLake and external tables. To choose the operational database in the first place, see Cloud SQL vs Spanner vs Bigtable.

Working across warehouses and operational databases is covered in the Data Engineering course.

Sources

• Introduction to federated queries (Google Cloud Documentation) — Google Cloud
• Cloud SQL federated queries (Google Cloud Documentation) — Google Cloud
• Federated query functions (Google Cloud Documentation) — Google Cloud
• Connect to Cloud SQL (Google Cloud Documentation) — Google Cloud
• Introduction to connections (Google Cloud Documentation) — Google Cloud
• Datastream overview (Google Cloud Documentation) — Google Cloud
• Chinook sample database (GitHub) — GitHub