file_storage

Documentation

Index

• Constants
• func NewGCSBulker(bulkerConfig bulker.Config) (bulker.Bulker, error)
• func NewReplacePartitionStream(id string, p implementations.FileAdapter, tableName string, ...) (bulker.BulkerStream, error)
• func NewReplaceTableStream(id string, p implementations.FileAdapter, tableName string, ...) (bulker.BulkerStream, error)
• func NewS3Bulker(bulkerConfig bulker.Config) (bulker.Bulker, error)
• func NewTransactionalStream(id string, p implementations.FileAdapter, tableName string, ...) (bulker.BulkerStream, error)
• type AbstractFileStorageStream
  • func (ps *AbstractFileStorageStream) Abort(ctx context.Context) (state bulker.State)
  • func (ps *AbstractFileStorageStream) Complete(ctx context.Context) (state bulker.State, err error)
  • func (ps *AbstractFileStorageStream) Consume(ctx context.Context, object types2.Object) (state bulker.State, processedObject types2.Object, err error)
  • func (ps *AbstractFileStorageStream) ConsumeJSON(ctx context.Context, json []byte) (state bulker.State, processedObject types2.Object, err error)
  • func (ps *AbstractFileStorageStream) ConsumeMap(ctx context.Context, mp map[string]any) (state bulker.State, processedObject types2.Object, err error)
• type DeduplicationLine
• type GCSBulker
  • func (gcs *GCSBulker) CreateStream(id, tableName string, mode bulker.BulkMode, ...) (bulker.BulkerStream, error)
• type GCSConfig
• type ReplacePartitionStream
  • func (ps *ReplacePartitionStream) Complete(ctx context.Context) (state bulker.State, err error)
• type ReplaceTableStream
  • func (ps *ReplaceTableStream) Complete(ctx context.Context) (state bulker.State, err error)
• type S3Bulker
  • func (s3 *S3Bulker) CreateStream(id, tableName string, mode bulker.BulkMode, ...) (bulker.BulkerStream, error)
• type TransactionalStream

Constants

const FilenameDate = "2006_01_02T15_04_05"

const GCSAutocommitUnsupported = "Stream mode is not supported for GCS. Please use 'batch' mode"

const S3AutocommitUnsupported = "Stream mode is not supported for GCS. Please use 'batch' mode"

Functions

func NewGCSBulker

func NewGCSBulker(bulkerConfig bulker.Config) (bulker.Bulker, error)

func NewReplacePartitionStream

func NewReplacePartitionStream(id string, p implementations.FileAdapter, tableName string, streamOptions ...bulker.StreamOption) (bulker.BulkerStream, error)

func NewReplaceTableStream

func NewReplaceTableStream(id string, p implementations.FileAdapter, tableName string, streamOptions ...bulker.StreamOption) (bulker.BulkerStream, error)

func NewS3Bulker

func NewS3Bulker(bulkerConfig bulker.Config) (bulker.Bulker, error)

func NewTransactionalStream

func NewTransactionalStream(id string, p implementations.FileAdapter, tableName string, streamOptions ...bulker.StreamOption) (bulker.BulkerStream, error)

Types

type AbstractFileStorageStream

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

func (*AbstractFileStorageStream) Abort

func (ps *AbstractFileStorageStream) Abort(ctx context.Context) (state bulker.State)

func (*AbstractFileStorageStream) Complete

func (ps *AbstractFileStorageStream) Complete(ctx context.Context) (state bulker.State, err error)

func (*AbstractFileStorageStream) Consume

func (ps *AbstractFileStorageStream) Consume(ctx context.Context, object types2.Object) (state bulker.State, processedObject types2.Object, err error)

func (*AbstractFileStorageStream) ConsumeJSON

func (ps *AbstractFileStorageStream) ConsumeJSON(ctx context.Context, json []byte) (state bulker.State, processedObject types2.Object, err error)

func (*AbstractFileStorageStream) ConsumeMap

func (ps *AbstractFileStorageStream) ConsumeMap(ctx context.Context, mp map[string]any) (state bulker.State, processedObject types2.Object, err error)

type DeduplicationLine

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

type GCSBulker

type GCSBulker struct {
	implementations2.GoogleCloudStorage
}

func (*GCSBulker) CreateStream

func (gcs *GCSBulker) CreateStream(id, tableName string, mode bulker.BulkMode, streamOptions ...bulker.StreamOption) (bulker.BulkerStream, error)

type GCSConfig

type GCSConfig struct {
	implementations2.FileConfig `mapstructure:",squash" json:",inline" yaml:",inline"`
	Bucket                      string `mapstructure:"bucket,omitempty" json:"bucket,omitempty" yaml:"bucket,omitempty"`
	AccessKey                   any    `mapstructure:"accessKey,omitempty" json:"accessKey,omitempty" yaml:"accessKey,omitempty"`
}

type ReplacePartitionStream

type ReplacePartitionStream struct {
	AbstractFileStorageStream
}

func (*ReplacePartitionStream) Complete

func (ps *ReplacePartitionStream) Complete(ctx context.Context) (state bulker.State, err error)

type ReplaceTableStream

type ReplaceTableStream struct {
	AbstractFileStorageStream
}

func (*ReplaceTableStream) Complete

func (ps *ReplaceTableStream) Complete(ctx context.Context) (state bulker.State, err error)

type S3Bulker

type S3Bulker struct {
	implementations.S3
}

func (*S3Bulker) CreateStream

func (s3 *S3Bulker) CreateStream(id, tableName string, mode bulker.BulkMode, streamOptions ...bulker.StreamOption) (bulker.BulkerStream, error)

type TransactionalStream

type TransactionalStream struct {
	AbstractFileStorageStream
}