dbutil

Documentation

Index

• Constants
• Variables
• func BuildConditions(equal map[string]interface{}, in map[string]interface{}) ([]string, []interface{}, error)
• func BuildIndexDDL(tableName string, indexName string, fields []string, backend types.BackendType) string
• func BuildTableSchema(params BuildTableSchemaParams) string
• func DBValueType(backend types.BackendType, valueType types.ValueType) (string, error)
• func DeserializeByValueType(i interface{}, valueType types.ValueType, backend types.BackendType) (interface{}, error)
• func DropTable(ctx context.Context, dbOpt DBOpt, tableName string) error
• func Fill(size int, elem string, sep string) string
• func GetColumnFormat(backendType types.BackendType) (string, error)
• func InsertRecordsToTable(ctx context.Context, dbOpt DBOpt, tableName string, records []interface{}, ...) error
• func InsertRecordsToTableTx(tx *sqlx.Tx, ctx context.Context, tableName string, records []interface{}, ...) error
• func IsTableNotFoundError(err error, backend types.BackendType) (bool, error)
• func LoadDataFromSource(backend types.BackendType, batchSize int) func(tx *sqlx.Tx, ctx context.Context, opt LoadDataFromSourceOpt) error
• func OfflineBatchSnapshotTableName(groupID int, revisionID int64) string
• func OfflineStreamCdcTableName(groupID int, revision int64) string
• func OfflineStreamSnapshotTableName(groupID int, revision int64) string
• func OnlineBatchTableName(revisionID int) string
• func OnlineStreamTableName(groupID int) string
• func OpenMysqlDB(host, port, user, password, database string) (*sqlx.DB, error)
• func OpenPostgresDB(host, port, user, password, database string) (*sqlx.DB, error)
• func OpenSQLite(dbFile string) (*sqlx.DB, error)
• func QuoteFn(backendType types.BackendType) func(...string) string
• func RandString(n int) string
• func ReadLine(opt ReadLineOpt) ([]interface{}, error)
• func SerializeByValue(i interface{}, backend types.BackendType) (string, error)
• func SerializeByValueType(i interface{}, valueType types.ValueType, backend types.BackendType) (interface{}, error)
• func TempTable(prefix string) string
• func ValueType(backend types.BackendType, dbValueType string) (types.ValueType, error)
• func WithTransaction(db *sqlx.DB, ctx context.Context, fn TxFn) (err error)
• type BuildTableSchemaParams
• type Column
• type ColumnList
  • func (c ColumnList) Names() []string
• type DBOpt
  • func (d *DBOpt) BuildInsertQuery(tableName string, records []interface{}, columns []string) (string, []interface{}, error)
  • func (d *DBOpt) ExecContext(ctx context.Context, query string, args ...interface{}) error
• type LoadDataFromSourceOpt
• type ReadLineOpt
• type RowMap
• type TxFn

Constants

const (
	MYSQL_NO_SUCH_TABLE = 1146
)

Variables

var OpenRedshiftDB = OpenPostgresDB

Functions

func BuildConditions

func BuildConditions(equal map[string]interface{}, in map[string]interface{}) ([]string, []interface{}, error)

func BuildIndexDDL

func BuildIndexDDL(tableName string, indexName string, fields []string, backend types.BackendType) string

func BuildTableSchema

func BuildTableSchema(params BuildTableSchemaParams) string

func DBValueType

func DBValueType(backend types.BackendType, valueType types.ValueType) (string, error)

func DeserializeByValueType

func DeserializeByValueType(i interface{}, valueType types.ValueType, backend types.BackendType) (interface{}, error)

func DropTable

func DropTable(ctx context.Context, dbOpt DBOpt, tableName string) error

func Fill

func Fill(size int, elem string, sep string) string

func GetColumnFormat

func GetColumnFormat(backendType types.BackendType) (string, error)

func InsertRecordsToTable

