diff --git a/go.mod b/go.mod
index 8e6266d..4729553 100644
--- a/go.mod
+++ b/go.mod
@@ -3,6 +3,8 @@
 go 1.18
 
 require (
+	cloud.google.com/go v0.105.0
+	cloud.google.com/go/bigquery v1.43.0
 	cloud.google.com/go/errorreporting v0.1.0
 	cloud.google.com/go/logging v1.6.1
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go v1.0.0
@@ -15,18 +17,20 @@
 	go.opencensus.io v0.24.0
 	go.opentelemetry.io/otel v1.4.0
 	go.opentelemetry.io/otel/sdk v1.4.0
+	golang.org/x/exp v0.0.0-20230131160201-f062dba9d201
 	golang.org/x/exp/event v0.0.0-20220218215828-6cf2b201936e
 	golang.org/x/mod v0.7.0
 	golang.org/x/net v0.5.0
 	golang.org/x/tools v0.5.1-0.20230117180257-8aba49bb5ea2
+	google.golang.org/api v0.103.0
 	honnef.co/go/tools v0.2.2
 	mvdan.cc/unparam v0.0.0-20220926085101-66de63301820
 )
 
 require (
-	cloud.google.com/go v0.105.0 // indirect
 	cloud.google.com/go/compute v1.12.1 // indirect
 	cloud.google.com/go/compute/metadata v0.2.1 // indirect
+	cloud.google.com/go/iam v0.7.0 // indirect
 	cloud.google.com/go/longrunning v0.3.0 // indirect
 	cloud.google.com/go/monitoring v1.8.0 // indirect
 	cloud.google.com/go/trace v1.4.0 // indirect
@@ -41,6 +45,7 @@
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/google/uuid v1.3.0 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.2.0 // indirect
 	github.com/googleapis/gax-go/v2 v2.7.0 // indirect
 	github.com/imdario/mergo v0.3.13 // indirect
@@ -60,7 +65,7 @@
 	golang.org/x/sync v0.1.0 // indirect
 	golang.org/x/sys v0.4.0 // indirect
 	golang.org/x/text v0.6.0 // indirect
-	google.golang.org/api v0.103.0 // indirect
+	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/genproto v0.0.0-20221201164419-0e50fba7f41c // indirect
 	google.golang.org/grpc v1.50.1 // indirect
diff --git a/go.sum b/go.sum
index 200485b..4f2b161 100644
--- a/go.sum
+++ b/go.sum
@@ -37,16 +37,21 @@
 cloud.google.com/go/bigquery v1.5.0/go.mod h1:snEHRnqQbz117VIFhE8bmtwIDY80NLUZUMb4Nv6dBIg=
 cloud.google.com/go/bigquery v1.7.0/go.mod h1://okPTzCYNXSlb24MZs83e2Do+h+VXtc4gLoIoXIAPc=
 cloud.google.com/go/bigquery v1.8.0/go.mod h1:J5hqkt3O0uAFnINi6JXValWIb1v0goeZM77hZzJN/fQ=
+cloud.google.com/go/bigquery v1.43.0 h1:u0fvz5ysJBe1jwUPI4LuPwAX+o+6fCUwf3ECeg6eDUQ=
+cloud.google.com/go/bigquery v1.43.0/go.mod h1:ZMQcXHsl+xmU1z36G2jNGZmKp9zNY5BUua5wDgmNCfw=
 cloud.google.com/go/compute v0.1.0/go.mod h1:GAesmwr110a34z04OlxYkATPBEfVhkymfTBXtfbBFow=
 cloud.google.com/go/compute v1.2.0/go.mod h1:xlogom/6gr8RJGBe7nT2eGsQYAFUbbv8dbC29qE3Xmw=
 cloud.google.com/go/compute v1.12.1 h1:gKVJMEyqV5c/UnpzjjQbo3Rjvvqpr9B1DFSbJC4OXr0=
 cloud.google.com/go/compute v1.12.1/go.mod h1:e8yNOBcBONZU1vJKCvCoDw/4JQsA0dpM4x/6PIIOocU=
 cloud.google.com/go/compute/metadata v0.2.1 h1:efOwf5ymceDhK6PKMnnrTHP4pppY5L22mle96M1yP48=
 cloud.google.com/go/compute/metadata v0.2.1/go.mod h1:jgHgmJd2RKBGzXqF5LR2EZMGxBkeanZ9wwa75XHJgOM=
+cloud.google.com/go/datacatalog v1.8.0 h1:6kZ4RIOW/uT7QWC5SfPfq/G8sYzr/v+UOmOAxy4Z1TE=
 cloud.google.com/go/datastore v1.0.0/go.mod h1:LXYbyblFSglQ5pkeyhO+Qmw7ukd3C+pD7TKLgZqpHYE=
 cloud.google.com/go/datastore v1.1.0/go.mod h1:umbIZjpQpHh4hmRpGhH4tLFup+FVzqBi1b3c64qFpCk=
 cloud.google.com/go/errorreporting v0.1.0 h1:z40EhrjRspplwbpO+9DSnC4kgDokBi94T/gYwtdKL5Q=
 cloud.google.com/go/errorreporting v0.1.0/go.mod h1:cZSiBMvrnl0X13pD9DwKf9sQ8Eqy3EzHqkyKBZxiIrM=
+cloud.google.com/go/iam v0.7.0 h1:k4MuwOsS7zGJJ+QfZ5vBK8SgHBAvYN/23BWsiihJ1vs=
+cloud.google.com/go/iam v0.7.0/go.mod h1:H5Br8wRaDGNc8XP3keLc4unfUUZeyH3Sfl9XpQEYOeg=
 cloud.google.com/go/logging v1.6.1 h1:ZBsZK+JG+oCDT+vaxwqF2egKNRjz8soXiS6Xv79benI=
 cloud.google.com/go/logging v1.6.1/go.mod h1:5ZO0mHHbvm8gEmeEUHrmDlTDSu5imF6MUP9OfilNXBw=
 cloud.google.com/go/longrunning v0.3.0 h1:NjljC+FYPV3uh5/OwWT6pVU+doBqMg2x/rZlE+CamDs=
@@ -63,6 +68,7 @@
 cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohlUTyfDhBk=
 cloud.google.com/go/storage v1.8.0/go.mod h1:Wv1Oy7z6Yz3DshWRJFhqM/UCfaWIRTdp0RXyy7KQOVs=
 cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9ullr3+Kg0=
+cloud.google.com/go/storage v1.27.0 h1:YOO045NZI9RKfCj1c5A/ZtuuENUc8OAW+gHdGnDgyMQ=
 cloud.google.com/go/trace v1.4.0 h1:qO9eLn2esajC9sxpqp1YKX37nXC3L4BfGnPS0Cx9dYo=
 cloud.google.com/go/trace v1.4.0/go.mod h1:UG0v8UBqzusp+z63o7FK74SdFE+AXpCLdFb1rshXG+Y=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
@@ -198,9 +204,11 @@
 github.com/google/go-cmp v0.5.7/go.mod h1:n+brtR0CgQNWTVd5ZUFpTBC8YFBDLK/h/bpaJ8/DtOE=
 github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
 github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/martian v2.1.0+incompatible h1:/CP5g8u/VJHijgedC/Legn3BAbAaWPgecwXBIDzw5no=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
 github.com/google/martian/v3 v3.0.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0=
 github.com/google/martian/v3 v3.1.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0=
+github.com/google/martian/v3 v3.2.1 h1:d8MncMlErDFTwQGBK1xhv026j9kqhvw1Qv9IbWT1VLQ=
 github.com/google/martian/v3 v3.2.1/go.mod h1:oBOf6HBosgwRXnUGWUB05QECsc6uvmMiJ3+6W4l/CUk=
 github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
 github.com/google/pprof v0.0.0-20190515194954-54271f7e092f/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
@@ -219,6 +227,8 @@
 github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
+github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/enterprise-certificate-proxy v0.2.0 h1:y8Yozv7SZtlU//QXbezB6QkpuE6jMD2/gfzk4AftXjs=
 github.com/googleapis/enterprise-certificate-proxy v0.2.0/go.mod h1:8C0jb7/mgJe/9KK8Lm7X9ctZC2t60YyIpYEI16jx0Qg=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
@@ -352,6 +362,8 @@
 golang.org/x/exp v0.0.0-20200119233911-0405dc783f0a/go.mod h1:2RIsYlXP63K8oxa1u096TMicItID8zy7Y6sNkU49FU4=
 golang.org/x/exp v0.0.0-20200207192155-f17229e696bd/go.mod h1:J/WKrq2StrnmMY6+EHIKF9dgMWnmCNThgcyBT1FY9mM=
 golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6/go.mod h1:3jZMyOhIsHpP37uCMkUooju7aAi5cS1Q23tOzKc+0MU=
+golang.org/x/exp v0.0.0-20230131160201-f062dba9d201 h1:BEABXpNXLEz0WxtA+6CQIz2xkg80e+1zrhWyMcq8VzE=
+golang.org/x/exp v0.0.0-20230131160201-f062dba9d201/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc=
 golang.org/x/exp/event v0.0.0-20220218215828-6cf2b201936e h1:K2AuHMC+jaRTzAcivRwKOzjTZ1925Yx4xHMg07YoBQc=
 golang.org/x/exp/event v0.0.0-20220218215828-6cf2b201936e/go.mod h1:AVlZHjhWbW/3yOcmKMtJiObwBPJajBlUpQXRijFNrNc=
 golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js=
@@ -602,6 +614,8 @@
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 h1:H2TDz8ibqkAF6YGhCdN3jS9O0/s90v0rJh3X/OLHEUk=
+golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2/go.mod h1:K8+ghG5WaK9qNqU5K3HdILfMLy1f3aNYFI/wnl100a8=
 google.golang.org/api v0.4.0/go.mod h1:8k5glujaEP+g9n7WNsDg8QP6cUVNI86fCNMcbazEtwE=
 google.golang.org/api v0.7.0/go.mod h1:WtwebWUNSVBH/HAw79HIFXZNqEvBhG+Ra+ax0hx3E3M=
 google.golang.org/api v0.8.0/go.mod h1:o4eAsZoiT+ibD93RtjEohWalFOjRDx6CVaqeizhEnKg=
diff --git a/internal/bigquery/bigquery.go b/internal/bigquery/bigquery.go
new file mode 100644
index 0000000..b87fdea
--- /dev/null
+++ b/internal/bigquery/bigquery.go
@@ -0,0 +1,336 @@
+// Copyright 2022 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// Package bigquery provides a client for reading and writing to BigQuery.
+package bigquery
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"net/http"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	bq "cloud.google.com/go/bigquery"
+	"cloud.google.com/go/civil"
+	"golang.org/x/exp/maps"
+	"golang.org/x/pkgsite-metrics/internal/derrors"
+	"google.golang.org/api/googleapi"
+	"google.golang.org/api/iterator"
+)
+
+// Client is a client for connecting to BigQuery.
+type Client struct {
+	client  *bq.Client
+	dataset *bq.Dataset
+}
+
+// NewClient creates a new client for connecting to BigQuery, referring to a single dataset.
+// The dataset must already exist.
+func NewClient(ctx context.Context, projectID, datasetID string) (_ *Client, err error) {
+	defer derrors.Wrap(&err, "New(ctx, %q, %q)", projectID, datasetID)
+	client, err := bq.NewClient(ctx, projectID)
+	if err != nil {
+		return nil, err
+	}
+	dataset := client.DatasetInProject(projectID, datasetID)
+	if _, err := dataset.Metadata(ctx); err != nil {
+		return nil, err
+	}
+	return &Client{
+		client:  client,
+		dataset: dataset,
+	}, nil
+}
+
+// NewClientCreate is like NewClient, but it creates the dataset if it doesn't exist.
+func NewClientCreate(ctx context.Context, projectID, datasetID string) (_ *Client, err error) {
+	if err := CreateDataset(ctx, projectID, datasetID); err != nil {
+		return nil, err
+	}
+	return NewClient(ctx, projectID, datasetID)
+}
+
+// CreateDataset creates a BigQuery dataset if it does not already exist.
+func CreateDataset(ctx context.Context, projectID, datasetID string) (err error) {
+	defer derrors.Wrap(&err, "CreateDataset(%q, %q)", projectID, datasetID)
+	client, err := bq.NewClient(ctx, projectID)
+	if err != nil {
+		return err
+	}
+	dataset := client.DatasetInProject(projectID, datasetID)
+	err = dataset.Create(ctx, &bq.DatasetMetadata{Name: datasetID})
+	if err != nil && !isAlreadyExistsError(err) {
+		return err
+	}
+	return nil
+}
+
+// isNotFoundError reports whether the error, which should come from a cloud.google.com/go/bigquery
+// client call, is a NotFound error.
+func isNotFoundError(err error) bool {
+	return hasCode(err, http.StatusNotFound)
+}
+
+func isAlreadyExistsError(err error) bool {
+	// The BigQuery API uses 409 for something that exists.
+	return hasCode(err, http.StatusConflict)
+}
+
+func hasCode(err error, code int) bool {
+	var gerr *googleapi.Error
+	if !errors.As(err, &gerr) {
+		return false
+	}
+	return gerr.Code == code
+}
+
+// Table returns a handle for the given tableID in the client's dataset.
+func (c *Client) Table(tableID string) *bq.Table {
+	return c.dataset.Table(tableID)
+}
+
+// FullTableName returns the fully-qualified name of the table, suitable for
+// use in queries.
+func (c *Client) FullTableName(tableID string) string {
+	// From https://github.com/googleapis/google-cloud-go/blob/bigquery/v1.43.0/bigquery/table.go#L544.
+	return fmt.Sprintf("%s.%s.%s", c.dataset.ProjectID, c.dataset.DatasetID, tableID)
+}
+
+// CreateTable creates a table with the given name if it doesn't exist.
+func (c *Client) CreateTable(ctx context.Context, tableID string) (err error) {
+	defer derrors.Wrap(&err, "CreateTable(%q)", tableID)
+	schema := tableSchema(tableID)
+	if schema == nil {
+		return fmt.Errorf("no schema registered for table %q", tableID)
+	}
+	err = c.Table(tableID).Create(ctx, &bq.TableMetadata{Schema: schema})
+	if err != nil && !isAlreadyExistsError(err) {
+		return err
+	}
+	return nil
+}
+
+// CreateOrUpdateTable creates a table if it does not exist, or updates it if it does.
+// It returns true if it created the table.
+func (c *Client) CreateOrUpdateTable(ctx context.Context, tableID string) (created bool, err error) {
+	defer derrors.Wrap(&err, "CreateOrUpdateTable(%q)", tableID)
+	meta, err := c.Table(tableID).Metadata(ctx) // check if the table already exists
+	if err != nil {
+		if !isNotFoundError(err) {
+			return false, err
+		}
+		return true, c.CreateTable(ctx, tableID)
+	}
+	schema := tableSchema(tableID)
+	if schema == nil {
+		return false, fmt.Errorf("no schema registered for table %q", tableID)
+	}
+	_, err = c.Table(tableID).Update(ctx, bq.TableMetadataToUpdate{Schema: schema}, meta.ETag)
+	return false, err
+}
+
+// Upload inserts a row into the table.
+func (c *Client) Upload(ctx context.Context, tableID string, row any) (err error) {
+	defer derrors.Wrap(&err, "Upload(ctx, %q)", tableID)
+	u := c.Table(tableID).Inserter()
+	if s, ok := row.(interface{ SetUploadTime(time.Time) }); ok {
+		s.SetUploadTime(time.Now())
+	}
+	return u.Put(ctx, row)
+}
+
+// UploadMany inserts multiple rows into the table.
+// Each row should be a struct pointer.
+// The chunkSize parameter limits the number of rows sent in a single request; this may
+// be necessary to avoid reaching the maximum size of a request.
+// If chunkSize is <= 0, all rows will be sent in one request.
+func UploadMany[T any](ctx context.Context, client *Client, tableID string, rows []T, chunkSize int) (err error) {
+	defer derrors.Wrap(&err, "UploadMany(%q), %d rows, chunkSize=%d", tableID, len(rows), chunkSize)
+
+	now := time.Now()
+	// Set upload time.
+	for _, r := range rows {
+		if s, ok := any(r).(interface{ SetUploadTime(time.Time) }); ok {
+			s.SetUploadTime(now)
+		}
+	}
+
+	ins := client.Table(tableID).Inserter()
+	if chunkSize <= 0 {
+		return ins.Put(ctx, rows)
+	}
+	start := 0
+	for start < len(rows) {
+		end := start + chunkSize
+		if end > len(rows) {
+			end = len(rows)
+		}
+		for {
+			if err := ins.Put(ctx, rows[start:end]); err == nil {
+				break
+			} else if hasCode(err, http.StatusRequestEntityTooLarge) && end-start > 1 {
+				// Request too large; reduce this chunk size by half.
+				end = start + (end-start)/2
+				continue
+			} else {
+				return err
+			}
+		}
+		start = end
+	}
+	return nil
+}
+
+// ForEachRow calls f for each row in the given iterator.
+// It returns as soon as f returns false.
+func ForEachRow[T any](iter *bq.RowIterator, f func(*T) bool) error {
+	for {
+		var row T
+		err := iter.Next(&row)
+		if err == iterator.Done {
+			break
+		}
+		if err != nil {
+			return err
+		}
+		if !f(&row) {
+			break
+		}
+	}
+	return nil
+}
+
+// All returns all rows returned by iter.
+func All[T any](iter *bq.RowIterator) ([]*T, error) {
+	var ts []*T
+	err := ForEachRow(iter, func(t *T) bool {
+		ts = append(ts, t)
+		return true
+	})
+	if err != nil {
+		return nil, err
+	}
+	return ts, nil
+}
+
+func (c *Client) Query(ctx context.Context, q string) (*bq.RowIterator, error) {
+	return c.client.Query(q).Read(ctx)
+}
+
+// NullString constructs a bq.NullString.
+func NullString(s string) bq.NullString {
+	return bq.NullString{StringVal: s, Valid: true}
+}
+
+// NullInt constructs a bq.NullInt.
+func NullInt(i int) bq.NullInt64 {
+	return bq.NullInt64{Int64: int64(i), Valid: true}
+}
+
+// NullTime constructs a bq.NullTime.
+func NullTime(t time.Time) bq.NullTime {
+	return bq.NullTime{Time: civil.TimeOf(t), Valid: true}
+}
+
+// schemaVersion computes a relatively short string from a schema, such that
+// different schemas result in different strings with high probability.
+func schemaVersion(schema bq.Schema) string {
+	hash := sha256.Sum256([]byte(schemaString(schema)))
+	return hex.EncodeToString(hash[:])
+}
+
+// schemaString returns a long, human-readable string summarizing schema.
+func schemaString(schema bq.Schema) string {
+	var b strings.Builder
+	for i, field := range schema {
+		if i > 0 {
+			b.WriteRune(';')
+		}
+		b.WriteString(field.Name)
+		if field.Repeated {
+			b.WriteString(",rep")
+		}
+		if field.Required {
+			b.WriteString(",req")
+		}
+		b.WriteByte(':')
+		if field.Type == bq.RecordFieldType {
+			fmt.Fprintf(&b, "(%s)", schemaString(field.Schema))
+		} else {
+			b.WriteString(string(field.Type))
+		}
+	}
+	return b.String()
+}
+
+var (
+	tableMu sync.Mutex
+	tables  = map[string]bq.Schema{}
+)
+
+func addTable(tableID string, s bq.Schema) {
+	tableMu.Lock()
+	defer tableMu.Unlock()
+	tables[tableID] = s
+}
+
+// tableSchema returns the schema associated with the given table,
+// or nil if there is none.
+func tableSchema(tableID string) bq.Schema {
+	tableMu.Lock()
+	defer tableMu.Unlock()
+	return tables[tableID]
+}
+
+// Tables returns all the tables used by the worker.
+func Tables() []string {
+	tableMu.Lock()
+	defer tableMu.Unlock()
+	tableIDs := maps.Keys(tables)
+	sort.Strings(tableIDs)
+	return tableIDs
+}
+
+// partitionQuery returns a query that returns one row for each distinct value
+// of partitionColumn in tableName.
+// The selected row will be the first one according to the orderings, which
+// should be comma-separated ORDER BY clauses.
+//
+// For example, say the students table holds student names and classes.
+// Then
+//
+//	partitionQuery("students", "class", "name ASC")
+//
+// will construct a query returning the student in each class whose name is
+// alphabetically first.
+//
+// (BigQuery SQL has no DISTINCT ON feature and doesn't allow columns of type RECORD
+// in queries with DISTINCT, so we have to take this approach.)
+func partitionQuery(tableName, partitionColumn, orderings string) string {
+	// This query first organizes the table rows into windows that have the same partitionColumn.
+	// The rows in each window are sorted by the given orderings.
+	// They are then assigned numbers, where 1 is the first row in the window.
+	// Finally, only the first row in each window is chosen.
+	// (ROW_NUMBER guarantees that each row has a distinct number; RANK assigns the
+	// same number to identical rows, which means that the result may still contain
+	// duplicates.)
+	const qf = `
+		SELECT * EXCEPT (rownum)
+		FROM (
+			SELECT *, ROW_NUMBER() OVER (
+				PARTITION BY %s
+				ORDER BY %s
+			) AS rownum
+			FROM %s
+		) WHERE rownum = 1
+	`
+
+	return fmt.Sprintf(qf, partitionColumn, orderings, "`"+tableName+"`")
+}