bqsink

README

bqsink

Expressive BigQuery ingestion library for Go.

bqsink writes rows to BigQuery and keeps the destination table's schema in sync with a schema declared in Go code.

The declaration is the source of truth, and it belongs to the row type. The schema comes from struct tags, or from the type's own BigQueryTableMetadata where tags cannot say it, and the real table follows. No option describes the table: bqsink never infers a schema from the data being written either, so a field added to a struct becomes a column rather than a silent mismatch at write time.

type AccessLog struct {
	Timestamp time.Time `bqsink:"timestamp,required"`
	UserID    string    `bqsink:"user_id"`
	Path      string    `bqsink:"path"`
}

client, err := bigquery.NewClient(ctx, "my-project")
if err != nil {
	return err
}

w, err := (&bqsink.LoadJobs{}).NewWriter(client, bqsink.Relation{
	DatasetID: "logs",
	TableID:   "access",
})
if err != nil {
	return err
}
defer func() {
	if cerr := w.Close(ctx); cerr != nil && err == nil {
		err = cerr
	}
}()

s, err := bqsink.NewSinker(w, bqsink.DeclarationOf[AccessLog]())
if err != nil {
	return err
}

rows := []AccessLog{
	{Timestamp: time.Now(), UserID: "u1", Path: "/"},
	{Timestamp: time.Now(), UserID: "u2", Path: "/about"},
}
n, err := s.Sink(ctx, rows)
if err != nil {
	// rows[n:] never reached BigQuery. They are still yours to deal with.
	return err
}
return nil

Building takes two steps: a writer for the transport (LoadJobs here; see Transports), and a Sinker for the declaration. Closing belongs to the writer, since that is what holds a connection — a Sinker buffers nothing of its own to flush and so has no Close.

Sink hands the rows it is given to the writer and waits for its WriteResult to settle, so nothing is buffered inside the Sinker itself: the rows it is given are the batch handed to the writer. What the writer then does with that batch is its own business — LoadJobs.FlushRows can hold rows back across calls, for instance (see Flushing rows). Sink returns how many of them that settlement counts as done, and a non-nil error whenever that is fewer than it was given: rows[n:] are exactly the ones that did not make it, and nothing else records them.

What counts as done depends on the writer, not on Sink: delivery to BigQuery for one that promises it, or only acceptance into a buffer of its own for one that promises that instead, such as a LoadJobsWriter with LoadJobs.FlushRows set. A Sink call whose rows only reached that buffer can still return (n, nil) — the submission it goes on to trigger, or fails to, is reported later, by FlushRows or Close, not by this call (see Flushing rows). Give Sink a whole batch rather than one row at a time, especially with LoadJobs, where every call is a load job unless FlushRows says otherwise.

A slice is a batch of its elements and anything else is a single row, so Sink(ctx, rows) and Sink(ctx, row) are both ordinary calls. Every row in a batch has to be of one type, which a []AccessLog gives for free and a []any can break.

The row type is settled by NewSinker, not by the first Sink. It reads the Declaration it is given and keeps it for as long as the Sinker lives: a later Sink handing over another type is an error, and a second type needs a Sinker of its own. Everything the declaration decides — a struct that cannot be mapped to a row, a column missing from a spelled out schema, a FillRow with a value receiver — is therefore reported by NewSinker, which talks to nothing and reads nothing. Nothing contacts BigQuery until the first Sink, which is what reconciles the real table with the declaration and hands the writer the settled schema.

DeclarationOf[T]() is the ordinary way to build a Declaration, for a row type known at compile time. Where the schema is only settled at run time, DeclarationFromMetadata(md *bigquery.TableMetadata, marshalers ...*Marshalers) reads it from data instead: the row type is fixed to map[string]any, since there is no Go type to derive columns from, and md supplies the schema a struct's tags would otherwise describe. md must not be nil and its Schema must not be empty, since there would then be nothing to check a row's keys against; either is reported here, the same as a struct tag DeclarationOf could not parse.

A row built this way is a plain map[string]any: a key the schema does not declare is an error, and a column the map omits becomes NULL, the same as a struct field would. RowFiller has no way in — a map cannot implement it — so a caller wanting a column like _ingestion_at fills it into the map directly, before Sink.

Options

NewSinker takes two, and none of them describes the table. What the table looks like belongs to the row type, which reaches NewSinker as a Declaration; these settle how bqsink behaves around that instead. How rows travel is settled earlier still, on the writer's own constructor — see Transports.

Option	Default	Section
WithMigrationStrategy	AppendNewColumns{CreateIfMissing: true}, four retries	Migration
WithLogger	records are discarded	Logging

Per-type marshaling overrides are not an Option either: DeclarationOf and DeclarationFromMetadata both take marshalers ...*Marshalers themselves, so that they settle how a value is written the same way they settle everything else about the table — see Custom column types.

A migration strategy is configured with a struct literal rather than an option of its own, so that its settings never look like one:

bqsink.WithMigrationStrategy(bqsink.SyncAllColumns{}, bqsink.DefaultRetryPolicy)

A writer's settings take the same shape — LoadJobs and StorageWrite are struct literals too, configured where they are built rather than through an Option.

Declaring the schema

Struct tags describe the columns. Columns are NULLABLE by default, unlike bigquery.InferSchema.

type Row struct {
	ID       string            `bqsink:"id,required"`     // REQUIRED
	Name     string            `bqsink:"name"`            // NULLABLE
	Seen     time.Time         `bqsink:"seen,nullifzero"` // zero value becomes NULL
	Tags     []string          `bqsink:"tags"`            // REPEATED STRING
	Attrs    map[string]string `bqsink:"attrs"`           // JSON
	Detail   Detail            `bqsink:"detail,record"`   // RECORD
	Internal string            `bqsink:"-"`               // dropped
}

Option	Effect
required	the column is REQUIRED rather than NULLABLE
nullifzero	a zero value is written as NULL, using IsZero() where the type has one
record	a struct expands into a RECORD rather than becoming JSON
date	a time.Time becomes a DATE rather than a TIMESTAMP
datetime	a time.Time becomes a DATETIME
time	a time.Time becomes a TIME
-	the field appears in neither the schema nor the rows

required and nullifzero cannot be combined. On a repeated column nullifzero means "no elements", so a nil or empty slice becomes NULL.

Writing a time.Time as a DATE, DATETIME or TIME

Carrying a calendar day around as a time.Time is ordinary Go, so the column it becomes is a tag away:

type Booking struct {
	CreatedAt time.Time `bqsink:"created_at"`                  // TIMESTAMP
	StayOn    time.Time `bqsink:"stay_on,date"`                // DATE
	CheckInAt time.Time `bqsink:"check_in_at,datetime"`        // DATETIME
	OpensAt   time.Time `bqsink:"opens_at,time"`               // TIME
}

Each option drops what the column does not record: the time of day, the UTC offset, and the date. The conversion reads the value's own location, so the calendar a column records is chosen by handing over a time.Time already in it:

b.StayOn = t.In(jst) // records the Tokyo date, which past 15:00 UTC is tomorrow

That is why there is no separate timezone option. t and t.In(jst) are the same instant, so a TIMESTAMP column is unaffected either way.

These are the only options that change a column's type away from the Go type's own. Dropping a component is the whole conversion — there is no rounding to choose and nothing that can fail. A conversion needing either, such as a float64 written to an INTEGER column, belongs in a FieldMarshaler or MarshalFunc where the caller states the policy; declare the Go field as int64 if that is what the column is. Only time.Time takes these options, not a named type whose underlying type is time.Time, and not alongside record or one another.

Type mapping

STRING     string
BOOL       bool
INTEGER    int, int8, int16, int32, int64, uint8, uint16, uint32
FLOAT      float32, float64
BYTES      []byte
TIMESTAMP  time.Time
DATE       civil.Date
TIME       civil.Time
DATETIME   civil.DateTime
NUMERIC    big.Rat, uint, uint64
JSON       a struct, a map with string keys, json.RawMessage, or any

A type carrying structure BigQuery has no column type for becomes JSON. That covers structs, maps with string keys, json.RawMessage and any. JSON leaves the shape inside the column free, so adding a field to a nested struct needs no migration; record opts back into a RECORD, which keeps the columnar layout that lets BigQuery read one nested field without scanning the rest.

uint and uint64 become NUMERIC rather than INTEGER, because BigQuery's INTEGER is a signed INT64 and cannot hold the upper half of a uint64.

Embedded structs are promoted into the outer struct, following the rules of encoding/json. An embedded type with no exported fields, such as a sync.Mutex, contributes no columns.

Columns tags cannot express

For BIGNUMERIC precision, policy tags and the like, spell the schema out on the row type itself. There is no option for declaring a schema: what the table holds is said in one place, next to the fields it describes.

type Row struct {
	Amount *big.Rat `bqsink:"amount"`
}

func (Row) BigQueryTableMetadata() *bigquery.TableMetadata {
	return &bigquery.TableMetadata{
		Schema: bigquery.Schema{
			{Name: "amount", Type: bigquery.BigNumericFieldType, Precision: 38, Scale: 9,
				Description: "billed amount"},
		},
	}
}

The same method carries partitioning, clustering, labels and expiration, so a type that already has one simply adds Schema to it.

Every column the struct would write has to be present in the schema; NewSinker fails otherwise rather than letting BigQuery reject the write. A schema wider than the struct is fine — the extra columns stay NULL.

Partitioning, clustering and descriptions

The physical layout lives in tag keys of its own, because Go's convention for a tag string is a concatenation of space-separated key:"value" pairs and a description may contain the commas a comma-separated list cannot:

