Partitioning and Clustering in BigQuery

BigQuery charges and performs by how much data a query has to read. A table with three years of events answers a question about yesterday quickly only if it can avoid reading the other 1,094 days. Partitioning and clustering are the two mechanisms that make that possible. They are often confused because they serve the same goal, but they work at different levels, and the best tables use both.

Two mechanisms, two levels

Partitioning divides a table into segments by the value of one column, usually a date. When a query filters on that column, BigQuery skips the segments that cannot match. Google calls this pruning (Google Cloud: partitioned tables).

Clustering sorts the table's storage blocks by the values of up to four columns. Each block records the range of values it holds, so a query filtering on a clustered column reads only the blocks whose range could contain a match. This is block pruning (Google Cloud: clustered tables).

The third row is the one people miss. A dry run on a partitioned table reports the bytes the query will read. On a clustered table the number of blocks to be scanned is not known in advance, so the estimate is an upper bound and the true cost appears only after execution (Google Cloud: estimate and control costs).

The effect, measured on real data

The script below models both mechanisms on the New York yellow taxi trips for January 2024 (NYC TLC: trip record data). It is a local model, not BigQuery. A block here is a fixed run of 5,000 rows, whereas BigQuery sizes its blocks adaptively. The rule it applies is the documented one: skip a block when its minimum and maximum cannot include the value requested.

The query is simple: trips picked up in zone 132, which is JFK Airport, on 15 January.

# A local model of partition pruning and block pruning on real trip data, using DuckDB and NumPy.
# It is not BigQuery: "blocks" here are fixed runs of 5,000 rows, whereas BigQuery sizes its storage
# blocks adaptively. The model shows the mechanism: a block is skipped when its min-max range of the
# clustering column cannot contain the value asked for.
import duckdb
import numpy as np

SOURCE = "../../dedata/yellow_tripdata_2024-01.parquet"       # NYC yellow taxi trips, January 2024
BLOCK, ZONE, DAY = 5_000, 132, "2024-01-15"                   # zone 132 is JFK Airport

con = duckdb.connect()
total = con.execute(f"SELECT COUNT(*) FROM read_parquet('{SOURCE}')").fetchone()[0]
day = con.execute(f"""
    SELECT PULocationID AS zone, payment_type AS pay
    FROM read_parquet('{SOURCE}')
    WHERE CAST(tpep_pickup_datetime AS DATE) = DATE '{DAY}'
    ORDER BY tpep_pickup_datetime""").fetchnumpy()
zone, pay = np.asarray(day["zone"]), np.asarray(day["pay"])
answer = int((zone == ZONE).sum())


def scanned(order):
    """Rows in the blocks that survive pruning, and how many blocks there are, for one sort order."""
    z = zone[order]
    blocks = [z[i:i + BLOCK] for i in range(0, len(z), BLOCK)]
    kept = [b for b in blocks if b.min() <= ZONE <= b.max()]
    return sum(len(b) for b in kept), len(kept), len(blocks)


layouts = [
    ("partitioned by day only",               np.arange(len(zone))),     # arrival order
    ("by day, clustered by zone",             np.argsort(zone, kind="stable")),
    ("by day, clustered by payment type, zone", np.lexsort((zone, pay))),
]

print(f"query: trips from zone {ZONE} on {DAY}; matching rows: {answer:,}\n")
print(f"{'table layout':40} {'rows scanned':>12} {'of table':>9}   blocks read")
print(f"{'no partitioning':40} {total:12,} {1:9.2%}   all")
for label, order in layouts:
    rows, kept, blocks = scanned(order)
    print(f"{label:40} {rows:12,} {rows / total:9.2%}   {kept} of {blocks}")

query: trips from zone 132 on 2024-01-15; matching rows: 5,651

table layout                             rows scanned  of table   blocks read
no partitioning                             2,964,624   100.00%   all
partitioned by day only                        77,033     2.60%   16 of 16
by day, clustered by zone                      10,000     0.34%   2 of 16
by day, clustered by payment type, zone        27,033     0.91%   6 of 16

Four lines, four lessons:

• No partitioning: every one of 2,964,624 rows is read to find 5,651.
• Partitioned by day: only the 77,033 rows of 15 January are read, 2.6% of the table. All sixteen blocks of that day are still scanned, because the zones are scattered through them in arrival order.
• Clustered by zone within the day: two blocks of sixteen are read. The scan falls to 10,000 rows, a third of one percent of the table.
• Clustered by payment type, then zone: six blocks are read, not two. The table is sorted by zone only within each payment type, so zone 132 appears once per payment type.

The last line is the general rule about column order. The first clustering column gets the full benefit. A later column helps only queries that also filter on the columns before it.

