export

Documentation

Overview

Package export implements mechanisms to subsets of data, from one database to another. Note that it only supports auto-incrementing (single) primary/foreign keys, which must fit in int64.

Index

• Constants
• Variables
• type Dialect
• type Exporter
  • func (x *Exporter) Export(ctx context.Context) error
• type InsertRows
• type JoinRef
• type Mapper
• type Reader
• type ReaderImpl
  • func (x *ReaderImpl[C, R]) Close() error
  • func (x *ReaderImpl[C, R]) QueryContext(ctx context.Context, query string, args ...any) (Rows, error)
• type Result
• type Row
• type RowTransformer
• type Rows
• type Schema
  • func (x *Schema) ColumnIndexes(table Table, columns []string) (primaryKey int, foreignKeys map[string]int, ok bool)
• type SelectBatch
• type SelectRows
• type SimpleMapper
  • func (x SimpleMapper) Load(_ context.Context, table Table, src int64) (dst int64, ok bool, _ error)
  • func (x SimpleMapper) Store(_ context.Context, table Table, src, dst int64) error
• type Snippet
• type Table
  • func (x Table) MarshalJSON() ([]byte, error)
  • func (x Table) MarshalText() ([]byte, error)
  • func (x Table) String() string
  • func (x *Table) UnmarshalJSON(b []byte) error
  • func (x *Table) UnmarshalText(text []byte) error
• type Target
• type Template
  • func (x *Template) Schema() (*Schema, error)
• type UnimplementedDialect
  • func (UnimplementedDialect) InsertRows(*InsertRows) (*Snippet, error)
  • func (UnimplementedDialect) SelectBatch(*SelectBatch) (*Snippet, error)
  • func (UnimplementedDialect) SelectRows(*SelectRows) (*Snippet, error)
• type Writer
• type WriterImpl
  • func (x *WriterImpl[C, R]) Close() error
  • func (x *WriterImpl[C, R]) ExecContext(ctx context.Context, query string, args ...any) (Result, error)

Constants

const (
	DefaultBatchSize = 1000
)

Variables

var (
	ErrUnimplemented = errors.New(`go-sql/export: unimplemented`)
)

Types

type Dialect

type Dialect interface {
	SelectBatch(args *SelectBatch) (*Snippet, error)
	SelectRows(args *SelectRows) (*Snippet, error)
	InsertRows(args *InsertRows) (*Snippet, error)
	// contains filtered or unexported methods
}

Dialect models an SQL dialect. Note that all methods should return (nil, nil) or (nil, ErrUnimplemented) if args is nil. See also UnimplementedDialect.

type Exporter

type Exporter struct {
	Reader Reader
	Writer Writer
	// Mapper maps old ids to new ids.
	Mapper Mapper
	// RowTransformer may be provided as a hook to modify rows before they are inserted.
	RowTransformer RowTransformer
	Schema         *Schema
	Logger         *logiface.Logger[logiface.Event]
	// Offset provides an initial offset, to start querying from.
	Offset map[string]int64
	// Filters further restrict the target data set.
	Filters []*Snippet
	// BatchSize configures the max limit, and defaults to DefaultBatchSize if 0.
	BatchSize int
	// MaxSelectIn configures the maximum number of IDs to "SELECT ... WHERE <id> in (...<values>)".
	// If zero it defaults to the (resolved) batch size.
	MaxSelectIn int
	// MaxOffsetConditions configures the maximum number of offsets columns to support.
	// Default is unlimited.
	MaxOffsetConditions int
}

func (*Exporter) Export

func (x *Exporter) Export(ctx context.Context) error

type InsertRows

type InsertRows struct {
	Schema  *Schema
	Table   Table
	Columns []string
	Values  []any
}

type JoinRef

type JoinRef struct {
	// Alias represents the joined table.
	Alias string

	// Column is the column name, in ONE OF Alias's table OR the parent Target's table, depending on Reverse.
	Column string

	// Reverse is set to true to indicate that Column belongs to the parent Target, rather than Alias.
	Reverse bool
}

JoinRef models part of an expression like "JOIN table b ON a.b_id = b._id" OR "JOIN table a ON a.b_id = b._id".

type Mapper

type Mapper interface {
	Load(ctx context.Context, table Table, src int64) (dst int64, ok bool, err error)
	Store(ctx context.Context, table Table, src, dst int64) error
}

type Reader

type Reader interface {
	Dialect
	// contains filtered or unexported methods
}

type ReaderImpl

type ReaderImpl[C databaseReader[R], R Rows] struct {
	Dialect
	DB C
}

func (*ReaderImpl[C, R]) Close

