client

Documentation

Index

• Constants
• Variables
• func New(ctx context.Context, logger zerolog.Logger, specBytes []byte, ...) (plugin.Client, error)
• func TestConnection(ctx context.Context, _ zerolog.Logger, specBytes []byte) error
• type Client
  • func (c *Client) Close(ctx context.Context) error
  • func (c *Client) ColumnToBigQuerySchema(col schema.Column) *bigquery.FieldSchema
  • func (c *Client) DataTypeToBigQuerySchema(dt arrow.DataType) bigquery.Schema
  • func (c *Client) DataTypeToBigQueryType(dt arrow.DataType) bigquery.FieldType
  • func (c *Client) MigrateTables(ctx context.Context, msgs message.WriteMigrateTables) error
  • func (c *Client) Read(ctx context.Context, table *schema.Table, res chan<- arrow.Record) error
  • func (c *Client) Write(ctx context.Context, res <-chan message.WriteMessage) error
  • func (c *Client) WriteTableBatch(ctx context.Context, name string, msgs message.WriteInserts) error
• type Spec
  • func (Spec) JSONSchemaExtend(sc *jsonschema.Schema)
  • func (s *Spec) SetDefaults()
  • func (s *Spec) Validate() error
• type TimePartitioningOption
  • func (TimePartitioningOption) JSONSchemaExtend(sc *jsonschema.Schema)
  • func (t TimePartitioningOption) Validate() error

Constants

const (
	TimePartitioningOptionNone = TimePartitioningOption("none")
	TimePartitioningOptionHour = TimePartitioningOption("hour")
	TimePartitioningOptionDay  = TimePartitioningOption("day")
)

Variables

var JSONSchema string

var TimePartitioningOptions = []TimePartitioningOption{
	TimePartitioningOptionNone,
	TimePartitioningOptionHour,
	TimePartitioningOptionDay,
}

Functions

func New

func New(ctx context.Context, logger zerolog.Logger, specBytes []byte, opts plugin.NewClientOptions) (plugin.Client, error)

func TestConnection

func TestConnection(ctx context.Context, _ zerolog.Logger, specBytes []byte) error

Types

type Client

type Client struct {
	plugin.UnimplementedSource

	batchwriter.UnimplementedDeleteStale
	batchwriter.UnimplementedDeleteRecord
	// contains filtered or unexported fields
}

func (*Client) Close

func (c *Client) Close(ctx context.Context) error

func (*Client) ColumnToBigQuerySchema

func (c *Client) ColumnToBigQuerySchema(col schema.Column) *bigquery.FieldSchema

func (*Client) DataTypeToBigQuerySchema

func (c *Client) DataTypeToBigQuerySchema(dt arrow.DataType) bigquery.Schema

func (*Client) DataTypeToBigQueryType

func (c *Client) DataTypeToBigQueryType(dt arrow.DataType) bigquery.FieldType

func (*Client) MigrateTables

func (c *Client) MigrateTables(ctx context.Context, msgs message.WriteMigrateTables) error

func (*Client) Read

func (c *Client) Read(ctx context.Context, table *schema.Table, res chan<- arrow.Record) error

func (*Client) Write

func (c *Client) Write(ctx context.Context, res <-chan message.WriteMessage) error

func (*Client) WriteTableBatch

func (c *Client) WriteTableBatch(ctx context.Context, name string, msgs message.WriteInserts) error

type Spec

type Spec struct {
	// The id of the project where the destination BigQuery database resides.
	ProjectID string `json:"project_id" jsonschema:"required,minLength=1"`

	//  The name of the BigQuery dataset within the project, e.g. `my_dataset`.
	//  This dataset needs to be created before running a sync or migration.
	DatasetID string `json:"dataset_id" jsonschema:"required,minLength=1"`

	// The data location of the BigQuery dataset. If set, will be used as the default location for job operations.
	// Pro-tip: this can solve "dataset not found" issues for newly created datasets.
	DatasetLocation string `json:"dataset_location"`

	// The time partitioning to use when creating tables. The partition time column used will always be `_cq_sync_time` so that all rows for a sync run will be partitioned on the hour/day the sync started.
	TimePartitioning TimePartitioningOption `json:"time_partitioning"`

	// GCP service account key content.
	// This allows for using different service accounts for the GCP source and BigQuery destination.
	// If using service account keys, it is best to use [environment or file variable substitution](/docs/advanced-topics/environment-variable-substitution).
	ServiceAccountKeyJSON string `json:"service_account_key_json"`

	// The BigQuery API endpoint to use. This is useful for testing against a local emulator.
	Endpoint string `json:"endpoint"`

	// Number of records to write before starting a new object.
	BatchSize int64 `json:"batch_size" jsonschema:"minimum=1,default=10000"`

	// Number of bytes (as Arrow buffer size) to write before starting a new object.
	BatchSizeBytes int64 `json:"batch_size_bytes" jsonschema:"minimum=1,default=5242880"`

	// Maximum interval between batch writes.
	BatchTimeout configtype.Duration `json:"batch_timeout"`
}

func (Spec) JSONSchemaExtend

func (Spec) JSONSchemaExtend(sc *jsonschema.Schema)

func (*Spec) SetDefaults

func (s *Spec) SetDefaults()

func (*Spec) Validate

func (s *Spec) Validate() error

type TimePartitioningOption

type TimePartitioningOption string

func (TimePartitioningOption) JSONSchemaExtend

func (TimePartitioningOption) JSONSchemaExtend(sc *jsonschema.Schema)

func (TimePartitioningOption) Validate

func (t TimePartitioningOption) Validate() error