type AccessLog struct {
	Timestamp time.Time `bqsink:"timestamp,required" partition:"day"`
	UserID    string    `bqsink:"user_id" cluster:"1"`
	Region    string    `bqsink:"region" cluster:"2"`
	Amount    *big.Rat  `bqsink:"amount" description:"billed amount, including tax"`
}

Key	Value
partition	day (the default for an empty value), hour, month or year, optionally followed by ,require to demand a partition filter on every query
cluster	the column's position, counting from 1; the order decides how well BigQuery can prune
description	documents the column

A layout BigQuery would refuse is rejected before anything is written rather than at CREATE TABLE:

• one partitioning column per table, not repeated, and TIMESTAMP, DATE or DATETIME
• partition:"hour" on a DATE column — the one granularity DATE cannot carry
• more than four clustering columns, a repeated position, or a gap in the positions
• clustering on FLOAT, JSON, BYTES or a repeated column

require needs no check of its own: it is written inside the partition tag, so there is no way to ask for a partition filter without a partitioning column.

Table level settings

A description and labels have no column to hang off, so they go on an embedded bqsink.TableMeta, which contributes no column of its own:

type AccessLog struct {
	bqsink.TableMeta `description:"one row per request" labels:"team=data,env=prod"`

	Timestamp time.Time `bqsink:"timestamp,required" partition:"day"`
	UserID    string    `bqsink:"user_id"`
}

Key	Value
description	documents the table
labels	a key=value,key=value list; a value may be empty, a key may not

Neither separator needs escaping, since BigQuery allows only lowercase letters, digits, underscores and dashes in a label's key and value. Only these two keys are read there — a bqsink, partition or cluster tag on TableMeta is rejected rather than ignored, and a TableMeta reached through another embedded struct is not searched for.

Anything else — expiration, and whatever the tags do not cover — comes from a method:

func (AccessLog) BigQueryTableMetadata() *bigquery.TableMetadata {
	return &bigquery.TableMetadata{
		ExpirationTime: time.Now().Add(30 * 24 * time.Hour),
	}
}

Declaring the same thing both ways is an error rather than one silently winning: NewSinker fails if the metadata sets Description, Labels, TimePartitioning, RangePartitioning or Clustering that a tag also settles.

Custom column types

A type can declare the column it becomes:

func (Payload) BigQueryFieldType() bigquery.FieldType { return bigquery.JSONFieldType }

func (p Payload) MarshalBigQueryValue() (bigquery.Value, error) {
	b, err := json.Marshal(p)
	return string(b), err
}

For types you do not own, register the mapping instead, passed to DeclarationOf or DeclarationFromMetadata rather than to an Option:

bqsink.DeclarationOf[Row](
	bqsink.MarshalFunc(bigquery.StringFieldType, func(id ExternalID) (bigquery.Value, error) {
		return id.String(), nil
	}),
)

A registered mapping wins over the type's own FieldMarshaler.

Columns bqsink fills in

Embed IngestionMetadata to get columns describing how the row was written:

type AccessLog struct {
	bqsink.IngestionMetadata          // _ingestion_at, _ingestion_id, _ingestion_row_id
	UserID string `bqsink:"user_id"`
}

Column	Value
_ingestion_at	when Sink was called for the row
_ingestion_id	identifies the Sinker, a version 7 UUID
_ingestion_row_id	identifies the row, a version 7 UUID

_ingestion_row_id is what makes at-least-once delivery workable. A retried row keeps the same id, so duplicates are identifiable downstream. Version 7 UUIDs sort by the time they were made, which also suits a clustering key.

Write your own type when the names or the values need to differ:

type Provenance struct {
	At    time.Time `bqsink:"ingested_at"`
	RunID string    `bqsink:"run_id"`
}

func (p *Provenance) FillRow(_ context.Context, info bqsink.AppendInfo) error {
	p.At = info.Time
	p.RunID = os.Getenv("WORKFLOW_RUN_ID")
	return nil
}

type AccessLog struct {
	Provenance
	UserID string `bqsink:"user_id"`
}

FillRow runs once per row, on a copy, before the conversion and before any retry. So a value made inside it — time.Now(), a ULID, a trace id — does not drift when the row is retried. AppendInfo carries only what the row cannot work out for itself: the destination, the Sinker's id and creation time, the row's id and time.

It needs a pointer receiver; with a value receiver it would fill a copy that is then discarded, and NewSinker rejects that.

Migration

The first Sink reads the table's state, asks the strategy what to change, and applies the answer before writing anything. There is no separate method for it: NewSinker already knows the declared schema from the Declaration it was given, and the first batch is simply the earliest point at which reconciling it with BigQuery becomes unavoidable — a Sinker with no rows to write has nothing to reconcile.

Strategy	Behaviour
AppendNewColumns{}	adds missing columns, relaxes REQUIRED to NULLABLE
SyncAllColumns{}	the above, and drops columns the declaration no longer has
MigrationNone{}	leaves the schema alone

The default is AppendNewColumns{CreateIfMissing: true}: following the declaration is what bqsink is for, and neither change it makes destroys anything. So the example above creates the table if it is absent and adds a column when the struct gains a field.

All three take CreateIfMissing. Where it is off and the table does not exist, Sink returns an error wrapping ErrTableMissing rather than creating it.

bqsink.WithMigrationStrategy(bqsink.MigrationNone{})  // write to a table something else owns

bqsink.WithMigrationStrategy(bqsink.SyncAllColumns{
	IgnoreColumns:   []string{"managed_elsewhere"},
	CreateIfMissing: true,
})

IgnoreColumns names columns bqsink does not manage: they are neither dropped nor reported as conflicts.

A difference BigQuery cannot reconcile — a changed type, NULLABLE turning REQUIRED, or a change inside a RECORD — is reported as an error wrapping ErrSchemaConflict rather than migrated.

The migration runs once per Sinker and caches its outcome, success or failure alike. Recovering from a failure means building a new Sinker.

Transports

A writer is built directly from the transport's own settings, before there is a Sinker to hand it to. There is no default transport any more — an unset Option used to mean &StorageWrite{}; now the choice is which constructor you call:

w, err := (&bqsink.StorageWrite{}).NewWriter(client, relation)
w, err := (&bqsink.LoadJobs{}).NewWriter(client, relation)

Neither talks to BigQuery yet: NewWriter only builds the writer, and what happens over the network happens on the first Sink, through the writer NewSinker was given.

Every WriteRows returns a WriteResult, whose Wait reports what that call's rows are settled as. That settlement is the writer's own promise, not the same for every one: delivery to BigQuery for a writer such as StorageWrite, or only acceptance into a buffer of its own for one such as a LoadJobsWriter with LoadJobs.FlushRows set. Using Sink hides most of this: it calls WriteRows and blocks on Wait before returning, so the (n, err) Sink gives back already is that settlement — but what it settles is still whichever promise the writer made, not necessarily delivery (see Flushing rows below).

SinkAsync does the same batch reading, type checking, first-call reconciliation, per-row FillRow and conversion that Sink does, but hands back the WriteResult without calling Wait on it, leaving that to the caller. It suits a caller keeping several batches in flight at once, or one that wants to decide for itself when, or whether, to wait. Sink's own guard against a writer under-reporting — returning an error when Wait reports fewer rows than were given — is not repeated here, since SinkAsync returns before there is anything to compare Wait's answer against; a caller wanting the same guard compares the n its own call to Wait returns against how many rows it gave SinkAsync.

StorageWrite uses the BigQuery Storage Write API. Each append is all or nothing: none of the rows in a rejected request land. Only the default and committed stream types are supported.

LoadJobs renders the rows as newline delimited JSON and submits a load job, blocking until BigQuery finishes it, which takes seconds to minutes. A load job is all or nothing too. Nothing is serialised on a lock, so concurrent calls submit concurrent jobs — and a table's daily job quota is why rows belong in batches rather than one call each; FlushRows gathers several calls into fewer jobs (see Flushing rows).

Large batches can be staged in Cloud Storage instead of being uploaded with the job:

import "github.com/mashiike/bqsink/bqgcs"

w, err := (&bqsink.LoadJobs{
	Staging: &bqgcs.Staging{Client: gcsClient, Bucket: "staging-bucket", Prefix: "bqsink"},
}).NewWriter(client, relation)

bqgcs is a separate package so that using bqsink does not pull in cloud.google.com/go/storage.

Flushing rows

LoadJobs.FlushRows gathers rows across WriteRows calls and submits them as one load job once that many are held, instead of a job per call:

w, err := (&bqsink.LoadJobs{FlushRows: 10_000}).NewWriter(client, relation)

It is the threshold that submits a batch, not a Wait call: once a WriteRows appends enough rows to reach FlushRows, that call submits the batch itself, under RetryPolicy, and blocks until the job finishes before returning — the way bufio.Writer flushes on a full buffer rather than on Flush. Every earlier call that joined the same batch already returned, its rows merely accepted rather than delivered, so a single goroutine calling Sink on its own, one batch after another, does end up with fewer jobs than calls once enough rows have gone by. Several goroutines calling Sink on the same writer at once share a batch the same way and gain the same thing: whichever of them fills it is the one that submits, and the rest share its job.

FlushRows earns its keep where rows are given to the writer directly through WriteRows — a lower-level entry point than Sink, taking already-marshalled []bqsink.Row, and usable once some Sinker built on this writer has made its first Sink call to bind the schema — gathering them without waiting on each result right away:

r1, err := w.WriteRows(ctx, rows1) // joins the open batch, no job yet
if err != nil {
	return err
}
r2, err := w.WriteRows(ctx, rows2) // joins the same batch
if err != nil {
	return err
}

