Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
- API
  
  Reference documentation for the pkg.go.dev API
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

bqsink

package module

Go to main page

Versions in this module

v0

v0.0.2

Aug 21, 2026

Changes in this version

type Sinker

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

v0.0.1

Aug 12, 2026

Changes in this version

+ const ClusterTagKey

+ const DescriptionTagKey

+ const LabelsTagKey

+ const PartitionTagKey

+ const TagKey

+ var ErrSchemaConflict = errors.New("bqsink: schema conflict")

+ var ErrTableMissing = errors.New("bqsink: table does not exist")

+ var ErrWriterClosed = errors.New("the writer is closed")

+ func DefaultRetryPolicy() gax.Retryer

+ type AppendInfo struct

+ Relation Relation

+ RowID string

+ SinkerCreatedAt time.Time

+ SinkerID string

+ Time time.Time

+ type AppendNewColumns struct

+ CreateIfMissing bool

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

+ type ConflictReason int

+ const ConflictNested

+ const ConflictRepeated

+ const ConflictRequired

+ const ConflictType

+ func (r ConflictReason) String() string

+ type Declaration struct

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

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

+ type FieldMarshaler interface

+ BigQueryFieldType func() bigquery.FieldType

+ MarshalBigQueryValue func() (bigquery.Value, error)

+ type Flusher interface

+ FlushRows func(ctx context.Context) (WriteResult, error)

+ type IngestionMetadata struct

+ IngestionAt time.Time

+ IngestionID string

+ IngestionRowID string

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

+ type LoadJobs struct

+ FlushRows int

+ RetryPolicy func() gax.Retryer

+ Staging Stager

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

+ func (w *LoadJobs) Validate() error

+ type LoadJobsWriter struct

+ 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 interface

+ BindLogger func(logger *slog.Logger)

+ type Marshalers struct

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

+ type MigrationNone struct

+ CreateIfMissing bool

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

+ type MigrationStrategy interface

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

+ type Option func(*config) error

+ func WithLogger(logger *slog.Logger) Option

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

+ type Relation struct

+ DatasetID string

+ ProjectID string

+ TableID string

+ func ParseRelation(s string) (Relation, error)

+ func (r Relation) String() string

+ type Row struct

+ ID string

+ Values map[string]bigquery.Value

+ type RowFiller interface

+ FillRow func(ctx context.Context, info AppendInfo) error

+ type RowsWriter interface

+ BindSchema func(ctx context.Context, schema bigquery.Schema) error

+ Client func() *bigquery.Client

+ Close func(ctx context.Context) error

+ Relation func() Relation

+ WriteRows func(ctx context.Context, rows []Row) (WriteResult, error)

+ type SchemaChange struct

+ AddColumns bigquery.Schema

+ CreateTable bool

+ DropColumns []string

+ RelaxColumns []string

+ func (c SchemaChange) Empty() bool

+ type SchemaConflict struct

+ Got *bigquery.FieldSchema

+ Name string

+ Reason ConflictReason

+ Want *bigquery.FieldSchema

+ func (c SchemaConflict) String() string

+ type SchemaDiff struct

+ Added bigquery.Schema

+ Conflicts []SchemaConflict

+ Relaxed []string

+ Removed []string

+ func DiffSchema(want, got bigquery.Schema) SchemaDiff

+ func (d SchemaDiff) Empty() bool

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

+ type Sinker struct

+ 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)

+ type Stager interface

+ Stage func(ctx context.Context, rows []byte) (uri string, cleanup func(context.Context) error, err error)

+ type StorageWrite struct

+ ClientOptions []option.ClientOption

+ DisableWriteRetries bool

+ StreamName string

+ StreamType managedwriter.StreamType

+ WriterOptions []managedwriter.WriterOption

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

+ func (w *StorageWrite) Validate() error

+ type StorageWriter struct

+ 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 struct

+ CreateIfMissing bool

+ IgnoreColumns []string

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

+ func (s SyncAllColumns) Validate() error

+ type TableDefiner interface

+ BigQueryTableMetadata func() *bigquery.TableMetadata

+ type TableMeta struct

+ type TableState struct

+ Diff SchemaDiff

+ Exists bool

+ type Validator interface

+ Validate func() error

+ type WriteResult interface

+ Wait func(ctx context.Context) (n int, err error)

+ func ResolvedResult(n int, err error) WriteResult