Creating the table

Both properties are declared when the table is created (Google Cloud: create partitioned tables; Google Cloud: create clustered tables):

CREATE TABLE analytics.trips (
  pickup_at     TIMESTAMP,
  pickup_zone   INT64,
  payment_type  INT64,
  fare_amount   NUMERIC
)
PARTITION BY TIMESTAMP_TRUNC(pickup_at, DAY)
CLUSTER BY pickup_zone, payment_type
OPTIONS (
  require_partition_filter = TRUE,
  partition_expiration_days = 730
);

require_partition_filter rejects any query that does not say which partitions it wants. It turns the costliest mistake, an accidental full scan, into an error message. partition_expiration_days deletes old partitions automatically.

Choosing the partition column

Three kinds of partitioning exist:

• Time-unit column: a DATE, TIMESTAMP or DATETIME column in the data. This is the usual choice.
• Ingestion time: BigQuery assigns each row to a partition by when it arrived, exposed as the pseudocolumn _PARTITIONTIME.
• Integer range: ranges of an integer column, such as a customer number.

Pick the column that most queries filter on. For event data that is nearly always the event date. Then choose the granularity, using Google's guidance:

• Daily is the default and right for data spread over a wide range of dates.
• Hourly suits a high volume spanning a short range, typically under six months.
• Monthly or yearly suits a small amount of data per day over a long period.

Two limits bound the choice. A table can hold 10,000 partitions, and one job can modify 4,000 (Google Cloud: quotas and limits). Daily partitions therefore allow about 27 years. Hourly partitions allow about 14 months, which is why they need an expiration.

Partitions can also be too small. Google suggests clustering instead when a partition would hold less than about 10 GB, because many small partitions add metadata without adding benefit.

Choosing clustering columns

• Cluster on what queries filter or aggregate by, beyond the partition column: a customer, a region, a status.
• High cardinality is welcome. Columns with many distinct values gain the most, the opposite of partitioning.
• Order by how often each column is filtered. As the model showed, the first column is served best.
• Do not expect much on small tables. Google notes that tables or partitions under 64 MB gain little.

Clustering is maintained for you. As rows are added, BigQuery reclusters in the background, with no effect on your query capacity. One catch: adding clustering to an existing table affects new data only.

Filters that prune, and filters that do not

A partitioned table saves nothing if the query cannot be pruned. The rules are specific (Google Cloud: query partitioned tables):

-- Prunes: the partitioning column compared with constants
SELECT COUNT(*) FROM analytics.trips
WHERE pickup_at >= '2024-01-15' AND pickup_at < '2024-01-16';

-- Prunes: a supported function on the column, with a constant
SELECT COUNT(*) FROM analytics.trips
WHERE DATE(pickup_at) = '2024-01-15';

-- Does not prune: the value comes from another query
SELECT COUNT(*) FROM analytics.trips
WHERE pickup_at >= (SELECT MAX(loaded_at) FROM analytics.load_log);

The principle is that BigQuery must know which partitions to read before it reads any data. So:

• Use constants. A value taken from a subquery or another column is unknown until execution.
• Isolate the column. Put the partitioning column alone on one side of the comparison, or wrap it only in a supported function such as DATE or TIMESTAMP_TRUNC.
• Check with a dry run. If the estimate equals the whole table, the filter is not pruning.

When the boundary really does come from another table, fetch it first, then run the main query with that value as a literal or a parameter.

A decision in four questions

1. Do most queries filter on a date? Partition by it, daily unless the volume argues otherwise.
2. Would partitions be smaller than about 10 GB, or number more than 10,000? Use a coarser granularity, or cluster on the date instead.
3. Do queries filter on other columns? Cluster on up to four of them, most used first.
4. Must costs be known before a query runs? Rely on partitioning for that, and treat clustering as a further saving.

For the pricing behind these savings, see BigQuery cost optimization. For loading one partition at a time, see Cloud Storage to BigQuery, and for the architecture underneath, BigQuery explained.

Table design for warehouses is covered in the Data Engineering course.

Sources

• Introduction to partitioned tables (Google Cloud Documentation) — Google Cloud
• Introduction to clustered tables (Google Cloud Documentation) — Google Cloud
• Query partitioned tables (Google Cloud Documentation) — Google Cloud
• Create partitioned tables (Google Cloud Documentation) — Google Cloud
• Create clustered tables (Google Cloud Documentation) — Google Cloud
• Quotas and limits (Google Cloud Documentation) — Google Cloud
• Estimate and control costs (Google Cloud Documentation) — Google Cloud
• TLC Trip Record Data (NYC Taxi and Limousine Commission) — NYC Taxi and Limousine Commission