db

Documentation

Index

• Variables
• func OnModuleHashMismatchNames() []string
• type ColumnInfo
  • func NewColumnInfo(name string, databaseTypeName string, scanType any) *ColumnInfo
• type CursorError
• type DSN
• type Loader
  • func NewLoader(clickhouseDSN string, flushInterval time.Duration, ...) (*Loader, error)
  • func (l *Loader) Delete(tableName string, primaryKey string) error
  • func (l *Loader) DeleteAllCursors(ctx context.Context) (err error)
  • func (l *Loader) DeleteCursor(ctx context.Context, moduleHash string) error
  • func (l *Loader) EntriesCount() uint64
  • func (l *Loader) Flush(ctx context.Context, outputModuleHash string, cursor *sink.Cursor) (err error)
  • func (l *Loader) FlushInterval() time.Duration
  • func (l *Loader) GetAllCursors(ctx context.Context) (out map[string]*sink.Cursor, err error)
  • func (l *Loader) GetAvailableTablesInSchema() string
  • func (l *Loader) GetCursor(ctx context.Context, outputModuleHash string) (cursor *sink.Cursor, mistmatchDetected bool, err error)
  • func (l *Loader) GetDatabase() string
  • func (l *Loader) GetIdentifier() string
  • func (l *Loader) HasTable(tableName string) bool
  • func (l *Loader) Insert(tableName string, primaryKey string, data map[string]string) error
  • func (l *Loader) InsertCursor(ctx context.Context, moduleHash string, c *sink.Cursor) error
  • func (l *Loader) LoadTables() error
  • func (l *Loader) MarshalLogObject(encoder zapcore.ObjectEncoder) error
  • func (l *Loader) Setup(ctx context.Context, schemaFile string) error
  • func (l *Loader) Update(tableName string, primaryKey string, data map[string]string) error
  • func (l *Loader) UpdateCursor(ctx context.Context, tx *sql.Tx, moduleHash string, c *sink.Cursor) error
• type OnModuleHashMismatch
  • func ParseOnModuleHashMismatch(name string) (OnModuleHashMismatch, error)
  • func (x OnModuleHashMismatch) IsValid() bool
  • func (x OnModuleHashMismatch) MarshalText() ([]byte, error)
  • func (x OnModuleHashMismatch) String() string
  • func (x *OnModuleHashMismatch) UnmarshalText(text []byte) error
• type Operation
  • func (o *Operation) String() string
• type OperationType
• type Queryable
• type TableInfo
  • func NewTableInfo(name, primaryKeyColumnName string, columnsByName map[string]*ColumnInfo) (*TableInfo, error)
• type TypeGetter

Variables

var ErrCursorNotFound = errors.New("cursor not found")

var ErrInvalidOnModuleHashMismatch = fmt.Errorf("not a valid OnModuleHashMismatch, try [%s]", strings.Join(_OnModuleHashMismatchNames, ", "))

Functions

func OnModuleHashMismatchNames

func OnModuleHashMismatchNames() []string

OnModuleHashMismatchNames returns a list of possible string values of OnModuleHashMismatch.

Types

type ColumnInfo

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

func NewColumnInfo

func NewColumnInfo(name string, databaseTypeName string, scanType any) *ColumnInfo

type CursorError

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

type DSN

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

type Loader

type Loader struct {
	*sql.DB
	// contains filtered or unexported fields
}

func NewLoader

func NewLoader(
	clickhouseDSN string,
	flushInterval time.Duration,
	moduleMismatchMode OnModuleHashMismatch,
	logger *zap.Logger,
	tracer logging.Tracer,
) (*Loader, error)

func (*Loader) Delete

func (l *Loader) Delete(tableName string, primaryKey string) error

Delete a row in the DB, it is assumed the table exists, you can do a check before with HasTable()

func (*Loader) DeleteAllCursors

func (l *Loader) DeleteAllCursors(ctx context.Context) (err error)

DeleteAllCursors deletes the active cursor for the given 'moduleHash'. If no cursor is active and no delete occurrred, returns ErrCursorNotFound. If the delete was not successful on the database, returns an error.

func (*Loader) DeleteCursor

func (l *Loader) DeleteCursor(ctx context.Context, moduleHash string) error

DeleteCursor deletes the active cursor for the given 'moduleHash'. If no cursor is active and no delete occurrred, returns ErrCursorNotFound. If the delete was not successful on the database, returns an error.