// later, e.g. from a ticker, or once enough calls have gone by:
res, err := w.FlushRows(ctx)
if err != nil {
	return err
}
if _, err := res.Wait(ctx); err != nil { // the job's own outcome
	return err
}
if _, err := r1.Wait(ctx); err != nil { // already settled by FlushRows
	return err
}
if _, err := r2.Wait(ctx); err != nil {
	return err
}

Discarding the WriteResult FlushRows returns without a Wait leaves no way to learn that job's own outcome, success or failure — err from FlushRows itself only reports whether the flush was accepted, not how the job it started turned out.

Close submits whatever FlushRows is still holding back, so rows gathered but never flushed are not lost to a shutdown; it is also where a threshold submission's failure gets reported if nothing has read it yet — that submission never had a WriteResult of its own to carry the outcome, so Close is the last chance.

Not calling Close, or ignoring the error it returns, can lose rows that a Sink or WriteRows call already reported as accepted. A submission that one of those calls triggered on its own, by filling the buffer to FlushRows, is folded into the error Close returns if nothing else reported it first — skip Close and that outcome reaches no one.

A FlushRows call's own submission is different: its outcome is carried solely by the WriteResult that call returned. Discard that result without a Wait and Close does not recover it either — the rows behind it are lost to every caller, not just the one that called FlushRows.

Retries

Retrying belongs to whoever is holding the rows. A writer still has them while they are in flight, so it is the one that can try again; Sinker never retries a write behind the writer's back.

// Migration: the policy is the second argument, since a strategy is a pure
// decision and retrying is not part of it. nil means one attempt.
bqsink.WithMigrationStrategy(bqsink.SyncAllColumns{}, bqsink.DefaultRetryPolicy)

// LoadJobs retries a load job itself.
(&bqsink.LoadJobs{RetryPolicy: myPolicy}).NewWriter(client, relation)

DefaultRetryPolicy is four retries with jittered backoff between 200ms and 5s, covering a concurrent change to the table, a rate limit, or a server side failure, over either HTTP or gRPC. It is what NewSinker uses for migration when WithMigrationStrategy is not given: two replicas deploying at once add the same column at once, and BigQuery reports that as a failed precondition that only a retry gets past.

StorageWrite has no policy to set: the client library's own automatic retries are the ones that know how to re-enqueue an append on a reconnected stream, so bqsink enables those and the number of attempts is theirs to decide. Set DisableWriteRetries to have bqsink leave them alone.

Writes are at-least-once. A retry can deliver a row twice and neither transport deduplicates.

Logging

bqsink.WithLogger(slog.Default())

Nothing is logged without it. The default discards every record rather than writing to the embedding program's slog.Default() uninvited. Every record names the relation.

Level	What it reports
Debug	the difference the migration found, the stream that was opened, the rows a flush carried
Info	a change made to the table, and a load job submitted and finished
Warn	something bqsink let pass

Error is unused. A failure is returned, and logging it as well would report it twice. What Warn is for is the opposite case, where nothing is returned and the record is the only trace: a failure a retry got past, a schema difference the strategy chose to leave alone, a column drop, or a failure that had to give way to a worse one — such as a stream that would not close after rows had already been lost.

Testing

go test ./...

Unit tests reach no network. Integration tests write to a real project and skip unless the environment names one:

BQSINK_TEST_PROJECT=my-project go test ./...
BQSINK_TEST_PROJECT=my-project BQSINK_TEST_BUCKET=my-bucket go test ./...

Variable	Effect
BQSINK_TEST_PROJECT	required; without it every integration test skips
BQSINK_TEST_DATASET	dataset to use, created if absent (bqsink_integration_test)
BQSINK_TEST_BUCKET	required only for the Cloud Storage staging tests

Each test uses a uniquely named table and removes it afterwards.

Not supported

• Migrating a change inside a RECORD. Detected and reported, not applied. Only reachable through the record tag, since a struct is JSON by default
• Pending and buffered streams. They need their rows committed or their offset flushed, which bqsink does not do, so they are rejected rather than silently losing rows
• Exactly-once delivery

License

MIT

Documentation

Overview

Package bqsink writes rows to BigQuery and keeps the destination table's schema in sync with a schema declared in Go code.

The declaration is the source of truth: the real table follows it, and bqsink never infers a schema from the data being written. What the table looks like is said by the row type and nowhere else — its struct tags, and its BigQueryTableMetadata method for what tags cannot express. No Option describes the table, since a row type carries the domain knowledge that gives its columns meaning, and two places to say what a table is means two answers to keep agreeing.

Writing rows

Writing takes two things. A RowsWriter holds what writing depends on — the table, the connection, the transport, how a transient failure is retried — and a Sinker holds what none of that changes: the declaration, bringing the real table in line with it, and turning a row into columns.

w, err := (&bqsink.LoadJobs{}).NewWriter(client, relation)
if err != nil {
	return err
}
defer w.Close(ctx)

s, err := bqsink.NewSinker(w, bqsink.DeclarationOf[AccessLog]())
if err != nil {
	return err
}
n, err := s.Sink(ctx, logs)

The declaration reaches NewSinker rather than being picked up from the first batch, so a mistake in it is reported before anything is written. Nothing contacts BigQuery until the first Sink, which is what reconciles the real table with the declaration and hands the writer the settled schema.

Sink returns a non-nil error whenever n is fewer than the rows it was given, so that rows[n:] are exactly the ones that did not. What counts as done there is the writer's own promise: how many reached BigQuery for a writer promising delivery, and only how many reached its own buffer for one promising acceptance instead, such as a LoadJobsWriter with LoadJobs.FlushRows set — what becomes of those once a job carries them is for FlushRows or Close to report, not Sink. A Sinker itself buffers nothing between calls — the rows handed to one Sink are the batch — though the writer it hands them to may. Closing belongs to the writer, since that is what holds a connection, and a Sinker has nothing waiting.

The Options settle how bqsink behaves around the declaration: what to do about a difference between it and the real table, and what gets logged. How rows travel and how a failed write is retried are settled on the writer instead.

Declaring the columns

A row type's struct tags describe its columns, and the per-type overrides passed to DeclarationOf or DeclarationFromMetadata refine how their values are written.

This differs from bigquery.InferSchema in two ways that matter in practice: columns are NULLABLE by default, and the "bqsink" tag is read instead of "bigquery". Mark a column REQUIRED with `bqsink:",required"`.

The tag's first element renames the column; an empty one keeps the Go field name verbatim, with no conversion to snake_case. A tag of `bqsink:"-"` drops the field, so it appears in neither the schema nor the rows written. Unexported fields are always dropped.

An embedded struct's fields are promoted into the outer struct, following the rules of encoding/json: a shallower field hides a deeper one of the same name, an explicit tag breaks a tie at equal depth, an unresolved tie removes that one column while leaving the rest promoted, and the columns come out in field declaration order. Naming an embedded field in its tag makes it a column of its own rather than something to descend into. An embedded type with no exported fields, such as a sync.Mutex, therefore contributes no columns at all.

The options after the name are:

required    the column is REQUIRED rather than NULLABLE
nullifzero  a zero value is written as NULL
record      a struct expands into a RECORD rather than becoming JSON
date        a time.Time becomes a DATE rather than a TIMESTAMP
datetime    a time.Time becomes a DATETIME
time        a time.Time becomes a TIME

"required" and "nullifzero" cannot be combined, since a REQUIRED column cannot hold NULL. Neither can two options that name the column's type, so "record" and the three below exclude one another.

"date", "datetime" and "time" each drop what the column does not record: the time of day, the UTC offset, and the date. They read the value's own location, so the calendar a column records is chosen by handing over a time.Time already in it, and no separate timezone option is needed:

Day time.Time `bqsink:"day,date"`  // time.Now().In(jst) records the Tokyo date

They are the only options that change a column's type from the Go type's own, because dropping a component is the whole conversion: there is no rounding to choose and nothing that can fail. A conversion needing either, such as a float64 written to an INTEGER column, belongs in a FieldMarshaler or MarshalFunc where the caller states the policy. Only time.Time takes them; a named type whose underlying type is time.Time does not, since bqsink cannot see through it.

"nullifzero" decides what counts as zero the way the "omitzero" option of encoding/json/v2 does: through an IsZero method where the type has one, and by the zero Go value otherwise. That is what makes a zero time.Time recognisable. On a repeated column it means no elements, so both a nil and an empty slice become NULL; without it they become an empty array, which BigQuery keeps distinct from NULL.

Separate tag keys describe the table rather than the column:

partition:"day"           partition by this column, by day
partition:"hour,require"  by hour, and demand a partition filter
cluster:"1"               the first clustering column
description:"..."         document the column

How Go types become columns

Go types map to BigQuery types as follows.

STRING     string
BOOL       bool
INTEGER    int, int8, int16, int32, int64, uint8, uint16, uint32
FLOAT      float32, float64
BYTES      []byte
TIMESTAMP  time.Time
DATE       civil.Date, or a time.Time tagged "date"
TIME       civil.Time, or a time.Time tagged "time"
DATETIME   civil.DateTime, or a time.Time tagged "datetime"
NUMERIC    big.Rat, uint, uint64
JSON       a struct, a map with string keys, json.RawMessage, or any

uint and uint64 become NUMERIC rather than INTEGER because BigQuery's INTEGER is INT64, which is signed and cannot hold the upper half of a uint64. BIGINT does not help, being an alias of INT64. The column is then no longer an integer type, so prefer int64 where the values allow it.

