mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-05 08:10:50 +00:00
8fca15e936
* First pass at RetryWithTx * Refactor RetryWithTx to take a writable bool (instead of reads, writes) * Implment DirectiveMethodDiff This commit adds support for a Directive to contain only the diffs (as opposed to the full Directive). * Update controller tests to allow for DirectiveMethodDiff (over Full) * Update RetryWithTx to retry on duplicate key constraint. If two concurrent processes call IngestShard() for the same shard, both were trying to insert the same job into the jobs table. That resulted in a duplicate key error from the database. We want to include that error in the list of errors for which RetryWithTx should retry. * Remove unused method: Directive.TranslatePartitions() * Replace query in a loop with a single query We had a query which was looking to see if a job already existed. That query was inside a loop, and could potentially generate 256 queries (for example). This commit replaces that logic so that we use a single query wiht an `IN ()` clause. * Convert to directive version-by-address This commit uses a separate directive version per address. It moves the version get/increment back inside the buildDirective method so that if two concurrent processes are building a directive for the same address, one of them will get rolled back trying to commit the version update. * Migration for directive version by address * Add a comment about DirectiveVersion lock/unlock logic * Remove AddLastWins * fix linter * handle error in walkdir * fix test failures from removing AddLastWins
110 lines
2.9 KiB
Go
110 lines
2.9 KiB
Go
package sqldb
|
|
|
|
import (
|
|
"context"
|
|
|
|
"database/sql"
|
|
|
|
"github.com/featurebasedb/featurebase/v3/dax"
|
|
"github.com/featurebasedb/featurebase/v3/dax/controller"
|
|
"github.com/featurebasedb/featurebase/v3/errors"
|
|
"github.com/featurebasedb/featurebase/v3/logger"
|
|
"github.com/gobuffalo/pop/v6"
|
|
)
|
|
|
|
// Transactor wraps a pop Connection to make it into a dax.Transactor
|
|
// which can be used by the controller agnostic of implementation.
|
|
type Transactor struct {
|
|
*pop.Connection
|
|
|
|
logger logger.Logger
|
|
}
|
|
|
|
func NewTransactor(cfg *controller.SQLDBConfig, log logger.Logger) (Transactor, error) {
|
|
conn, err := pop.NewConnection(&pop.ConnectionDetails{
|
|
Dialect: cfg.Dialect,
|
|
Database: cfg.Database,
|
|
Host: cfg.Host,
|
|
Port: cfg.Port,
|
|
User: cfg.User,
|
|
Password: cfg.Password,
|
|
URL: cfg.URL,
|
|
Pool: cfg.Pool,
|
|
IdlePool: cfg.IdlePool,
|
|
ConnMaxLifetime: cfg.ConnMaxLifetime,
|
|
ConnMaxIdleTime: cfg.ConnMaxIdleTime,
|
|
})
|
|
if err != nil {
|
|
return Transactor{Connection: nil}, errors.Wrap(err, "creating new connection")
|
|
}
|
|
|
|
return Transactor{
|
|
Connection: conn,
|
|
logger: log,
|
|
}, nil
|
|
}
|
|
|
|
// Start creates the database specified in the database connection, then runs
|
|
// any outstanding migrations.
|
|
func (t Transactor) Start() error {
|
|
conn := t.Connection
|
|
|
|
// Create the database if it doesn't exist.
|
|
if err := pop.CreateDB(conn); err != nil {
|
|
t.logger.Warnf("auto-creating database, got error '%v'", err)
|
|
}
|
|
|
|
// Open a connection to the database.
|
|
if err := conn.Open(); err != nil {
|
|
return errors.Wrap(err, "opening connection")
|
|
}
|
|
|
|
// Run migrations.
|
|
if mig, err := NewEmbedMigrator(dax.MigrationsFS, conn, t.logger); err != nil {
|
|
return errors.Wrap(err, "getting embedded migrator")
|
|
} else if err = mig.Up(); err != nil {
|
|
return errors.Wrap(err, "migrating DB")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (t Transactor) BeginTx(ctx context.Context, writable bool) (dax.Transaction, error) {
|
|
cn, err := t.NewTransactionContextOptions(ctx, &sql.TxOptions{
|
|
Isolation: sql.LevelRepeatableRead,
|
|
ReadOnly: !writable,
|
|
})
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "getting SQL transaction")
|
|
}
|
|
return &DaxTransaction{C: cn}, nil
|
|
}
|
|
|
|
func (t Transactor) Close() error {
|
|
return t.Connection.Close()
|
|
}
|
|
|
|
// DaxTransaction is a thin wrapper to create a dax.Transaction from a
|
|
// pop Transaction/Connection.
|
|
type DaxTransaction struct {
|
|
C *pop.Connection
|
|
}
|
|
|
|
func (w *DaxTransaction) Commit() error {
|
|
return w.C.TX.Commit()
|
|
}
|
|
|
|
func (w *DaxTransaction) Context() context.Context {
|
|
return w.C.Context()
|
|
}
|
|
func (w *DaxTransaction) Rollback() error {
|
|
return w.C.TX.Rollback()
|
|
}
|
|
|
|
// DropDatabase drops the database associated with the given
|
|
// Transactor (which embeds a live database connection). This is
|
|
// destructive, you will lose data.
|
|
func DropDatabase(trans Transactor) error {
|
|
conn := trans.Connection
|
|
return pop.DropDB(conn)
|
|
}
|