func InsertRecordsToTable(ctx context.Context, dbOpt DBOpt, tableName string, records []interface{}, columns []string) error

func InsertRecordsToTableTx

func InsertRecordsToTableTx(tx *sqlx.Tx, ctx context.Context, tableName string, records []interface{}, columns []string, backendType types.BackendType) error

func IsTableNotFoundError

func IsTableNotFoundError(err error, backend types.BackendType) (bool, error)

func LoadDataFromSource

func LoadDataFromSource(backend types.BackendType, batchSize int) func(tx *sqlx.Tx, ctx context.Context, opt LoadDataFromSourceOpt) error

Currying

func OfflineBatchSnapshotTableName

func OfflineBatchSnapshotTableName(groupID int, revisionID int64) string

func OfflineStreamCdcTableName

func OfflineStreamCdcTableName(groupID int, revision int64) string

func OfflineStreamSnapshotTableName

func OfflineStreamSnapshotTableName(groupID int, revision int64) string

func OnlineBatchTableName

func OnlineBatchTableName(revisionID int) string

func OnlineStreamTableName

func OnlineStreamTableName(groupID int) string

func OpenMysqlDB

func OpenMysqlDB(host, port, user, password, database string) (*sqlx.DB, error)

func OpenPostgresDB

func OpenPostgresDB(host, port, user, password, database string) (*sqlx.DB, error)

func OpenSQLite

func OpenSQLite(dbFile string) (*sqlx.DB, error)

func QuoteFn

func QuoteFn(backendType types.BackendType) func(...string) string

func RandString

func RandString(n int) string

func ReadLine

func ReadLine(opt ReadLineOpt) ([]interface{}, error)

ReadLine read a line from data source

func SerializeByValue

func SerializeByValue(i interface{}, backend types.BackendType) (string, error)

func SerializeByValueType

func SerializeByValueType(i interface{}, valueType types.ValueType, backend types.BackendType) (interface{}, error)

func TempTable

func TempTable(prefix string) string

func ValueType

func ValueType(backend types.BackendType, dbValueType string) (types.ValueType, error)

Used for inferring feature value type from a supported offline db value type

func WithTransaction

func WithTransaction(db *sqlx.DB, ctx context.Context, fn TxFn) (err error)

Types

type BuildTableSchemaParams

type BuildTableSchemaParams struct {
	TableName    string
	EntityName   string
	HasUnixMilli bool
	Features     types.FeatureList
	PrimaryKeys  []string
	Backend      types.BackendType
}

type Column

type Column struct {
	Name      string
	DbType    string
	ValueType types.ValueType
}

type ColumnList

type ColumnList []Column

func (ColumnList) Names

func (c ColumnList) Names() []string

type DBOpt

type DBOpt struct {
	Backend types.BackendType

	// Sqlx
	SqlxDB *sqlx.DB

	// BigQuery
	BigQueryDB *bigquery.Client
	DatasetID  *string

	// Cassandra
	CassandraDB *gocql.Session
}

func (*DBOpt) BuildInsertQuery

func (d *DBOpt) BuildInsertQuery(tableName string, records []interface{}, columns []string) (string, []interface{}, error)

func (*DBOpt) ExecContext

func (d *DBOpt) ExecContext(ctx context.Context, query string, args ...interface{}) error

type LoadDataFromSourceOpt

type LoadDataFromSourceOpt struct {
	Source     *offline.CSVSource
	EntityName string
	TableName  string
	Header     []string
	Features   types.FeatureList
	Backend    types.BackendType
}

type ReadLineOpt

type ReadLineOpt struct {
	Source     *offline.CSVSource
	EntityName string
	Header     []string
	Features   types.FeatureList
}

type RowMap

type RowMap = map[string]interface{}

type TxFn

type TxFn func(ctx context.Context, tx *sqlx.Tx) error

A Txfn is a function that will be called with an initialized `Transaction` object that can be used for executing statements and queries against a database.