A slice or array becomes a REPEATED field of its element type, except that a slice of bytes becomes BYTES. Pointers are followed, so *string is a NULLABLE STRING; a pointer does not by itself make a column NULLABLE, since that is already the default.

A type that carries structure BigQuery has no column type for becomes JSON. That covers structs, maps with string keys, json.RawMessage and any. Keeping a struct out of a JSON column and expanding it into a RECORD takes the "record" option: `bqsink:"inner,record"`. JSON leaves the shape inside the column free, so adding a field to a nested struct needs no migration, while a RECORD keeps the columnar layout that lets BigQuery read one nested field without scanning the rest.

A json.RawMessage is written through unchanged, since it already holds JSON text. Everything else is encoded with encoding/json, without escaping HTML, so that a URL stays readable rather than arriving full of &.

A type that implements FieldMarshaler, or one registered through MarshalFunc, takes the column type it declares instead of any of the above.

Types with no representation at all, including uintptr, a map with non-string keys, channels and functions, produce an error. Give the row type a BigQueryTableMetadata method spelling the schema out for columns none of this can express.

Index

• Constants
• Variables
• func DefaultRetryPolicy() gax.Retryer
• type AppendInfo
• type AppendNewColumns
  • func (a AppendNewColumns) Plan(state TableState, logger *slog.Logger) (SchemaChange, error)
• type ConflictReason
  • func (r ConflictReason) String() string
• type Declaration
  • func DeclarationFromMetadata(md *bigquery.TableMetadata, marshalers ...*Marshalers) Declaration
  • func DeclarationOf[T any](marshalers ...*Marshalers) Declaration
• type FieldMarshaler
• type Flusher
• type IngestionMetadata
  • func (m *IngestionMetadata) FillRow(_ context.Context, info AppendInfo) error
• type LoadJobs
  • func (w *LoadJobs) NewWriter(client *bigquery.Client, relation Relation) (*LoadJobsWriter, error)
  • func (w *LoadJobs) Validate() error
• type LoadJobsWriter
  • func (w *LoadJobsWriter) BindLogger(logger *slog.Logger)
  • func (w *LoadJobsWriter) BindSchema(_ context.Context, schema bigquery.Schema) error
  • func (w *LoadJobsWriter) Client() *bigquery.Client
  • func (w *LoadJobsWriter) Close(ctx context.Context) error
  • func (w *LoadJobsWriter) FlushRows(ctx context.Context) (WriteResult, error)
  • func (w *LoadJobsWriter) Relation() Relation
  • func (w *LoadJobsWriter) WriteRows(ctx context.Context, rows []Row) (WriteResult, error)
• type LoggerBindable
• type Marshalers
  • func MarshalFunc[T any](fieldType bigquery.FieldType, fn func(T) (bigquery.Value, error)) *Marshalers
• type MigrationNone
  • func (m MigrationNone) Plan(state TableState, logger *slog.Logger) (SchemaChange, error)
• type MigrationStrategy
• type Option
  • func WithLogger(logger *slog.Logger) Option
  • func WithMigrationStrategy(s MigrationStrategy, retryPolicy func() gax.Retryer) Option
• type Relation
  • func ParseRelation(s string) (Relation, error)
  • func (r Relation) String() string
• type Row
• type RowFiller
• type RowsWriter
• type SchemaChange
  • func (c SchemaChange) Empty() bool
• type SchemaConflict
  • func (c SchemaConflict) String() string
• type SchemaDiff
  • func DiffSchema(want, got bigquery.Schema) SchemaDiff
  • func (d SchemaDiff) Empty() bool
  • func (d SchemaDiff) Without(names []string) SchemaDiff
• type Sinker
  • func NewSinker(w RowsWriter, decl Declaration, opts ...Option) (*Sinker, error)
  • func (s *Sinker) Sink(ctx context.Context, rows any) (int, error)
  • func (s *Sinker) SinkAsync(ctx context.Context, rows any) (WriteResult, error)
  • func (s *Sinker) Start(ctx context.Context) error
• type Stager
• type StorageWrite
  • func (w *StorageWrite) NewWriter(client *bigquery.Client, relation Relation) (*StorageWriter, error)
  • func (w *StorageWrite) Validate() error
• type StorageWriter
  • func (w *StorageWriter) BindLogger(logger *slog.Logger)
  • func (w *StorageWriter) BindSchema(ctx context.Context, schema bigquery.Schema) error
  • func (w *StorageWriter) Client() *bigquery.Client
  • func (w *StorageWriter) Close(ctx context.Context) error
  • func (w *StorageWriter) Relation() Relation
  • func (w *StorageWriter) WriteRows(ctx context.Context, rows []Row) (WriteResult, error)
• type SyncAllColumns
  • func (s SyncAllColumns) Plan(state TableState, logger *slog.Logger) (SchemaChange, error)
  • func (s SyncAllColumns) Validate() error
• type TableDefiner
• type TableMeta
• type TableState
• type Validator
• type WriteResult
  • func ResolvedResult(n int, err error) WriteResult

Constants

const (
	// TagKey describes the column itself: its name and how its value is treated.
	TagKey = "bqsink"

	// PartitionTagKey makes the column the table's partitioning column. Its value
	// is the granularity, optionally followed by "require" to demand a partition
	// filter on every query.
	PartitionTagKey = "partition"

	// ClusterTagKey makes the column one of the table's clustering columns. Its
	// value is the position, counting from 1, since the order decides how well
	// BigQuery can prune.
	ClusterTagKey = "cluster"

	// DescriptionTagKey documents the column, or the table itself when it is on an
	// embedded TableMeta. It is a key of its own because a description may contain
	// the commas and spaces a comma-separated tag cannot.
	DescriptionTagKey = "description"

	// LabelsTagKey carries the table's labels as a "key=value,key=value" list. It
	// is only read from an embedded TableMeta, since a label describes the table
	// rather than any one column.
	LabelsTagKey = "labels"
)

Struct tags bqsink reads. Go's convention for a tag string is a concatenation of space-separated key:"value" pairs, so the physical layout of the table lives in keys of its own rather than crowding into the column's own tag.

type AccessLog struct {
	Timestamp time.Time `bqsink:"timestamp,required" partition:"day"`
	UserID    string    `bqsink:"user_id" cluster:"1"`
	Amount    *big.Rat  `bqsink:"amount" description:"billed amount, including tax"`
}

Variables

var (
	// ErrSchemaConflict reports that reconciling the declared schema with the
	// real table needs a change BigQuery does not allow, such as altering a
	// column's type or making a NULLABLE column REQUIRED.
	ErrSchemaConflict = errors.New("bqsink: schema conflict")

	// ErrTableMissing reports that the table does not exist and the migration
	// strategy did not ask for it to be created. Set CreateIfMissing on the
	// strategy to create it.
	ErrTableMissing = errors.New("bqsink: table does not exist")

	// ErrWriterClosed reports that a writer's Close was already called,
	// distinguishing that from any other reason a call might fail so that a
	// caller racing a graceful shutdown against an in-flight write can tell
	// the two apart. It is always wrapped with which writer and method
	// refused the call, never returned bare.
	ErrWriterClosed = errors.New("the writer is closed")
)

Functions

func DefaultRetryPolicy

func DefaultRetryPolicy() gax.Retryer

DefaultRetryPolicy returns the policy bqsink uses unless WithRetryPolicy replaces it: a failure that a later attempt could get past is retried up to four times, waiting between 200ms and 5s with jitter, and any other error is returned immediately.

It covers both a concurrent change to the table during the migration and a transient failure while writing.

Types

type AppendInfo

type AppendInfo struct {
	// Relation names the destination table.
	Relation Relation

	// SinkerID identifies the Sinker. It is decided in New, so it spans the
	// Sinker's whole life: a batch that builds one Sinker per run gets an id per
	// run, while a long lived process that keeps one around gets a single id until
	// it is replaced.
	//
	// It is a version 7 UUID, so it sorts in the order the Sinkers were built.
	SinkerID string

	// SinkerCreatedAt is when New was called.
	SinkerCreatedAt time.Time

	// RowID identifies this row. It is a version 7 UUID, so it sorts in the order
	// the rows were handed over, which suits a clustering key.
	//
	// A retry of the same row keeps the same value, so it can deduplicate what
	// at-least-once delivery may write twice.
	RowID string

	// Time is when Sink was called for this row. A retry keeps the same value.
	Time time.Time
}

AppendInfo carries what a row cannot work out for itself when it fills its own columns in.

Values a row can produce on its own are deliberately absent. FillRow runs once per row, before the conversion and before any retry, so a row that wants a timestamp or an identifier of its own design can simply make one there and it will not drift across retries.

type AppendNewColumns

type AppendNewColumns struct {
	// CreateIfMissing creates the table when it does not exist.
	CreateIfMissing bool
}

AppendNewColumns adds columns the declaration has and the table lacks, and relaxes REQUIRED columns the declaration marks NULLABLE. Columns the table has and the declaration lacks are left alone.

AppendNewColumns{CreateIfMissing: true} is the default strategy. Neither change it makes destroys anything, which is why it is safe to have on by default; drops need SyncAllColumns, which has to be asked for.

func (AppendNewColumns) Plan

func (a AppendNewColumns) Plan(state TableState, logger *slog.Logger) (SchemaChange, error)

Plan implements MigrationStrategy.

type ConflictReason

type ConflictReason int

ConflictReason says why a difference between the declared schema and the real table cannot be reconciled.