func (*Loader) EntriesCount

func (l *Loader) EntriesCount() uint64

func (*Loader) Flush

func (l *Loader) Flush(ctx context.Context, outputModuleHash string, cursor *sink.Cursor) (err error)

func (*Loader) FlushInterval

func (l *Loader) FlushInterval() time.Duration

func (*Loader) GetAllCursors

func (l *Loader) GetAllCursors(ctx context.Context) (out map[string]*sink.Cursor, err error)

GetAllCursors returns an unordered map given for each module's hash recorded the active cursor for it.

func (*Loader) GetAvailableTablesInSchema

func (l *Loader) GetAvailableTablesInSchema() string

func (*Loader) GetCursor

func (l *Loader) GetCursor(ctx context.Context, outputModuleHash string) (cursor *sink.Cursor, mistmatchDetected bool, err error)

func (*Loader) GetDatabase

func (l *Loader) GetDatabase() string

func (*Loader) GetIdentifier

func (l *Loader) GetIdentifier() string

GetIdentifier returns <database>/<schema> suitable for user presentation

func (*Loader) HasTable

func (l *Loader) HasTable(tableName string) bool

func (*Loader) Insert

func (l *Loader) Insert(tableName string, primaryKey string, data map[string]string) error

Insert a row in the DB, it is assumed the table exists, you can do a check before with HasTable()

func (*Loader) InsertCursor

func (l *Loader) InsertCursor(ctx context.Context, moduleHash string, c *sink.Cursor) error

func (*Loader) LoadTables

func (l *Loader) LoadTables() error

func (*Loader) MarshalLogObject

func (l *Loader) MarshalLogObject(encoder zapcore.ObjectEncoder) error

func (*Loader) Setup

func (l *Loader) Setup(ctx context.Context, schemaFile string) error

func (*Loader) Update

func (l *Loader) Update(tableName string, primaryKey string, data map[string]string) error

Update a row in the DB, it is assumed the table exists, you can do a check before with HasTable()

func (*Loader) UpdateCursor

func (l *Loader) UpdateCursor(ctx context.Context, tx *sql.Tx, moduleHash string, c *sink.Cursor) error

UpdateCursor updates the active cursor. If no cursor is active and no update occurred, returns ErrCursorNotFound. If the update was not successful on the database, returns an error.

type OnModuleHashMismatch

type OnModuleHashMismatch uint

ENUM(

 Ignore
	Warn
	Error

)

const (
	// OnModuleHashMismatchIgnore is a OnModuleHashMismatch of type Ignore.
	OnModuleHashMismatchIgnore OnModuleHashMismatch = iota
	// OnModuleHashMismatchWarn is a OnModuleHashMismatch of type Warn.
	OnModuleHashMismatchWarn
	// OnModuleHashMismatchError is a OnModuleHashMismatch of type Error.
	OnModuleHashMismatchError
)

func ParseOnModuleHashMismatch

func ParseOnModuleHashMismatch(name string) (OnModuleHashMismatch, error)

ParseOnModuleHashMismatch attempts to convert a string to a OnModuleHashMismatch.

func (OnModuleHashMismatch) IsValid

func (x OnModuleHashMismatch) IsValid() bool

IsValid provides a quick way to determine if the typed value is part of the allowed enumerated values

func (OnModuleHashMismatch) MarshalText

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

MarshalText implements the text marshaller method.

func (OnModuleHashMismatch) String

func (x OnModuleHashMismatch) String() string

String implements the Stringer interface.

func (*OnModuleHashMismatch) UnmarshalText

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

UnmarshalText implements the text unmarshaller method.

type Operation

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

func (*Operation) String

func (o *Operation) String() string

type OperationType

type OperationType string

const (
	OperationTypeInsert OperationType = "INSERT"
	OperationTypeUpdate OperationType = "UPDATE"
	OperationTypeDelete OperationType = "DELETE"
)

type Queryable

type Queryable interface {
	// contains filtered or unexported methods
}

type TableInfo

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

func NewTableInfo

func NewTableInfo(name, primaryKeyColumnName string, columnsByName map[string]*ColumnInfo) (*TableInfo, error)

type TypeGetter

type TypeGetter func(tableName string, columnName string) (reflect.Type, error)