db

package
v1.20.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 24, 2026 License: Apache-2.0 Imports: 35 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrCursorNotFound = errors.New("cursor not found")
View Source 
var ErrInvalidOnModuleHashMismatch = fmt.Errorf("not a valid OnModuleHashMismatch, try [%s]", strings.Join(_OnModuleHashMismatchNames, ", "))
View Source 
var PruneReversibleSegmentDuration = metrics.NewCounter("prune_reversible_segment_duration", "The amount of time spent pruning reversible segment in nanoseconds")
View Source 
var QueryExecutionDuration = metrics.NewCounterVec("tx_query_execution_duration", []string{"query_type"}, "The amount of time spent executing queries by type (normal/undo) in nanoseconds")

Functions

func EscapeIdentifier 

func EscapeIdentifier(valueToEscape string) string

func GenerateCreateTableSQL 

func GenerateCreateTableSQL(tables map[string]*TableInfo) string

func OnModuleHashMismatchNames 

func OnModuleHashMismatchNames() []string

OnModuleHashMismatchNames returns a list of possible string values of OnModuleHashMismatch.

func RegisterMetrics 

func RegisterMetrics()

func TestSinglePrimaryKeyTables 

func TestSinglePrimaryKeyTables(schema string) map[string]*TableInfo

func TestTables 

func TestTables(schema string, customTable map[string]*TableInfo) map[string]*TableInfo

Types

type ClickhouseDialect 

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

func NewClickhouseDialect 

func NewClickhouseDialect(schemaName string, cursorTableName string, cluster string) *ClickhouseDialect

func (ClickhouseDialect) AllowPkDuplicates 

func (d ClickhouseDialect) AllowPkDuplicates() bool

func (ClickhouseDialect) CreateUser 

func (d ClickhouseDialect) CreateUser(tx Tx, ctx context.Context, l *Loader, username string, password string, _database string, readOnly bool) error

func (ClickhouseDialect) DriverSupportRowsAffected 

func (d ClickhouseDialect) DriverSupportRowsAffected() bool

func (ClickhouseDialect) ExecuteSetupScript 

func (d ClickhouseDialect) ExecuteSetupScript(ctx context.Context, l *Loader, schemaSql string) error

func (ClickhouseDialect) Flush 

func (d ClickhouseDialect) Flush(tx Tx, ctx context.Context, l *Loader, outputModuleHash string, lastFinalBlock uint64) (int, error)

Clickhouse should be used to insert a lot of data in batches. The current official clickhouse driver doesn't support Transactions for multiple tables. The only way to add in batches is creating a transaction for a table, adding all rows and commiting it.

func (ClickhouseDialect) GetAllCursorsQuery 

func (d ClickhouseDialect) GetAllCursorsQuery(table string) string

func (ClickhouseDialect) GetCreateCursorQuery 

func (d ClickhouseDialect) GetCreateCursorQuery(schema string, withPostgraphile bool) string

func (ClickhouseDialect) GetCreateHistoryQuery 

func (d ClickhouseDialect) GetCreateHistoryQuery(schema string, withPostgraphile bool) string

func (ClickhouseDialect) GetPrimaryKey 

func (d ClickhouseDialect) GetPrimaryKey(db *sql.DB, schemaName, tableName string) ([]string, error)

func (ClickhouseDialect) GetTableColumns 

func (d ClickhouseDialect) GetTableColumns(db *sql.DB, schemaName, tableName string) ([]*sql.ColumnType, error)

func (ClickhouseDialect) GetTablesInSchema 

func (d ClickhouseDialect) GetTablesInSchema(db *sql.DB, schemaName string) ([][2]string, error)

func (ClickhouseDialect) GetUpdateCursorQuery 

func (d ClickhouseDialect) GetUpdateCursorQuery(table, moduleHash string, cursor *sink.Cursor, block_num uint64, block_id string) string

func (ClickhouseDialect) OnlyInserts 

func (d ClickhouseDialect) OnlyInserts() bool

func (ClickhouseDialect) ParseDatetimeNormalization 

func (d ClickhouseDialect) ParseDatetimeNormalization(value string) string

func (ClickhouseDialect) Revert 

func (d ClickhouseDialect) Revert(tx Tx, ctx context.Context, l *Loader, lastValidFinalBlock uint64) error

type ColumnInfo 

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

func NewColumnInfo 

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

type DSN 

type DSN struct {
	Host     string
	Port     int64
	Username string
	Password string
	Database string
	Options  DSNOptions
	// contains filtered or unexported fields
}

func ParseDSN 

func ParseDSN(dsn string) (*DSN, error)

func (*DSN) Clone 

func (c *DSN) Clone() *DSN

func (*DSN) ConnString 

func (c *DSN) ConnString() string

func (*DSN) Driver 

func (c *DSN) Driver() string

func (*DSN) Schema 

func (c *DSN) Schema() string

type DSNOptions 

type DSNOptions url.Values

DSNOptions is a thin wrapper around url.Values to provide helper methods and better names.

func (DSNOptions) Encode 

func (v DSNOptions) Encode() string

