Versions in this module Expand all Collapse all v1 v1.20.2 Jul 24, 2026 Changes in this version + var ErrCursorNotFound = errors.New("cursor not found") + var ErrInvalidOnModuleHashMismatch = fmt.Errorf("not a valid OnModuleHashMismatch, try [%s]", ...) + var PruneReversibleSegmentDuration = metrics.NewCounter("prune_reversible_segment_duration", ...) + var QueryExecutionDuration = metrics.NewCounterVec("tx_query_execution_duration", []string{ ... }, ...) + func EscapeIdentifier(valueToEscape string) string + func GenerateCreateTableSQL(tables map[string]*TableInfo) string + func OnModuleHashMismatchNames() []string + func RegisterMetrics() + func TestSinglePrimaryKeyTables(schema string) map[string]*TableInfo + func TestTables(schema string, customTable map[string]*TableInfo) map[string]*TableInfo + type ClickhouseDialect struct + func NewClickhouseDialect(schemaName string, cursorTableName string, cluster string) *ClickhouseDialect + func (d ClickhouseDialect) AllowPkDuplicates() bool + func (d ClickhouseDialect) CreateUser(tx Tx, ctx context.Context, l *Loader, username string, password string, ...) error + func (d ClickhouseDialect) DriverSupportRowsAffected() bool + func (d ClickhouseDialect) ExecuteSetupScript(ctx context.Context, l *Loader, schemaSql string) error + func (d ClickhouseDialect) Flush(tx Tx, ctx context.Context, l *Loader, outputModuleHash string, ...) (int, error) + func (d ClickhouseDialect) GetAllCursorsQuery(table string) string + func (d ClickhouseDialect) GetCreateCursorQuery(schema string, withPostgraphile bool) string + func (d ClickhouseDialect) GetCreateHistoryQuery(schema string, withPostgraphile bool) string + func (d ClickhouseDialect) GetPrimaryKey(db *sql.DB, schemaName, tableName string) ([]string, error) + func (d ClickhouseDialect) GetTableColumns(db *sql.DB, schemaName, tableName string) ([]*sql.ColumnType, error) + func (d ClickhouseDialect) GetTablesInSchema(db *sql.DB, schemaName string) ([][2]string, error) + func (d ClickhouseDialect) GetUpdateCursorQuery(table, moduleHash string, cursor *sink.Cursor, block_num uint64, ...) string + func (d ClickhouseDialect) OnlyInserts() bool + func (d ClickhouseDialect) ParseDatetimeNormalization(value string) string + func (d ClickhouseDialect) Revert(tx Tx, ctx context.Context, l *Loader, lastValidFinalBlock uint64) error + type ColumnInfo struct + func NewColumnInfo(name string, databaseTypeName string, scanType any) *ColumnInfo + type DSN struct + Database string + Host string + Options DSNOptions + Password string + Port int64 + Username string + func ParseDSN(dsn string) (*DSN, error) + func (c *DSN) Clone() *DSN + func (c *DSN) ConnString() string + func (c *DSN) Driver() string + func (c *DSN) Schema() string + type DSNOptions url.Values + func (v DSNOptions) Encode() string + func (v DSNOptions) EncodeWithSeparator(sep string) string + func (v DSNOptions) Get(key string) string + func (v DSNOptions) GetOr(key, defaultValue string) string + func (v DSNOptions) Iter() iter.Seq2[string, string] + func (v DSNOptions) RemoveOr(key, defaultValue string) string + type Dialect interface + AllowPkDuplicates func() bool + CreateUser func(tx Tx, ctx context.Context, l *Loader, username string, password string, ...) error + DriverSupportRowsAffected func() bool + ExecuteSetupScript func(ctx context.Context, l *Loader, schemaSql string) error + Flush func(tx Tx, ctx context.Context, l *Loader, outputModuleHash string, ...) (int, error) + GetAllCursorsQuery func(table string) string + GetCreateCursorQuery func(schema string, withPostgraphile bool) string + GetCreateHistoryQuery func(schema string, withPostgraphile bool) string + GetPrimaryKey func(db *sql.DB, schemaName, tableName string) ([]string, error) + GetTableColumns func(db *sql.DB, schemaName, tableName string) ([]*sql.ColumnType, error) + GetTablesInSchema func(db *sql.DB, schemaName string) ([][2]string, error) + GetUpdateCursorQuery func(table, moduleHash string, cursor *sink.Cursor, block_num uint64, ...) string + OnlyInserts func() bool + ParseDatetimeNormalization func(value string) string + Revert func(tx Tx, ctx context.Context, l *Loader, lastValidFinalBlock uint64) error + type FieldData struct + UpdateOp UpdateOp + Value string + type Loader struct + func NewLoader(dsn *DSN, cursorTableName string, historyTableName string, ...) (*Loader, error) + func NewTestLoader(t *testing.T, dsnRaw string, testTx *TestTx, tables map[string]*TableInfo, ...) *Loader + func (l *Loader) BatchBlockFlushInterval() int + func (l *Loader) Begin() (Tx, error) + func (l *Loader) BeginTx(ctx context.Context, opts *sql.TxOptions) (Tx, error) + func (l *Loader) CreateUser(ctx context.Context, username string, password string, database string, ...) (err error) + func (l *Loader) Delete(tableName string, primaryKey map[string]string, reversibleBlockNum *uint64) error + func (l *Loader) DeleteAllCursors(ctx context.Context) (deletedCount int64, err error) + func (l *Loader) DeleteCursor(ctx context.Context, moduleHash string) error + func (l *Loader) Flush(ctx context.Context, outputModuleHash string, cursor *sink.Cursor, ...) (rowFlushedCount int, err error) + func (l *Loader) FlushNeeded() bool + func (l *Loader) GetAllCursors(ctx context.Context) (out map[string]*sink.Cursor, err error) + func (l *Loader) GetAvailableTablesInSchema() []string + func (l *Loader) GetColumnsForTable(name string) []string + func (l *Loader) GetCursor(ctx context.Context, outputModuleHash string) (cursor *sink.Cursor, mismatchDetected bool, err error) + func (l *Loader) GetDSN() *DSN + func (l *Loader) GetIdentifier() string + func (l *Loader) GetPrimaryKey(tableName string, pk string) (map[string]string, error) + func (l *Loader) HasTable(tableName string) bool + func (l *Loader) Insert(tableName string, primaryKey map[string]string, data map[string]FieldData, ...) error + func (l *Loader) InsertCursor(ctx context.Context, moduleHash string, c *sink.Cursor) error + func (l *Loader) LiveBlockFlushInterval() int + func (l *Loader) LoadTables(schemaName string, cursorTableName string, historyTableName string) error + func (l *Loader) MarshalLogObject(encoder zapcore.ObjectEncoder) error + func (l *Loader) NextBatchOrdinal() uint64 + func (l *Loader) Revert(ctx context.Context, outputModuleHash string, cursor *sink.Cursor, ...) error + func (l *Loader) Setup(ctx context.Context, schemaName string, userSql string, withPostgraphile bool) error + func (l *Loader) Update(tableName string, primaryKey map[string]string, data map[string]FieldData, ...) error + func (l *Loader) UpdateCursor(ctx context.Context, tx Tx, moduleHash string, c *sink.Cursor) error + func (l *Loader) Upsert(tableName string, primaryKey map[string]string, data map[string]FieldData, ...) error + type OnModuleHashMismatch uint + const OnModuleHashMismatchError + const OnModuleHashMismatchIgnore + const OnModuleHashMismatchWarn + func ParseOnModuleHashMismatch(name string) (OnModuleHashMismatch, error) + func (x *OnModuleHashMismatch) UnmarshalText(text []byte) error + func (x OnModuleHashMismatch) IsValid() bool + func (x OnModuleHashMismatch) MarshalText() ([]byte, error) + func (x OnModuleHashMismatch) String() string + type Operation struct + func (o *Operation) String() string + type OperationType string + const OperationTypeDelete + const OperationTypeInsert + const OperationTypeUpdate + const OperationTypeUpsert + type OrderedMap struct + func NewOrderedMap[K comparable, V any]() *OrderedMap[K, V] + type PostgresDialect struct + func NewPostgresDialect(schemaName string, cursorTableName string, historyTableName string) *PostgresDialect + func (d PostgresDialect) AllowPkDuplicates() bool + func (d PostgresDialect) CreateUser(tx Tx, ctx context.Context, l *Loader, username string, password string, ...) error + func (d PostgresDialect) DriverSupportRowsAffected() bool + func (d PostgresDialect) ExecuteSetupScript(ctx context.Context, l *Loader, schemaSql string) error + func (d PostgresDialect) Flush(tx Tx, ctx context.Context, l *Loader, outputModuleHash string, ...) (int, error) + func (d PostgresDialect) GetAllCursorsQuery(table string) string + func (d PostgresDialect) GetCreateCursorQuery(schema string, withPostgraphile bool) string + func (d PostgresDialect) GetCreateHistoryQuery(schema string, withPostgraphile bool) string + func (d PostgresDialect) GetPrimaryKey(db *sql.DB, schemaName, tableName string) ([]string, error) + func (d PostgresDialect) GetTableColumns(db *sql.DB, schemaName, tableName string) ([]*sql.ColumnType, error) + func (d PostgresDialect) GetTablesInSchema(db *sql.DB, schemaName string) ([][2]string, error) + func (d PostgresDialect) GetUpdateCursorQuery(table, moduleHash string, cursor *sink.Cursor, block_num uint64, ...) string + func (d PostgresDialect) OnlyInserts() bool + func (d PostgresDialect) ParseDatetimeNormalization(value string) string + func (d PostgresDialect) Revert(tx Tx, ctx context.Context, l *Loader, lastValidFinalBlock uint64) error + type Queryable interface + type SystemTableError struct + type TableInfo struct + func NewTableInfo(schema, name string, pkList []string, columnsByName map[string]*ColumnInfo) (*TableInfo, error) + type TestTx struct + func (t *TestTx) Commit() error + func (t *TestTx) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) + func (t *TestTx) QueryContext(ctx context.Context, query string, args ...any) (out *sql.Rows, err error) + func (t *TestTx) Results() []string + func (t *TestTx) Rollback() error + type Tx interface + Commit func() error + ExecContext func(ctx context.Context, query string, args ...any) (sql.Result, error) + QueryContext func(ctx context.Context, query string, args ...any) (*sql.Rows, error) + Rollback func() error + type TypeGetter func(tableName string, columnName string) (reflect.Type, error) + type UnknownDriverError struct + Driver string + func (e UnknownDriverError) Error() string + type UpdateOp int32 + const UpdateOpAdd + const UpdateOpMax + const UpdateOpMin + const UpdateOpSet + const UpdateOpSetIfNull
Go to main page