const (
	// ConflictType means the column's type differs. BigQuery does not allow a
	// column's type to be changed by patching the table.
	ConflictType ConflictReason = iota

	// ConflictRepeated means one side is REPEATED and the other is not.
	ConflictRepeated

	// ConflictRequired means the declaration asks for REQUIRED where the table has
	// NULLABLE. Only the opposite direction is allowed.
	ConflictRequired

	// ConflictNested means the fields inside a RECORD differ. bqsink reports this
	// instead of migrating it, so that it cannot pass unnoticed.
	ConflictNested
)

func (ConflictReason) String

func (r ConflictReason) String() string

String implements fmt.Stringer.

type Declaration

type Declaration struct {
	// contains filtered or unexported fields
}

Declaration says what the destination table should hold.

It is evaluated once, when it is built with DeclarationOf: the struct tags describe the columns, and a BigQueryTableMetadata method describes what tags cannot. Evaluating it there settles the table's shape for good, which is why everything the declaration decides is reported there — a tag that cannot be parsed, a column its spelled out schema has no room for, a FillRow with a value receiver — rather than by NewSinker or the first row written.

A constructor here reports nothing itself, so that it composes into the NewSinker call; what it could not make sense of is carried on the Declaration and comes back from NewSinker.

fills is also settled here: whether the row type fills its own columns, so that a Sinker built from the Declaration knows it without asking on every row.

func DeclarationFromMetadata

func DeclarationFromMetadata(md *bigquery.TableMetadata, marshalers ...*Marshalers) Declaration

DeclarationFromMetadata returns what md declares about its table, for a row written as a map[string]any rather than a Go struct: the schema is settled at run time, in md, rather than by any type's tags.

The row type is fixed to map[string]any; there is no type parameter to name a different one. marshalers are the same per-type overrides DeclarationOf takes, looked up by each value's own dynamic type since a map has no fields of its own to carry them.

md must not be nil and md.Schema must not be empty, since there would then be nothing to check a row's keys against or convert its values into; either is reported as an error carried on the Declaration, the same as a struct tag DeclarationOf could not parse.

RowFiller has no effect on a row built this way: a map cannot implement it, so a caller wanting an ingestion column of its own kind fills it into the map directly, before Sink.

md is not retained by reference for its top-level fields, which are copied once here, but its Schema's *bigquery.FieldSchema elements are not deep copied. Mutating one of those after this call is a precondition violation this does not defend against beyond never panicking on it.

func DeclarationOf

func DeclarationOf[T any](marshalers ...*Marshalers) Declaration

DeclarationOf returns what T declares about its table, evaluated immediately.

The row type carries the domain knowledge that gives its columns meaning, so it is the one place that can say what the table holds.

marshalers registers per-type overrides of how a Go type becomes a BigQuery column, built with MarshalFunc. They apply both to the schema derived from the row type's struct tags and to the values written, and they win over a type's own FieldMarshaler. Where a TableDefiner supplies the schema outright the overrides do not reach it, since nothing is derived in that case, and they still decide how the values are written. Passing more than one keeps the mapping registered last; passing none leaves every type to its own FieldMarshaler or bqsink's own rules, which is what every call before marshalers existed did.

type FieldMarshaler

type FieldMarshaler interface {
	// BigQueryFieldType returns the type of the column this value becomes.
	BigQueryFieldType() bigquery.FieldType

	// MarshalBigQueryValue returns the value to write into that column.
	MarshalBigQueryValue() (bigquery.Value, error)
}

FieldMarshaler lets a Go type declare the BigQuery column it becomes, overriding what bqsink would otherwise derive from the Go type alone.

Implementing it settles both halves of the mapping at once: BigQueryFieldType says what the column is in the schema, and MarshalBigQueryValue says what is written into it. A struct that would otherwise become a RECORD can therefore be written as a JSON, STRING or BYTES column instead.

type Payload struct{ Data map[string]string }

func (Payload) BigQueryFieldType() bigquery.FieldType { return bigquery.JSONFieldType }

func (p Payload) MarshalBigQueryValue() (bigquery.Value, error) {
	b, err := json.Marshal(p.Data)
	return string(b), err
}

Either a value or a pointer receiver works. Use Marshalers instead for types whose definition is out of reach, such as those from another package.

RECORD is not accepted, because its nested schema cannot be derived from a field type alone; spell such a column out in BigQueryTableMetadata.

type Flusher

type Flusher interface {
	// FlushRows submits the rows buffered so far as one job and returns a
	// WriteResult already resolved with the outcome.
	FlushRows(ctx context.Context) (WriteResult, error)
}

Flusher is implemented by a writer that buffers rows of its own, letting a caller send what it holds without waiting for enough rows to arrive on their own.

type IngestionMetadata

type IngestionMetadata struct {
	// IngestionAt is when Sink was called for the row, which a batching transport
	// defers from when the row reaches BigQuery.
	IngestionAt time.Time `bqsink:"_ingestion_at"`

	// IngestionID identifies the ingestion the row belongs to, which is one
	// Sinker's lifetime. It comes from AppendInfo.SinkerID and is decided in New,
	// so building a Sinker per batch gives an id per batch, while a long lived
	// process holding one Sinker writes every row under the same id.
	//
	// It does not change from one Sink to the next; _ingestion_at and
	// _ingestion_row_id do.
	IngestionID string `bqsink:"_ingestion_id"`

	// IngestionRowID identifies the row and stays the same across a retry, so it
	// can deduplicate rows written more than once.
	IngestionRowID string `bqsink:"_ingestion_row_id"`
}

IngestionMetadata is an embeddable set of columns describing how a row was written, which bqsink fills in.

The columns are named with a leading underscore so that they sort ahead of the business columns and read as belonging to the pipeline rather than the data. BigQuery allows a column name to start with an underscore; its own pseudo columns such as _PARTITIONTIME are upper case, so these do not collide.

Embed it to get the three columns below; write a type of your own implementing RowFiller when the names or the values need to differ.

type AccessLog struct {
	bqsink.IngestionMetadata
	UserID string `bqsink:"user_id"`
}

func (*IngestionMetadata) FillRow

func (m *IngestionMetadata) FillRow(_ context.Context, info AppendInfo) error

FillRow implements RowFiller.

type LoadJobs

type LoadJobs struct {
	// Staging, when set, writes the rows through a Stager and has the load job
	// read them from there instead of carrying them itself.
	Staging Stager

	// RetryPolicy decides how a load job that failed in a way a later attempt
	// could get past is retried. The zero value means DefaultRetryPolicy; set it
	// to a policy of your own to change that, and note that returning a bare
	// gax.OnErrorFunc places no limit on the number of attempts.
	//
	// It is called once per load job, because a gax.Retryer carries the state of
	// its backoff and cannot be reused.
	RetryPolicy func() gax.Retryer

	// FlushRows gathers rows across calls into a buffer and submits them as one
	// load job once this many are held, rather than submitting a job per
	// WriteRows. A table's daily job quota is what makes that worth doing:
	// writing a row at a time without it submits a load job each time.
	//
	// The zero value submits a job per WriteRows, and that call's WriteResult
	// reports the job's own outcome.
	//
	// With FlushRows set, WriteRows itself only ever reports that the rows were
	// taken into the buffer, not that a job has run. A submission triggered by
	// the buffer reaching this many rows, or one a later call to FlushRows or
	// Close makes, is reported by whichever of those methods makes it, not by
	// the WriteResult of the WriteRows call that filled the buffer.
	//
	// A buffer reaching this many rows is submitted by the call that filled it,
	// which is what keeps the rows held here bounded, and FlushRows or Close
	// submits what is left. The call that fills the buffer blocks until that
	// job finishes, under RetryPolicy, before returning.
	FlushRows int
}

LoadJobs writes rows by rendering them as newline delimited JSON and submitting a BigQuery load job. It is the settings a LoadJobsWriter is made from.

A load job is all or nothing: the rows of one job either all land or, having been retried under RetryPolicy, none do. Rows are never left half written.

Rows are uploaded with the load job itself unless Staging is set. Set it to bqgcs.Staging to put them in Cloud Storage first, which suits large batches.

func (*LoadJobs) NewWriter

func (w *LoadJobs) NewWriter(client *bigquery.Client, relation Relation) (*LoadJobsWriter, error)

NewWriter returns a writer that loads rows into the table relation names.

It does not contact BigQuery: a load job is the first thing that does, and the schema it carries arrives with BindSchema. An empty ProjectID on relation is filled in from the client.

func (*LoadJobs) Validate

func (w *LoadJobs) Validate() error

Validate implements Validator.

type LoadJobsWriter

type LoadJobsWriter struct {
	// contains filtered or unexported fields
}

LoadJobsWriter writes rows to one BigQuery table with load jobs.

Its WriteRows never defers what it reports: with LoadJobs.FlushRows unset, it submits a load job for the call's own rows on the spot and reports that job's outcome. With FlushRows set, it instead reports that the rows were accepted into a buffer held here, and what becomes of them once a job carries them is FlushRows's or Close's to report, not WriteRows's.

A buffer reaching FlushRows is submitted by the call that filled it, which is what keeps the rows held here bounded. A failure of that submission is kept until FlushRows or Close next runs, rather than handed to any particular caller.

func (*LoadJobsWriter) BindLogger

func (w *LoadJobsWriter) BindLogger(logger *slog.Logger)

BindLogger implements LoggerBindable.

func (*LoadJobsWriter) BindSchema

func (w *LoadJobsWriter) BindSchema(_ context.Context, schema bigquery.Schema) error

BindSchema implements RowsWriter. The schema is what a load job declares, so nothing can be written before it arrives.

func (*LoadJobsWriter) Client

func (w *LoadJobsWriter) Client() *bigquery.Client

Client implements RowsWriter.

func (*LoadJobsWriter) Close