Encode encodes the values into “URL encoded” form ("bar=baz&foo=quux") sorted by key.

func (DSNOptions) EncodeWithSeparator 

func (v DSNOptions) EncodeWithSeparator(sep string) string

EncodeWithSeparator encodes the values into “URL encoded” like form ("bar=baz foo=quux") sorted by key where essentially the separator is used instead of '&'.

func (DSNOptions) Get 

func (v DSNOptions) Get(key string) string

Get returns the value associated with the key.

func (DSNOptions) GetOr 

func (v DSNOptions) GetOr(key, defaultValue string) string

GetOr returns the value associated with the key or defaultValue if not found.

func (DSNOptions) Iter 

func (v DSNOptions) Iter() iter.Seq2[string, string]

Iterate over the first value of each key, to be used in for range loops.

func (DSNOptions) RemoveOr 

func (v DSNOptions) RemoveOr(key, defaultValue string) string

RemoveOr removes the key from the options and returns its value or defaultValue if not found.

type Dialect 

type Dialect interface {
	GetCreateCursorQuery(schema string, withPostgraphile bool) string
	GetCreateHistoryQuery(schema string, withPostgraphile bool) string
	ExecuteSetupScript(ctx context.Context, l *Loader, schemaSql string) error
	DriverSupportRowsAffected() bool
	GetUpdateCursorQuery(table, moduleHash string, cursor *sink.Cursor, block_num uint64, block_id string) string
	GetAllCursorsQuery(table string) string
	ParseDatetimeNormalization(value string) string
	Flush(tx Tx, ctx context.Context, l *Loader, outputModuleHash string, lastFinalBlock uint64) (int, error)
	Revert(tx Tx, ctx context.Context, l *Loader, lastValidFinalBlock uint64) error
	OnlyInserts() bool
	AllowPkDuplicates() bool
	CreateUser(tx Tx, ctx context.Context, l *Loader, username string, password string, database string, readOnly bool) error
	GetTableColumns(db *sql.DB, schemaName, tableName string) ([]*sql.ColumnType, error)
	GetPrimaryKey(db *sql.DB, schemaName, tableName string) ([]string, error)
	GetTablesInSchema(db *sql.DB, schemaName string) ([][2]string, error)
}

type FieldData 

type FieldData struct {
	Value    string
	UpdateOp UpdateOp
}

FieldData holds a field's value and its update operation

type Loader 

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

func NewLoader 

func NewLoader(
	dsn *DSN,
	cursorTableName string, historyTableName string, clickhouseCluster string,
	batchBlockFlushInterval int,
	batchRowFlushInterval int,
	liveBlockFlushInterval int,
	OnModuleHashMismatch string,
	handleReorgs *bool,
	logger *zap.Logger,
	tracer logging.Tracer,
) (*Loader, error)

func NewTestLoader 

func NewTestLoader(
	t *testing.T,
	dsnRaw string,
	testTx *TestTx,
	tables map[string]*TableInfo,
	zlog *zap.Logger,
	tracer logging.Tracer,
) *Loader

func (*Loader) BatchBlockFlushInterval 

func (l *Loader) BatchBlockFlushInterval() int

func (*Loader) Begin 

func (l *Loader) Begin() (Tx, error)

func (*Loader) BeginTx 

func (l *Loader) BeginTx(ctx context.Context, opts *sql.TxOptions) (Tx, error)

func (*Loader) CreateUser 

func (l *Loader) CreateUser(ctx context.Context, username string, password string, database string, readOnly bool) (err error)

func (*Loader) Delete 

func (l *Loader) Delete(tableName string, primaryKey map[string]string, reversibleBlockNum *uint64) 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) (deletedCount int64, 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) Flush 

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

func (*Loader) FlushNeeded 

func (l *Loader) FlushNeeded() bool

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) GetColumnsForTable 

func (l *Loader) GetColumnsForTable(name string) []string

func (*Loader) GetCursor 

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

func (*Loader) GetDSN 

func (l *Loader) GetDSN() *DSN

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

func (*Loader) GetIdentifier 

func (l *Loader) GetIdentifier() string

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

func (*Loader) GetPrimaryKey 

func (l *Loader) GetPrimaryKey(tableName string, pk string) (map[string]string, error)

func (*Loader) HasTable 

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

func (*Loader) Insert 

func (l *Loader) Insert(tableName string, primaryKey map[string]string, data map[string]FieldData, reversibleBlockNum *uint64) 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) LiveBlockFlushInterval 

func (l *Loader) LiveBlockFlushInterval() int

func (*Loader) LoadTables 

func (l *Loader) LoadTables(schemaName string, cursorTableName string, historyTableName string) error

func (*Loader) MarshalLogObject 

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

func (*Loader) NextBatchOrdinal 

func (l *Loader) NextBatchOrdinal() uint64

NextBatchOrdinal returns the next ordinal for the current batch and increments the counter

func (*Loader) Revert 

func (l *Loader) Revert(ctx context.Context, outputModuleHash string, cursor *sink.Cursor, lastValidBlock uint64) error

func (*Loader) Setup 

