fetch

Documentation

Index

• func Fetch(ctx context.Context, cfg Config, logger zerolog.Logger, ...) (retErr error)
• func GetColumnTypes(ctx context.Context, logger zerolog.Logger, conn dbconn.Conn, ...) (columnsWithType, error)
• func GetConstraints(ctx context.Context, logger zerolog.Logger, conn dbconn.Conn, ...) ([]string, error)
• func GetCreateTableStmt(ctx context.Context, logger zerolog.Logger, conns dbconn.OrderedConns, ...) (string, error)
• func GetDropTableStmt(table dbtable.DBTable) (string, error)
• func GetTypeConvMap(customizedTypeMapPath string, logger zerolog.Logger) (typeconv.TableTypeMap, error)
• func IsClearContinuationTokenMode(cfg Config) bool
• func IsImportCopyOnlyMode(cfg Config) bool
• func ListContinuationTokens(ctx context.Context, testOnly bool, targetPgxConn *pgx.Conn, numResults int) (string, error)
• func MaybeCheckMultiRegionLocality(ctx context.Context, logger zerolog.Logger, conns dbconn.OrderedConns, ...) (checked bool, err error)
• type Config
• type CopyResult
  • func Copy(ctx context.Context, baseConn dbconn.Conn, logger zerolog.Logger, ...) (CopyResult, error)
• type PGIface
• type Pipe
  • func NewPipe(b *bytes.Buffer) *Pipe
  • func (p *Pipe) Close() error
  • func (p *Pipe) CloseWithError(err error) error
  • func (p *Pipe) Read(b []byte) (int, error)
  • func (p *Pipe) Write(b []byte) (int, error)

Functions

func Fetch

func Fetch(
	ctx context.Context,
	cfg Config,
	logger zerolog.Logger,
	conns dbconn.OrderedConns,
	blobStore datablobstorage.Store,
	tableFilter utils.FilterConfig,
	testingKnobs testutils.FetchTestingKnobs,
) (retErr error)

func GetColumnTypes

func GetColumnTypes(
	ctx context.Context, logger zerolog.Logger, conn dbconn.Conn, table dbtable.DBTable,
) (columnsWithType, error)

func GetConstraints

func GetConstraints(
	ctx context.Context, logger zerolog.Logger, conn dbconn.Conn, table dbtable.DBTable,
) ([]string, error)

func GetCreateTableStmt

func GetCreateTableStmt(
	ctx context.Context,
	logger zerolog.Logger,
	conns dbconn.OrderedConns,
	table dbtable.DBTable,
	overrideTypeMap typeconv.TableTypeMap,
) (string, error)

func GetDropTableStmt

func GetDropTableStmt(table dbtable.DBTable) (string, error)

func GetTypeConvMap

func GetTypeConvMap(
	customizedTypeMapPath string, logger zerolog.Logger,
) (typeconv.TableTypeMap, error)

func IsClearContinuationTokenMode

func IsClearContinuationTokenMode(cfg Config) bool

IsClearContinuationTokenMode determines if we must clear continuation tokens from the _molt_fetch_exceptions table. This is to ensure that there is only ever one set of active tokens at a time.

func IsImportCopyOnlyMode

func IsImportCopyOnlyMode(cfg Config) bool

func ListContinuationTokens

func ListContinuationTokens(
	ctx context.Context, testOnly bool, targetPgxConn *pgx.Conn, numResults int,
) (string, error)

func MaybeCheckMultiRegionLocality

func MaybeCheckMultiRegionLocality(
	ctx context.Context, logger zerolog.Logger, conns dbconn.OrderedConns, table *dbtable.DBTable,
) (checked bool, err error)

Types

type Config

type Config struct {
	FlushSize            int
	FlushRows            int
	Cleanup              bool
	UseCopy              bool
	TableConcurrency     int
	Shards               int
	FetchID              string
	ContinuationToken    string
	ContinuationFileName string
	// TestOnly means this fetch attempt is just for test, and hence all time/duration
	// stats are deterministic.
	TestOnly bool

	// The target table handling configs.
	Truncate bool

	DropAndRecreateNewSchema bool
	CustomizedTypeMapPath    string

	// NonInteractive relates to if user input should be prompted. If false,
	// user prompting is initiating before certain actions like wiping data.
	// If true, user prompting will be skipped and actions will be confirmed automatically.
	NonInteractive bool

	Compression        compression.Flag
	ExportSettings     dataexport.Settings
	OngoingReplication bool

	// AllowCockroachReplication gates Cockroach source replication using CDC sink.
	// TODO (rluu): remove this gate when feature is mature.
	AllowCockroachReplication bool

	// UsingDockerBackend lets fetch know if a docker backend is used
	// for the database spin up. This will affect URL pathing for
	// localstore and changefeed setup.
	UsingDockerBackend bool

	// IsTableConcurrencySet marks whether a user set the table concurrency flag.
	IsTableConcurrencySet bool
	// IsExportConcurrencySet marks whether a user set the export concurrency flag.
	IsExportConcurrencySet bool

	// ReplicatorFlags are the flags that can be passed into the replicator command.
	ReplicatorFlags string
}

type CopyResult

type CopyResult struct {
	StartTime time.Time
	EndTime   time.Time
}

func Copy

func Copy(
	ctx context.Context,
	baseConn dbconn.Conn,
	logger zerolog.Logger,
	table dbtable.VerifiedTable,
	resources []datablobstorage.Resource,
	isLocal bool,
	isClearContinuationTokenMode bool,
	exceptionLog *status.ExceptionLog,
) (CopyResult, error)

type PGIface

type PGIface interface {
	Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)
}

type Pipe

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

func NewPipe

func NewPipe(b *bytes.Buffer) *Pipe

func (*Pipe) Close

func (p *Pipe) Close() error

func (*Pipe) CloseWithError

func (p *Pipe) CloseWithError(err error) error

func (*Pipe) Read

func (p *Pipe) Read(b []byte) (int, error)

Read from the buffer into b. It blocks if the buffer is empty and waits for the p.c.Signal call from write to wake and check the condition again.

func (*Pipe) Write

func (p *Pipe) Write(b []byte) (int, error)

Write copies bytes from b into the buffer and sends a signal to wake the reader that is in a wait state.