func (w *LoadJobsWriter) Close(ctx context.Context) error

Close waits for a submission still in flight, whether WriteRows started it on its own by filling the buffer to FlushRows or a FlushRows call started it, then submits whatever the buffer still holds and releases the writer. Closing twice in sequence is harmless: the second call finds nothing left to submit or report and returns nil. Two concurrent calls are not so evenly matched: whichever observes the writer already closed returns nil at once, without waiting for the other to finish, so it learns nothing about the outcome that call goes on to report.

ctx bounds only the submission this call makes of whatever the buffer still holds; it does not bound the wait for a submission already in flight, so Close can run past ctx's own deadline if that submission is retrying under a RetryPolicy with no bound of its own.

What it folds into the error it returns, though, is narrower than what it waits for: only a WriteRows submission's outcome is kept in pending, so only that is reported here. A FlushRows call's own submission is never folded into pending; its outcome was carried solely by the WriteResult that call returned, and if nothing ever called Wait on it, Close does not recover it either — that outcome is lost to every caller, not just the one that made the FlushRows call.

func (*LoadJobsWriter) FlushRows

func (w *LoadJobsWriter) FlushRows(ctx context.Context) (WriteResult, error)

FlushRows implements Flusher. It submits whatever rows the buffer holds as one load job, under RetryPolicy, and returns a WriteResult already resolved with that job's outcome: n is how many of the rows this call itself sent landed, which is 0 when the job fails.

A submission WriteRows made on its own, by filling the buffer to FlushRows, is not reported when it happens: that call's own WriteResult already reported acceptance. Its outcome is kept until the next call to FlushRows or to Close, which folds it into the error returned here and then clears it, so it is reported exactly once — and only there: a caller that lets this WriteResult go without calling Wait on it loses that outcome for good, even though the rows behind it were never this call's own to report. Unlike Close, FlushRows does not wait for a submission still in flight, so that outcome may instead be left for whichever of the two runs next.

The outcome of this call's own submission, in contrast, is carried solely by the WriteResult returned here: it is never folded into pending, so it is never reported by a later FlushRows or by Close either. A caller that lets this WriteResult go without calling Wait on it has no way left to learn whether these rows landed.

func (*LoadJobsWriter) Relation

func (w *LoadJobsWriter) Relation() Relation

Relation implements RowsWriter.

func (*LoadJobsWriter) WriteRows

func (w *LoadJobsWriter) WriteRows(ctx context.Context, rows []Row) (WriteResult, error)

WriteRows implements RowsWriter.

With LoadJobs.FlushRows unset, it submits a load job for these rows on the spot, under RetryPolicy, and the WriteResult it returns is already resolved with that job's outcome.

With FlushRows set, it appends the rows to the buffer and returns a WriteResult resolved with their acceptance: len(rows) and a nil error, whether or not the append goes on to submit a job. When appending fills the buffer to FlushRows, this call also submits it on the spot, under RetryPolicy, and blocks until that job finishes before returning, but a failure of that submission is kept for FlushRows or Close to report rather than reflected in the WriteResult returned here.

type LoggerBindable

type LoggerBindable interface {
	BindLogger(logger *slog.Logger)
}

LoggerBindable is implemented by a writer with something to log. NewSinker hands it the logger WithLogger settled on, already carrying the relation.

type Marshalers

type Marshalers struct {
	// contains filtered or unexported fields
}

Marshalers is a list of per-type overrides, built with MarshalFunc and passed to DeclarationOf or DeclarationFromMetadata.

Despite the name it is not a collection of FieldMarshaler. FieldMarshaler is implemented by a type on its own behalf; Marshalers registers mappings from the outside, for types whose definition cannot be changed. A registered mapping wins over a type's own FieldMarshaler, since the caller asked for it explicitly.

A nil *Marshalers is equivalent to an empty list.

func MarshalFunc

func MarshalFunc[T any](fieldType bigquery.FieldType, fn func(T) (bigquery.Value, error)) *Marshalers

MarshalFunc constructs a type-specific marshaler that writes values of type T into a column of fieldType, converting them with fn.

T is inferred from fn, so it never has to be written out. It must not be a pointer type: a mapping registered for T already covers a *T field, and registering *T instead would never be found. Registering the same type twice keeps the mapping registered last.

RECORD is not a usable fieldType, because its nested schema cannot be derived from a field type alone; spell such a column out in BigQueryTableMetadata.

A problem with the arguments is reported when DeclarationOf or DeclarationFromMetadata evaluates the declaration, since a constructor cannot return an error and stay usable inline; it surfaces from NewSinker as the Declaration's own error.

type MigrationNone

type MigrationNone struct {
	// CreateIfMissing creates the table when it does not exist.
	CreateIfMissing bool
}

MigrationNone leaves an existing table's schema untouched.

It is not the default: AppendNewColumns is, since following the declaration is what bqsink is for. Choose this to write to a table something else owns.

func (MigrationNone) Plan

func (m MigrationNone) Plan(state TableState, logger *slog.Logger) (SchemaChange, error)

Plan implements MigrationStrategy.

It asks for no change beyond creating a missing table, but still reports conflicts, because writing to a table whose columns disagree with the declaration would fail anyway.

type MigrationStrategy

type MigrationStrategy interface {
	Plan(state TableState, logger *slog.Logger) (SchemaChange, error)
}

MigrationStrategy decides what to do about the difference between the declared schema and the real table.

Plan touches nothing: bqsink reads the table's state, asks the strategy what to change, and applies the answer. An implementation therefore needs no BigQuery access and can be tested on its own.

logger is the one WithLogger settled on and is never nil. It is there for the one thing the answer cannot express: a difference the strategy decided not to reconcile. Since bqsink is for keeping a table in step with the declaration, leaving a difference alone is worth a record, and the SchemaChange returned no longer mentions it.

type Option

type Option func(*config) error

Option configures a Sinker at construction time.

Options are the only place bqsink uses this pattern. A migration strategy is a struct instead, so that its settings never look like an Option, and how rows travel is settled on the writer rather than here.

No Option describes the table. What the table looks like belongs to the row type, through its struct tags and its BigQueryTableMetadata method, and reaches NewSinker as a Declaration, so that one place answers what the table should be. The Options here settle how bqsink behaves around that declaration: what to do about a difference from the real table, and what gets logged.

func WithLogger

func WithLogger(logger *slog.Logger) Option

WithLogger sends what bqsink has to say to logger, with the relation as an attribute on every record.

Without it nothing is logged at all: the default discards every record, so bqsink does not write to an embedding program's slog.Default() uninvited.

The writer is handed the same logger where it has something to log, so one call covers both sides.

Three levels are used and Error is not among them, since a failure is returned rather than logged.

Debug  what a transport did: the stream it opened, the rows it wrote, the
       difference the migration found
Info   a change bqsink made to the table, and a load job it ran
Warn   something bqsink had to let pass: a failure it could not return, or a
       difference the migration strategy chose not to reconcile

func WithMigrationStrategy

func WithMigrationStrategy(s MigrationStrategy, retryPolicy func() gax.Retryer) Option

WithMigrationStrategy selects the migration strategy and how the migration retries.

Without it the strategy is AppendNewColumns{CreateIfMissing: true} and the retries are DefaultRetryPolicy's: keeping the table in step with the declaration is what bqsink is for, adding a column is not destructive, and two replicas deploying at once add the same column at once, which BigQuery reports as a failed precondition that only a retry gets past.

Pass MigrationNone{} to leave an existing table alone, or SyncAllColumns{} to also drop columns the declaration no longer has.

retryPolicy says what to do about a failure a later attempt could get past, such as that failed precondition. A nil retryPolicy means the migration attempts the change once; pass DefaultRetryPolicy to keep the retries bqsink would otherwise use. It is called once per migration, because a gax.Retryer carries the state of its backoff and cannot be reused, and returning a bare gax.OnErrorFunc places no limit on the number of attempts.

Writing is not retried from here: that is the write strategy's own business, since it is the one holding the rows while they are in flight.

If s implements Validator, its Validate method decides whether the settings are usable and New fails when they are not.

type Relation

type Relation struct {
	ProjectID string
	DatasetID string
	TableID   string
}

Relation identifies the BigQuery table a Sinker writes to.

ProjectID may be left empty, in which case New fills it in from the project of the bigquery.Client it is given.

func ParseRelation

func ParseRelation(s string) (Relation, error)

ParseRelation parses a table reference written in standard SQL notation.

Both "project.dataset.table" and "dataset.table" are accepted; the latter leaves ProjectID empty. Splitting on "." is unambiguous because BigQuery allows only letters, digits and underscores in dataset and table names, and project IDs contain no dots either.

func (Relation) String

func (r Relation) String() string

String returns the relation in standard SQL notation, omitting the project when ProjectID is empty.

type Row

type Row struct {
	// ID identifies the row. It is what a transport names the row by when it has
	// something to say about it, and it is the value the _ingestion_row_id column
	// gets when the row type fills that column in.
	//
	// It plays no part in reporting what could not be written: the count a
	// WriteResult returns is what says that.
	ID string

	// Values are the columns to write.
	Values map[string]bigquery.Value
}

Row is one row on its way to BigQuery.

type RowFiller

type RowFiller interface {
	FillRow(ctx context.Context, info AppendInfo) error
}

RowFiller lets a row type fill in values just before it is written, which is how columns such as a write timestamp or a row id get theirs.

FillRow is called once per row, on a copy, before the conversion and before any retry. Two things follow from that. The value the caller passed to Sink is left untouched, unless the row was handed over as a pointer, in which case filling reaches the caller's own value. And a retried row carries the values it was first given, so RowID can be used to deduplicate.