func (l *Loader) Setup(ctx context.Context, schemaName string, userSql string, withPostgraphile bool) error

Setup creates the schemaName, cursors and history table where the <schemaBytes> is a byte array taken from somewhere.

func (*Loader) Update 

func (l *Loader) Update(tableName string, primaryKey map[string]string, data map[string]FieldData, reversibleBlockNum *uint64) 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 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. You can use tx=nil to run the query outside of a transaction.

func (*Loader) Upsert 

func (l *Loader) Upsert(tableName string, primaryKey map[string]string, data map[string]FieldData, reversibleBlockNum *uint64) error

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

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"
	OperationTypeUpsert OperationType = "UPSERT"
	OperationTypeUpdate OperationType = "UPDATE"
	OperationTypeDelete OperationType = "DELETE"
)

type OrderedMap 

type OrderedMap[K comparable, V any] struct {
	*orderedmap.OrderedMap[K, V]
}

Make the typing a bit easier

func NewOrderedMap 

func NewOrderedMap[K comparable, V any]() *OrderedMap[K, V]

type PostgresDialect 

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

func NewPostgresDialect 

func NewPostgresDialect(schemaName string, cursorTableName string, historyTableName string) *PostgresDialect

func (PostgresDialect) AllowPkDuplicates 

func (d PostgresDialect) AllowPkDuplicates() bool

func (PostgresDialect) CreateUser 

func (d PostgresDialect) CreateUser(tx Tx, ctx context.Context, l *Loader, username string, password string, database string, readOnly bool) error

func (PostgresDialect) DriverSupportRowsAffected 

func (d PostgresDialect) DriverSupportRowsAffected() bool

func (PostgresDialect) ExecuteSetupScript 

func (d PostgresDialect) ExecuteSetupScript(ctx context.Context, l *Loader, schemaSql string) error

func (PostgresDialect) Flush 

func (d PostgresDialect) Flush(tx Tx, ctx context.Context, l *Loader, outputModuleHash string, lastFinalBlock uint64) (int, error)

func (PostgresDialect) GetAllCursorsQuery 

func (d PostgresDialect) GetAllCursorsQuery(table string) string

func (PostgresDialect) GetCreateCursorQuery 

func (d PostgresDialect) GetCreateCursorQuery(schema string, withPostgraphile bool) string

func (PostgresDialect) GetCreateHistoryQuery 

func (d PostgresDialect) GetCreateHistoryQuery(schema string, withPostgraphile bool) string

func (PostgresDialect) GetPrimaryKey 

func (d PostgresDialect) GetPrimaryKey(db *sql.DB, schemaName, tableName string) ([]string, error)

func (PostgresDialect) GetTableColumns 

func (d PostgresDialect) GetTableColumns(db *sql.DB, schemaName, tableName string) ([]*sql.ColumnType, error)

func (PostgresDialect) GetTablesInSchema 

func (d PostgresDialect) GetTablesInSchema(db *sql.DB, schemaName string) ([][2]string, error)

func (PostgresDialect) GetUpdateCursorQuery 

func (d PostgresDialect) GetUpdateCursorQuery(table, moduleHash string, cursor *sink.Cursor, block_num uint64, block_id string) string

func (PostgresDialect) OnlyInserts 

func (d PostgresDialect) OnlyInserts() bool

func (PostgresDialect) ParseDatetimeNormalization 

func (d PostgresDialect) ParseDatetimeNormalization(value string) string

func (PostgresDialect) Revert 

func (d PostgresDialect) Revert(tx Tx, ctx context.Context, l *Loader, lastValidFinalBlock uint64) error

type Queryable 

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

type SystemTableError 

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

type TableInfo 

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

func NewTableInfo 

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

type TestTx 

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

func (*TestTx) Commit 

func (t *TestTx) Commit() error

func (*TestTx) ExecContext 

func (t *TestTx) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)

func (*TestTx) QueryContext 

func (t *TestTx) QueryContext(ctx context.Context, query string, args ...any) (out *sql.Rows, err error)

func (*TestTx) Results 

func (t *TestTx) Results() []string

func (*TestTx) Rollback 

func (t *TestTx) Rollback() error

type Tx 

type Tx interface {
	Rollback() error
	Commit() error
	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
	QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
}

type TypeGetter 

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

type UnknownDriverError 

type UnknownDriverError struct {
	Driver string
}

func (UnknownDriverError) Error 

func (e UnknownDriverError) Error() string

Error returns a formatted string description.

type UpdateOp 

type UpdateOp int32

UpdateOp defines the operation to apply when updating a field on conflict 

const (
	UpdateOpSet       UpdateOp = 0 // Direct assignment: col = value
	UpdateOpAdd       UpdateOp = 1 // Accumulate: col = COALESCE(col, 0) + value
	UpdateOpMax       UpdateOp = 2 // Maximum: col = GREATEST(COALESCE(col, 0), value)
	UpdateOpMin       UpdateOp = 3 // Minimum: col = LEAST(COALESCE(col, 0), value)
	UpdateOpSetIfNull UpdateOp = 4 // Set only if NULL: col = COALESCE(col, value)
)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.