func (x *ReaderImpl[C, R]) Close() error

func (*ReaderImpl[C, R]) QueryContext

func (x *ReaderImpl[C, R]) QueryContext(ctx context.Context, query string, args ...any) (Rows, error)

type Result

type Result interface {
	LastInsertId() (int64, error)
	RowsAffected() (int64, error)
}

type Row

type Row struct {
	Schema     *Schema
	Table      Table
	Columns    []string
	Values     []any
	PrimaryKey int64
}

type RowTransformer

type RowTransformer interface {
	TransformRow(ctx context.Context, row *Row) error
}

type Rows

type Rows interface {
	Close() error
	Columns() ([]string, error)
	Err() error
	Next() bool
	Scan(dest ...any) error
}

type Schema

type Schema struct {
	Template     *Template
	PrimaryKeys  map[Table]string
	ForeignKeys  map[Table]map[string]Table
	Dependencies map[Table][]Table
	// AliasOrder contains each key from Template.Targets, sorted by Target.Order + string sort.
	// Target.Order is in descending order (larger number first), and is higher priority than the string sort.
	// It is to be used control evaluation of the "offset" conditions in the SelectBatch WHERE clause, which, by
	// nature, must align with specific ORDER BY expressions.
	AliasOrder []string
}

Schema defines the structure of the data to be exported, and may be initialised via Template.Schema.

func (*Schema) ColumnIndexes

func (x *Schema) ColumnIndexes(table Table, columns []string) (primaryKey int, foreignKeys map[string]int, ok bool)

type SelectBatch

type SelectBatch struct {
	Schema              *Schema
	Offset              map[string]int64
	Filters             []*Snippet
	Limit               uint64
	MaxOffsetConditions int
}

type SelectRows

type SelectRows struct {
	Schema *Schema
	Table  Table
	IDs    []int64
}

type SimpleMapper

type SimpleMapper map[Table]map[int64]int64

func (SimpleMapper) Load

func (x SimpleMapper) Load(_ context.Context, table Table, src int64) (dst int64, ok bool, _ error)

func (SimpleMapper) Store

func (x SimpleMapper) Store(_ context.Context, table Table, src, dst int64) error

type Snippet

type Snippet struct {
	SQL  string
	Args []any
}

Snippet models a SQL snippet + associated args.

type Table

type Table struct {
	Schema string
	Name   string
}

func (Table) MarshalJSON

func (x Table) MarshalJSON() ([]byte, error)

func (Table) MarshalText

func (x Table) MarshalText() ([]byte, error)

func (Table) String

func (x Table) String() string

func (*Table) UnmarshalJSON

func (x *Table) UnmarshalJSON(b []byte) error

func (*Table) UnmarshalText

func (x *Table) UnmarshalText(text []byte) error

type Target

type Target struct {
	// ForeignKey is optional, and models the join condition for this Target.
	ForeignKey *JoinRef
	Table      Table
	// PrimaryKey is the column of the table's primary key. It's after SELECT and (in some cases) ON.
	PrimaryKey string
	// Order is used to indicate the order of the predicates (expressions in WHERE and ORDER BY).
	// Larger numbers should appear first.
	Order int
}

type Template

type Template struct {
	// Targets are all joined tables with alias (can include the same table more than once).
	Targets map[string]*Target

	// Filters are SQL snippets to AND together in groups, to filter the batch data set.
	Filters []*Snippet
}

Template is the input used to generate a Schema, and is based on a query in the format of "SELECT [<pk> as <alias> ...] {FROM/JOIN} [WHERE] [ORDER BY ...]", where Targets are joined tables.

func (*Template) Schema

func (x *Template) Schema() (*Schema, error)

type UnimplementedDialect

type UnimplementedDialect struct{}

func (UnimplementedDialect) InsertRows

func (UnimplementedDialect) InsertRows(*InsertRows) (*Snippet, error)

func (UnimplementedDialect) SelectBatch

func (UnimplementedDialect) SelectBatch(*SelectBatch) (*Snippet, error)

func (UnimplementedDialect) SelectRows

func (UnimplementedDialect) SelectRows(*SelectRows) (*Snippet, error)

type Writer

type Writer interface {
	Dialect
	// contains filtered or unexported methods
}

type WriterImpl

type WriterImpl[C databaseWriter[R], R Result] struct {
	Dialect
	DB C
}

func (*WriterImpl[C, R]) Close

func (x *WriterImpl[C, R]) Close() error

func (*WriterImpl[C, R]) ExecContext

func (x *WriterImpl[C, R]) ExecContext(ctx context.Context, query string, args ...any) (Result, error)