It needs a pointer receiver: with a value receiver it would write into a copy that is then discarded, and the first Sink rejects that rather than letting the columns stay empty.

Embedding is the point, since a promoted method makes the outer row satisfy the interface. IngestionMetadata is a ready-made set of columns; a type of your own works the same way when the column names or the values need to differ.

type AccessLog struct {
	bqsink.IngestionMetadata
	UserID string `bqsink:"user_id"`
}

type RowsWriter

type RowsWriter interface {
	// Relation names the table this writer writes to.
	//
	// It is the writer's own answer rather than something it repeats back, so a
	// writer bound to a stream that already exists reports the table that stream
	// belongs to.
	Relation() Relation

	// Client returns the BigQuery client to reconcile the table through, or nil
	// when the writer is not connected to BigQuery at all.
	//
	// A nil client leaves the real table unreadable, so NewSinker refuses every
	// migration strategy but MigrationNone.
	Client() *bigquery.Client

	// BindSchema hands over the declared schema, once it has been reconciled with
	// the real table and before the first WriteRows. It is called once.
	//
	// A transport that derives something from the schema, such as a proto
	// descriptor, does it here and keeps it for the writer's lifetime.
	BindSchema(ctx context.Context, schema bigquery.Schema) error

	// WriteRows hands rows over and returns what will say whether they landed.
	//
	// A non-nil error means the rows were not taken at all. Otherwise they are the
	// writer's, and a writer that buffers may keep them past this call.
	WriteRows(ctx context.Context, rows []Row) (WriteResult, error)

	// Close releases what the writer holds, having settled the rows it still has.
	// Rows a WriteResult was promised for but which were never sent are sent here,
	// so closing is the last chance to decide their fate.
	Close(ctx context.Context) error
}

RowsWriter sends rows to a single BigQuery table.

A writer holds everything writing depends on: the table it writes to, the connection it goes through, how a transient failure is retried, and the rows a buffering transport has not sent yet. A Sinker holds what none of that changes — the declaration, reconciling the table with it, and turning rows into columns — so that the two are settled apart from one another.

An implementation must be safe for concurrent use.

type SchemaChange

type SchemaChange struct {
	// CreateTable asks for the table to be created. The other fields are ignored
	// when it is set, since a created table already matches the declaration.
	CreateTable bool

	// AddColumns holds columns to add. BigQuery only accepts NULLABLE and REPEATED
	// columns on an existing table.
	AddColumns bigquery.Schema

	// RelaxColumns names columns to turn from REQUIRED into NULLABLE.
	RelaxColumns []string

	// DropColumns names columns to drop with an ALTER TABLE statement, which
	// destroys the data they hold irreversibly. They are dropped after AddColumns
	// and RelaxColumns have been applied, so that a failure in between leaves the
	// table holding more than the declaration asks for rather than less.
	DropColumns []string
}

SchemaChange is what a MigrationStrategy asks bqsink to do.

func (SchemaChange) Empty

func (c SchemaChange) Empty() bool

Empty reports whether there is nothing to do.

type SchemaConflict

type SchemaConflict struct {
	// Name is the column's name.
	Name string

	// Reason says what kind of difference this is.
	Reason ConflictReason

	// Want is the column as declared.
	Want *bigquery.FieldSchema

	// Got is the column as the table has it.
	Got *bigquery.FieldSchema
}

SchemaConflict describes a difference BigQuery cannot reconcile by patching the table's metadata, or one bqsink does not implement.

func (SchemaConflict) String

func (c SchemaConflict) String() string

String implements fmt.Stringer.

type SchemaDiff

type SchemaDiff struct {
	// Added holds columns the declaration has and the table lacks.
	Added bigquery.Schema

	// Removed names columns the table has and the declaration lacks.
	Removed []string

	// Relaxed names columns the table marks REQUIRED where the declaration says
	// NULLABLE. BigQuery allows this direction.
	Relaxed []string

	// Conflicts holds differences that cannot be reconciled.
	Conflicts []SchemaConflict
}

SchemaDiff lists how a declared schema differs from a real table's.

func DiffSchema

func DiffSchema(want, got bigquery.Schema) SchemaDiff

DiffSchema compares want, the declared schema, with got, the schema the table currently has.

Adding and dropping columns is only resolved at the top level. A change inside a RECORD lands in Conflicts rather than being migrated.

func (SchemaDiff) Empty

func (d SchemaDiff) Empty() bool

Empty reports whether the declared schema and the table already agree.

func (SchemaDiff) Without

func (d SchemaDiff) Without(names []string) SchemaDiff

Without returns d with every mention of the named columns dropped, letting a strategy leave columns it does not manage alone.

type Sinker

type Sinker struct {
	// contains filtered or unexported fields
}

Sinker writes rows of one declared type to one BigQuery table.

It holds what does not depend on how the rows travel: the declaration, bringing the real table in line with it, and turning a row into columns. Where the rows go and how they get there belongs to the RowsWriter it was built with, which is also the thing to close when there is nothing left to write.

A Sinker is safe for concurrent use.

func NewSinker

func NewSinker(w RowsWriter, decl Declaration, opts ...Option) (*Sinker, error)

NewSinker returns a Sinker writing what decl declares through w.

decl is evaluated once, when it is built with DeclarationOf, so what NewSinker checks about it is only that it built cleanly: decl.err, if DeclarationOf could not make sense of the row type, comes back here rather than on the first write. Nothing contacts BigQuery either: the real table is read once the first batch arrives, since a Sinker with no rows to write has nothing to reconcile.

Without options the migration strategy is AppendNewColumns{CreateIfMissing: true}, which creates the table if it is absent and adds the columns the declaration gained, and its retries are DefaultRetryPolicy's. Pass MigrationNone{} to leave the table alone.

The table is reconciled through the client w reports. A writer with no client cannot be reconciled at all, so only MigrationNone is allowed with one, and it is then said in the log that nothing was checked.

Where w has something to log, being LoggerBindable, it is handed the logger WithLogger settled on. Closing w is the caller's own business: it made it, and a Sinker buffers nothing that would need flushing first.

func (*Sinker) Sink

func (s *Sinker) Sink(ctx context.Context, rows any) (int, error)

Sink writes rows and waits for the writer's WriteResult to settle, returning how many of them that settlement counts as done. What counts as done is the writer's own promise, documented on WriteResult: how many reached BigQuery for a writer promising delivery, and only how many reached its own buffer for one promising acceptance instead, such as a LoadJobsWriter with LoadJobs.FlushRows set — what becomes of those once a job carries them is for FlushRows or Close to report, not Sink.

A slice is a batch of its elements and anything else is a single row, so a []AccessLog and one AccessLog go through the same call. An empty or nil slice writes nothing and reports no error.

Every row has to be of the type the declaration named, which a slice of a concrete type gives for free and a []any can break. A nil row is refused.

The first call is what reconciles the real table with the declaration and hands the writer the settled schema. Its outcome is kept: a failure there is returned by every later call as well, so recovering from one means building a new Sinker.

The rows travel as one batch, so how many are given here is what decides how much a load job carries or an append sends. Nothing is buffered here between calls, though the writer may hold rows back when it was asked to.

Sink returns a non-nil error whenever n is fewer than the rows it was given, so that rows[n:] are exactly the ones the writer did not settle as done. The caller still holds them, which is what makes dealing with them the caller's choice; nothing else records what was lost. n counts rows the writer settled and not rows prepared: a row that cannot be converted leaves n at 0 even though the rows before it were converted.

If the row type implements RowFiller, FillRow is called on a copy of each element first, so that the row can fill in columns such as a write timestamp. That happens once per row, before the conversion and before any retry.

A transient failure is retried by the writer, so a row can reach BigQuery more than once. Neither transport bqsink ships deduplicates, so the guarantee is at-least-once; IngestionMetadata's _ingestion_row_id is what makes those duplicates identifiable.

func (*Sinker) SinkAsync

func (s *Sinker) SinkAsync(ctx context.Context, rows any) (WriteResult, error)

SinkAsync writes rows the way Sink does — the same batch reading, type check, first-call reconciliation, per-row FillRow and conversion — but hands the writer's WriteResult back without calling Wait on it, so the caller decides when, or whether, to. That does not make the call itself prompt: a writer may still do the work, and block for it, inside WriteRows rather than behind Wait.

Whether the returned WriteResult is already resolved and what it promises are the writer's own business, documented on WriteResult: delivery for one writer, only acceptance into a buffer for another. SinkAsync itself promises no more than having handed the rows to the writer; a non-nil error here means they were not even that, and the returned WriteResult is nil. On a nil error the WriteResult is never nil.

Sink's own guard against a writer under-reporting is Sink's alone: SinkAsync does not know how many rows a caller is about to hand the returned WriteResult's Wait to compare against. A caller that wants the same guard gets it by comparing the n Wait returns against how many rows it gave SinkAsync itself.

func (*Sinker) Start

func (s *Sinker) Start(ctx context.Context) error

Start reconciles the real table with the declaration and binds the writer's schema, the same work a first Sink or SinkAsync call would otherwise trigger.

Calling it explicitly lets that work run on a ctx scoped to process startup rather than to whichever request happens to call Sink first, since that ctx outlives migrate's retries and the connection a writer such as StorageWriter keeps open afterward (see start). A later Sink or SinkAsync still triggers this work if Start was never called; calling Start again after it already ran, whether it succeeded or failed, returns the same cached outcome without contacting BigQuery.

The work runs at most once, so it runs on whichever ctx got there first: call Start and wait for it to return before letting Sink or SinkAsync run concurrently, or a request that outraces Start binds the work to that request's own ctx instead.

