Implement prepared statements caching for insert row.

masterypro · Apr 25, 2016 · fb7539c · fb7539c
1 parent ee355f2
commit fb7539c
Show file tree

Hide file tree

Showing 22 changed files with 70 additions and 33 deletions.
diff --git a/action/action.go b/action/action.go
@@ -2,7 +2,7 @@ package action
 
 type Action interface {
 	// Execute the given action
-	Execute(c Context) error
+	Execute(c *Context) error
 	// Returns whether current action should be executed
 	// for a targetExpression
 	Filter(targetExpression string) bool

diff --git a/action/altercolumn.go b/action/altercolumn.go
@@ -17,7 +17,7 @@ func init() {
 	gob.Register(&AlterColumn{})
 }
 
-func (a *AlterColumn) Execute(c Context) error {
+func (a *AlterColumn) Execute(c *Context) error {
 	if a.Column.Name != a.NewColumn.Name {
 		_, err := c.Tx.Exec(
 			fmt.Sprintf(

diff --git a/action/alterschema.go b/action/alterschema.go
@@ -15,7 +15,7 @@ func init() {
 	gob.Register(&AlterSchema{})
 }
 
-func (a *AlterSchema) Execute(c Context) error {
+func (a *AlterSchema) Execute(c *Context) error {
 	_, err := c.Tx.Exec(
 		fmt.Sprintf("ALTER SCHEMA \"%s\" RENAME TO \"%s\";", a.SourceName, a.TargetName),
 	)

diff --git a/action/altertable.go b/action/altertable.go
@@ -16,7 +16,7 @@ func init() {
 	gob.Register(&AlterTable{})
 }
 
-func (a *AlterTable) Execute(c Context) error {
+func (a *AlterTable) Execute(c *Context) error {
 	_, err := c.Tx.Exec(
 		fmt.Sprintf("ALTER TABLE \"%s\".\"%s\" RENAME TO \"%s\";", a.SchemaName, a.SourceName, a.TargetName),
 	)

diff --git a/action/altertype.go b/action/altertype.go
@@ -16,7 +16,7 @@ func init() {
 	gob.Register(&AlterType{})
 }
 
-func (a *AlterType) Execute(c Context) error {
+func (a *AlterType) Execute(c *Context) error {
 	_, err := c.Tx.Exec(
 		fmt.Sprintf("ALTER TYPE \"%s\".\"%s\" RENAME TO \"%s\";", a.SchemaName, a.SourceName, a.TargetName),
 	)

diff --git a/action/context.go b/action/context.go
@@ -2,11 +2,45 @@ package action
 
 import (
 	"github.com/jmoiron/sqlx"
+	"crypto/md5"
 )
 
 // Defines the execution context of actions
 type Context struct {
 	Tx *sqlx.Tx
 	Db *sqlx.DB
+	stmtCache map[string]*sqlx.Stmt
 }
 
+func NewContext(tx *sqlx.Tx, db *sqlx.DB) *Context {
+	return &Context{
+		tx,
+		db,
+		make(map[string]*sqlx.Stmt),
+	}
+}
+
+func (c *Context) getHash(str string) string {
+	h := md5.New()
+	return string(h.Sum([]byte(str)))
+}
+
+func (c *Context) GetPreparedStatement(statement string) (*sqlx.Stmt, error) {
+	hash := c.getHash(statement)
+
+	if stmt, ok := c.stmtCache[hash]; ok {
+		return stmt, nil
+	} else {
+		stmt, err := c.Tx.Preparex(statement)
+
+		if err != nil {
+			return nil, err
+		}
+
+		c.stmtCache[hash] = stmt
+
+		return stmt, nil
+	}
+
+	return nil, nil
+}
diff --git a/action/createcolumn.go b/action/createcolumn.go
@@ -16,7 +16,7 @@ func init() {
 	gob.Register(&CreateColumn{})
 }
 
-func (a *CreateColumn) Execute(c Context) error {
+func (a *CreateColumn) Execute(c *Context) error {
 	_, err := c.Tx.Exec(
 		fmt.Sprintf(
 			"ALTER TABLE \"%s\".\"%s\" ADD COLUMN \"%s\" %s\"%s\";",

diff --git a/action/createenum.go b/action/createenum.go
@@ -16,7 +16,7 @@ func init() {
 	gob.Register(&CreateEnum{})
 }
 
-func (a *CreateEnum) Execute(c Context) error {
+func (a *CreateEnum) Execute(c *Context) error {
 	// ALTER TYPE... ADD VALUE cannot run inside a transaction
 	_, err := c.Db.Exec(
 		fmt.Sprintf(

diff --git a/action/createschema.go b/action/createschema.go
@@ -14,7 +14,7 @@ func init() {
 	gob.Register(&CreateSchema{})
 }
 
-func (a *CreateSchema) Execute(c Context) error {
+func (a *CreateSchema) Execute(c *Context) error {
 	_, err := c.Tx.Exec(
 		fmt.Sprintf("CREATE SCHEMA IF NOT EXISTS \"%s\";", a.SchemaName),
 	)

diff --git a/action/createtable.go b/action/createtable.go
@@ -17,7 +17,7 @@ func init() {
 	gob.Register(&CreateTable{})
 }
 
-func (a *CreateTable) Execute(c Context) error {
+func (a *CreateTable) Execute(c *Context) error {
 	_, err := c.Tx.Exec(
 		fmt.Sprintf(
 			"CREATE TABLE \"%s\".\"%s\" (\"%s\" %s\"%s\" PRIMARY KEY);",

diff --git a/action/createtrigger.go b/action/createtrigger.go
@@ -22,7 +22,7 @@ func init() {
 	gob.Register(&CreateTrigger{})
 }
 
-func (a *CreateTrigger) Execute(c Context) error {
+func (a *CreateTrigger) Execute(c *Context) error {
 	_, err := c.Tx.Exec(
 		fmt.Sprintf(
 			`

diff --git a/action/createtype.go b/action/createtype.go
@@ -15,7 +15,7 @@ func init() {
 	gob.Register(&CreateType{})
 }
 
-func (a *CreateType) Execute(c Context) error {
+func (a *CreateType) Execute(c *Context) error {
 	_, err := c.Tx.Exec(
 		fmt.Sprintf(
 			"CREATE TYPE \"%s\".\"%s\" AS ENUM ();",

diff --git a/action/deleterow.go b/action/deleterow.go
@@ -16,7 +16,7 @@ func init() {
 	gob.Register(&DeleteRow{})
 }
 
-func (a *DeleteRow) Execute(c Context) error {
+func (a *DeleteRow) Execute(c *Context) error {
 	_, err := c.Tx.Exec(
 		fmt.Sprintf(
 			`DELETE FROM "%s"."%s" WHERE "%s" = $1;`,

diff --git a/action/dropcolumn.go b/action/dropcolumn.go
@@ -16,7 +16,7 @@ func init() {
 	gob.Register(&DropColumn{})
 }
 
-func (a *DropColumn) Execute(c Context) error {
+func (a *DropColumn) Execute(c *Context) error {
 	_, err := c.Tx.Exec(
 		fmt.Sprintf("ALTER TABLE \"%s\".\"%s\" DROP COLUMN \"%s\";", a.SchemaName, a.TableName, a.Column.Name),
 	)

diff --git a/action/dropschema.go b/action/dropschema.go
@@ -14,7 +14,7 @@ func init() {
 	gob.Register(&DropSchema{})
 }
 
-func (a *DropSchema) Execute(c Context) error {
+func (a *DropSchema) Execute(c *Context) error {
 	_, err := c.Tx.Exec(
 		fmt.Sprintf("DROP SCHEMA \"%s\";", a.SchemaName),
 	)

diff --git a/action/droptable.go b/action/droptable.go
@@ -15,7 +15,7 @@ func init() {
 	gob.Register(&DropTable{})
 }
 
-func (a *DropTable) Execute(c Context) error {
+func (a *DropTable) Execute(c *Context) error {
 	_, err := c.Tx.Exec(
 		fmt.Sprintf("DROP TABLE \"%s\".\"%s\";", a.SchemaName, a.TableName),
 	)

diff --git a/action/droptrigger.go b/action/droptrigger.go
@@ -16,7 +16,7 @@ func init() {
 	gob.Register(&DropTrigger{})
 }
 
-func (a *DropTrigger) Execute(c Context) error {
+func (a *DropTrigger) Execute(c *Context) error {
 	_, err := c.Tx.Exec(
 		fmt.Sprintf(
 			`DROP TRIGGER IF EXISTS "%s" ON "%s"."%s";`,

diff --git a/action/droptype.go b/action/droptype.go
@@ -15,7 +15,7 @@ func init() {
 	gob.Register(&DropType{})
 }
 
-func (a *DropType) Execute(c Context) error {
+func (a *DropType) Execute(c *Context) error {
 	_, err := c.Tx.Exec(
 		fmt.Sprintf("DROP TYPE \"%s\".\"%s\";", a.SchemaName, a.TypeName),
 	)

diff --git a/action/insertrow.go b/action/insertrow.go
@@ -21,7 +21,7 @@ func init() {
 	gob.Register(&time.Time{})
 }
 
-func (a *InsertRow) Execute(c Context) error {
+func (a *InsertRow) Execute(c *Context) error {
 	escapedCols := make([]string, 0)
 	escapedRows := make([]string, 0)
 	values := make([]interface{}, 0)
@@ -62,7 +62,7 @@ func (a *InsertRow) Execute(c Context) error {
 		return err
 	}
 
-	_, err = c.Tx.Exec(
+	stmt, err := c.GetPreparedStatement(
 		fmt.Sprintf(
 			`
 				INSERT INTO "%s"."%s" (%s) VALUES (%s);
@@ -72,9 +72,14 @@ func (a *InsertRow) Execute(c Context) error {
 			strings.Join(escapedCols, ","),
 			strings.Join(escapedRows, ","),
 		),
-		values...,
 	)
 
+	if err != nil {
+		return err
+	}
+
+	_, err = stmt.Exec(values...)
+
 	// Try to UPDATE (upsert) if INSERT fails...
 	if err != nil {
 		// Rollback to SAVEPOINT

diff --git a/action/updaterow.go b/action/updaterow.go
@@ -19,7 +19,7 @@ func init() {
 	gob.Register(&UpdateRow{})
 }
 
-func (a *UpdateRow) Execute(c Context) error {
+func (a *UpdateRow) Execute(c *Context) error {
 	escapedRows := make([]string, 0)
 	replacementsCount := 0
 	values := make([]interface{}, 0)

diff --git a/applier/applier.go b/applier/applier.go
@@ -1,7 +1,6 @@
 package applier
 
 import (
-	"github.com/jmoiron/sqlx"
 	"github.com/pagarme/teleport/action"
 	"github.com/pagarme/teleport/database"
 	"log"
@@ -46,12 +45,9 @@ func (a *Applier) Watch(sleepTime time.Duration) {
 	}
 }
 
-func (a *Applier) applyAction(act action.Action, tx *sqlx.Tx) error {
+func (a *Applier) applyAction(act action.Action, context *action.Context) error {
 	// Execute action of the given event
-	err := act.Execute(action.Context{
-		Tx: tx,
-		Db: a.db.Db,
-	})
+	err := act.Execute(context)
 
 	if err != nil {
 		log.Printf("Error applying action %#v: %v", act, err)
@@ -109,8 +105,10 @@ func (a *Applier) applyBatch(batch *database.Batch) (bool, error) {
 			return false, err
 		}
 
+		context := action.NewContext(tx, a.db.Db)
+
 		for _, act := range actions {
-			err := a.applyAction(act, tx)
+			err := a.applyAction(act, context)
 
 			if err != nil {
 				return updateBatchStatus(err)
@@ -130,6 +128,7 @@ func (a *Applier) applyBatch(batch *database.Batch) (bool, error) {
 		currentBatchSize := 0
 		previousStatement := batch.LastExecutedStatement
 
+		currentContext := action.NewContext(tx, a.db.Db)
 		act, err = batch.ReadAction(reader);
 
 		for err == nil {
@@ -138,7 +137,7 @@ func (a *Applier) applyBatch(batch *database.Batch) (bool, error) {
 
 			// Start applying from previous stop point
 			if currentStatement > previousStatement {
-				err = a.applyAction(act, tx)
+				err = a.applyAction(act, currentContext)
 
 				if err != nil {
 					return updateBatchStatus(err)
@@ -165,6 +164,8 @@ func (a *Applier) applyBatch(batch *database.Batch) (bool, error) {
 					// Restart transaction
 					tx = a.db.NewTransaction()
 
+					currentContext = action.NewContext(tx, a.db.Db)
+
 					// Reset batch size
 					currentBatchSize = 0
 				}

diff --git a/database/class.go b/database/class.go
@@ -48,10 +48,7 @@ func (c *Class) InstallTriggers() error {
 	tx := c.Schema.Db.NewTransaction()
 
 	for _, currentAction := range actions {
-		err := currentAction.Execute(action.Context{
-			Tx: tx,
-			Db: c.Schema.Db.Db,
-		})
+		err := currentAction.Execute(action.NewContext(tx, c.Schema.Db.Db,))
 
 		if err != nil {
 			log.Printf("Error creating triggers on %s: %v", c.RelationName, err)