type Stager

type Stager interface {
	// Stage writes rows and returns the URI a load job should read. The returned
	// cleanup, when not nil, removes what Stage created and is called once the
	// load job has finished, whether or not it succeeded.
	Stage(ctx context.Context, rows []byte) (uri string, cleanup func(context.Context) error, err error)
}

Stager puts the rows somewhere a load job can read them, instead of uploading them with the job itself.

Staging through Cloud Storage suits large batches: the upload is a plain object write that can be retried on its own, and the load job then reads from a URI rather than carrying the data.

type StorageWrite

type StorageWrite struct {
	// StreamType selects the kind of stream to create. The zero value means
	// managedwriter.DefaultStream, which appends immediately and at least once.
	//
	// Only DefaultStream and CommittedStream are supported. PendingStream needs
	// its rows committed in a batch and BufferedStream needs its offset flushed,
	// neither of which bqsink does, so rows written to them would never become
	// visible.
	StreamType managedwriter.StreamType

	// StreamName writes to a stream that already exists instead of creating one.
	// StreamType is ignored when it is set.
	StreamName string

	// DisableWriteRetries stops bqsink from turning the client library's automatic
	// write retries on. The zero value leaves them on, since a row that never lands
	// is what bqsink is for; turn them off where an append being sent twice matters
	// more than it arriving, which is the exactly-once pattern the client library
	// warns they complicate.
	//
	// It says what bqsink does rather than what the stream ends up doing: the client
	// library has no way to switch retries back off, so an EnableWriteRetries(true)
	// of your own in WriterOptions still stands.
	DisableWriteRetries bool

	// ClientOptions configure the managedwriter client.
	ClientOptions []option.ClientOption

	// WriterOptions configure the stream. bqsink applies these first and then sets
	// the destination table and the schema descriptor itself, so an option naming
	// either of those has no effect: the relation and the declared schema are the
	// source of truth.
	WriterOptions []managedwriter.WriterOption
}

StorageWrite writes rows through the BigQuery Storage Write API. It is the settings a StorageWriter is made from.

What a WriteResult reports is particular to the writer that returned it. StorageWrite's WriteRows sends the rows as one append and returns before BigQuery has acknowledged them, so the WriteResult it hands back is genuinely pending: there is no earlier point, such as the rows merely being accepted, for its Wait to report instead. An append is all or nothing: BigQuery appends none of the rows in a request it rejects.

BigQuery caps how large an append request may be, and StorageWrite does not split one, so a batch past that cap is rejected rather than divided. Where batches are large enough for that to be a worry, LoadJobs is the transport for them.

Retrying is left to the client library's own automatic retries, which suit at-least-once delivery. The number of attempts is therefore the library's to decide, so StorageWrite has no policy to set the way LoadJobs does; what it has is DisableWriteRetries, for turning them off.

func (*StorageWrite) NewWriter

func (w *StorageWrite) NewWriter(client *bigquery.Client, relation Relation) (*StorageWriter, error)

NewWriter returns a writer that appends rows to the table relation names.

It does not contact BigQuery: opening the managedwriter client and the stream happens in BindSchema, once the migration has settled the schema. An empty ProjectID on relation is filled in from the client.

When StreamName is set, the stream already belongs to a table, so NewWriter checks relation against it rather than letting relation be silently ignored: it fails if the stream's table and relation name different tables.

func (*StorageWrite) Validate

func (w *StorageWrite) Validate() error

Validate implements Validator.

type StorageWriter

type StorageWriter struct {
	// contains filtered or unexported fields
}

StorageWriter writes rows to one BigQuery table through the Storage Write API. It implements RowsWriter.

BindSchema opens the managedwriter client and stream; nothing is sent before that. WriteRows sends an append and returns without waiting for BigQuery to accept it, so a WriteResult nobody calls Wait on before Close never reports whether its rows landed: StorageWriter buffers no rows of its own to report them on Close's behalf the way LoadJobsWriter does.

func (*StorageWriter) BindLogger

func (w *StorageWriter) BindLogger(logger *slog.Logger)

BindLogger implements LoggerBindable.

func (*StorageWriter) BindSchema

func (w *StorageWriter) BindSchema(ctx context.Context, schema bigquery.Schema) error

BindSchema implements RowsWriter.

It derives the proto descriptor from the schema once, which is possible because the migration has already settled it, and opens a managedwriter client and stream.

func (*StorageWriter) Client

func (w *StorageWriter) Client() *bigquery.Client

Client implements RowsWriter.

func (*StorageWriter) Close

func (w *StorageWriter) Close(ctx context.Context) error

Close implements RowsWriter. It closes both the stream and the client, so that neither leaks its gRPC connection.

Only the first failure is returned, so the other is logged: nothing else would say that a connection did not shut down cleanly. Close does not report the outcome of a WriteResult nobody waited on, unlike LoadJobsWriter's: StorageWriter holds no rows of its own to report them with, so that outcome is simply lost.

func (*StorageWriter) Relation

func (w *StorageWriter) Relation() Relation

Relation implements RowsWriter.

func (*StorageWriter) WriteRows

func (w *StorageWriter) WriteRows(ctx context.Context, rows []Row) (WriteResult, error)

WriteRows implements RowsWriter. It marshals the rows and sends them as one append, returning a WriteResult whose Wait reports BigQuery's real ack of the append: the rows' one and only delivery result, since StorageWrite has no earlier point to call them accepted.

An append is all or nothing, so Wait reports len(rows) or 0: BigQuery appends none of the rows in a request it rejects.

type SyncAllColumns

type SyncAllColumns struct {
	// IgnoreColumns names columns bqsink does not manage. They are neither
	// dropped nor reported as conflicts, which suits columns another system owns.
	IgnoreColumns []string

	// CreateIfMissing creates the table when it does not exist.
	CreateIfMissing bool
}

SyncAllColumns does what AppendNewColumns does and additionally drops columns the table has and the declaration lacks.

func (SyncAllColumns) Plan

func (s SyncAllColumns) Plan(state TableState, logger *slog.Logger) (SchemaChange, error)

Plan implements MigrationStrategy.

func (SyncAllColumns) Validate

func (s SyncAllColumns) Validate() error

Validate implements Validator.

type TableDefiner

type TableDefiner interface {
	BigQueryTableMetadata() *bigquery.TableMetadata
}

TableDefiner lets a row type declare table level settings such as partitioning, clustering, labels and expiration.

The returned metadata's Schema field, if set, overrides the schema derived from struct tags. That is where a column struct tags cannot describe belongs, such as BIGNUMERIC precision, a column description or a policy tag: bqsink has no Option for declaring a schema, since the row type is meant to be the one place that says what the table holds.

Read-only fields (ETag, CreationTime, LastModifiedTime, NumBytes, NumRows, FullID, Type) are ignored.

type TableMeta

type TableMeta struct{}

TableMeta lets a row type settle what describes the table as a whole, through tags on the embedded field, so that a description or a set of labels needs no BigQueryTableMetadata method.

It contributes no column. Embed it as a direct field of the row type:

type AccessLog struct {
	bqsink.TableMeta `description:"one row per request" labels:"team=data,env=prod"`

	Timestamp time.Time `bqsink:"timestamp,required" partition:"day"`
	UserID    string    `bqsink:"user_id"`
}

Only DescriptionTagKey and LabelsTagKey are read here. Anything describing a column, including the column tag itself, is rejected rather than ignored, since TableMeta has no column to apply it to. Declaring the same thing here and in BigQueryTableMetadata is an error rather than one silently winning.

type TableState

type TableState struct {
	// Exists reports whether the table exists.
	Exists bool

	// Diff is how the declared schema differs from the table's. It is zero when
	// Exists is false.
	Diff SchemaDiff
}

TableState describes the destination table as the migration found it.

type Validator

type Validator interface {
	Validate() error
}

Validator lets a strategy check its own settings. When a strategy implements it, the Option carrying that strategy calls Validate and New fails if it returns an error, so a misconfigured strategy is caught before any row is written rather than on the first Sink.

type WriteResult

type WriteResult interface {
	// Wait returns what the WriteResult above promises. A result already
	// resolved returns it at once; a writer that defers delivery, such as
	// StorageWriter, has Wait do the waiting.
	//
	// Cancelling ctx before a deferred result resolves reports that
	// cancellation rather than the rows' actual fate, and does not take the rows
	// back: they may still land.
	Wait(ctx context.Context) (n int, err error)
}

WriteResult says what the writer that returned it is promising about the rows of that call, and the promise is not the same for every writer.

A writer that sends rows itself and waits for BigQuery to accept them, such as StorageWriter, promises delivery: its WriteResult is not resolved until Wait is called, and n then says how many of the rows landed.

A writer that instead buffers rows of its own, such as LoadJobsWriter with LoadJobs.FlushRows set, promises acceptance: its WriteResult is already resolved by the time WriteRows returns, and n says how many rows were taken into the buffer, not how many have landed. What becomes of them once a job carries them is FlushRows's or Close's to report, not this WriteResult's.

A LoadJobsWriter with FlushRows unset submits a job for every call's rows on the spot, so its WriteResult promises delivery too, already resolved with that job's outcome by the time WriteRows returns.

Whichever it promises, a WriteResult that reports fewer rows than were placed in its care always comes with a non-nil error, whether those rows were handed to WriteRows or held in a buffer FlushRows submits.

func ResolvedResult

func ResolvedResult(n int, err error) WriteResult

ResolvedResult returns a WriteResult that has already settled, which is what a transport writing its rows before it returns hands back.