Merge pull request #275 from cschleiden/mysql-attributes-table

Move mysql attributes to separate table

Author: cschleiden

backend/mysql/db/migrations/000003_add_attributes_table.down.sql

@@ -0,0 +1,15 @@
+ALTER TABLE `activities` ADD COLUMN `attributes` MEDIUMBLOB NULL;
+UPDATE `activities` SET `attributes` = `attributes`.`data` FROM `attributes` WHERE `activities`.`event_id` = `attributes`.`event_id` AND `activities`.`instance_id` = `attributes`.`instance_id` AND `activities`.`execution_id` = `attributes`.`execution_id`;
+ALTER TABLE `activities` MODIFY COLUMN `attributes` MEDIUMBLOB NOT NULL;
+
+ALTER TABLE `history` ADD COLUMN `attributes` MEDIUMBLOB NULL;
+UPDATE `history` SET `attributes` = `attributes`.`data` FROM `attributes` WHERE `history`.`event_id` = `attributes`.`event_id` AND `history`.`instance_id` = `attributes`.`instance_id` AND `history`.`execution_id` = `attributes`.`execution_id`;
+ALTER TABLE `history` MODIFY COLUMN `attributes` MEDIUMBLOB NOT NULL;
+
+ALTER TABLE `pending_events` ADD COLUMN `attributes` MEDIUMBLOB NULL;
+UPDATE `pending_events` SET `attributes` = `attributes`.`data` FROM `attributes` WHERE `pending_events`.`event_id` = `attributes`.`event_id` AND `pending_events`.`instance_id` = `attributes`.`instance_id` AND `pending_events`.`execution_id` = `attributes`.`execution_id`;
+ALTER TABLE `pending_events` MODIFY COLUMN `attributes` MEDIUMBLOB NOT NULL;
+
+
+-- Drop attributes table
+DROP TABLE `attributes`;

backend/mysql/db/migrations/000003_add_attributes_table.up.sql

@@ -0,0 +1,22 @@
+CREATE TABLE IF NOT EXISTS `attributes` (
+  `id` BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+  `event_id` NVARCHAR(128) NOT NULL,
+  `instance_id` NVARCHAR(128) NOT NULL,
+  `execution_id` NVARCHAR(128) NOT NULL,
+  `data` MEDIUMBLOB NOT NULL,
+
+  UNIQUE INDEX `idx_attributes_instance_id_execution_id_event_id` (`instance_id`, `execution_id`, `event_id`),
+  INDEX `idx_attributes_event_id` (`event_id`)
+);
+
+-- Move activity attributes to attributes table
+INSERT IGNORE INTO `attributes` (`event_id`, `instance_id`, `execution_id`, `data`) SELECT `activity_id`, `instance_id`, `execution_id`, `attributes` FROM `activities`;
+ALTER TABLE `activities` DROP COLUMN `attributes`;
+
+-- Move history attributes to attributes table
+INSERT IGNORE INTO `attributes` (`event_id`, `instance_id`, `execution_id`, `data`) SELECT `event_id`, `instance_id`, `execution_id`, `attributes` FROM `history`;
+ALTER TABLE `history` DROP COLUMN `attributes`;
+
+-- Move pending_events attributes to attributes table
+INSERT IGNORE INTO `attributes` (`event_id`, `instance_id`, `execution_id`, `data`) SELECT `event_id`, `instance_id`, `execution_id`, `attributes` FROM `pending_events`;
+ALTER TABLE `pending_events` DROP COLUMN `attributes`;

backend/mysql/events.go

@@ -3,6 +3,7 @@ package mysql
 import (
 	"context"
 	"database/sql"
+	"fmt"
 	"strings"
 
 	"github.com/cschleiden/go-workflows/backend/history"
@@ -26,21 +27,34 @@ func insertEvents(ctx context.Context, tx *sql.Tx, tableName string, instance *c
 		}
 		batchEvents := events[batchStart:batchEnd]
 
+		aquery := "INSERT IGNORE INTO `attributes` (event_id, instance_id, execution_id, data) VALUES (?, ?, ?, ?)" + strings.Repeat(", (?, ?, ?, ?)", len(batchEvents)-1)
+		aargs := make([]interface{}, 0, len(batchEvents)*4)
+
 		query := "INSERT INTO `" + tableName +
-			"` (event_id, sequence_id, instance_id, execution_id, event_type, timestamp, schedule_event_id, attributes, visible_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)" +
-			strings.Repeat(", (?, ?, ?, ?, ?, ?, ?, ?, ?)", len(batchEvents)-1)
+			"` (event_id, sequence_id, instance_id, execution_id, event_type, timestamp, schedule_event_id, visible_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)" +
+			strings.Repeat(", (?, ?, ?, ?, ?, ?, ?, ?)", len(batchEvents)-1)
 
-		args := make([]interface{}, 0, len(batchEvents)*7)
+		args := make([]interface{}, 0, len(batchEvents)*8)
 
 		for _, newEvent := range batchEvents {
 			a, err := history.SerializeAttributes(newEvent.Attributes)
 			if err != nil {
 				return err
 			}
 
+			aargs = append(aargs, newEvent.ID, instance.InstanceID, instance.ExecutionID, a)
+
 			args = append(
 				args,
-				newEvent.ID, newEvent.SequenceID, instance.InstanceID, instance.ExecutionID, newEvent.Type, newEvent.Timestamp, newEvent.ScheduleEventID, a, newEvent.VisibleAt)
+				newEvent.ID, newEvent.SequenceID, instance.InstanceID, instance.ExecutionID, newEvent.Type, newEvent.Timestamp, newEvent.ScheduleEventID, newEvent.VisibleAt)
+		}
+
+		if _, err := tx.ExecContext(
+			ctx,
+			aquery,
+			aargs...,
+		); err != nil {
+			return fmt.Errorf("inserting attributes: %w", err)
 		}
 
 		_, err := tx.ExecContext(
@@ -49,7 +63,7 @@ func insertEvents(ctx context.Context, tx *sql.Tx, tableName string, instance *c
 			args...,
 		)
 		if err != nil {
-			return err
+			return fmt.Errorf("inserting events: %w", err)
 		}
 	}
 
@@ -59,7 +73,7 @@ func insertEvents(ctx context.Context, tx *sql.Tx, tableName string, instance *c
 func removeFutureEvent(ctx context.Context, tx *sql.Tx, instance *core.WorkflowInstance, scheduleEventID int64) error {
 	_, err := tx.ExecContext(
 		ctx,
-		"DELETE FROM `pending_events` WHERE instance_id = ? AND execution_id = ? AND schedule_event_id = ? AND visible_at IS NOT NULL",
+		"DELETE `pending_events`, `attributes` FROM `pending_events` INNER JOIN `attributes` ON `pending_events`.event_id = `attributes`.event_id WHERE `pending_events`.instance_id = ? AND `pending_events`.execution_id = ? AND `pending_events`.schedule_event_id = ? AND `pending_events`.visible_at IS NOT NULL",
 		instance.InstanceID,
 		instance.ExecutionID,
 		scheduleEventID,

backend/mysql/mysql.go

@@ -190,6 +190,10 @@ func (b *mysqlBackend) RemoveWorkflowInstance(ctx context.Context, instance *cor
 		return err
 	}
 
+	if _, err := tx.ExecContext(ctx, "DELETE FROM `attributes` WHERE instance_id = ? AND execution_id = ?", instance.InstanceID, instance.ExecutionID); err != nil {
+		return err
+	}
+
 	return tx.Commit()
 }
 
@@ -231,15 +235,15 @@ func (b *mysqlBackend) GetWorkflowInstanceHistory(ctx context.Context, instance
 	if lastSequenceID != nil {
 		historyEvents, err = tx.QueryContext(
 			ctx,
-			"SELECT event_id, sequence_id, event_type, timestamp, schedule_event_id, attributes, visible_at FROM `history` WHERE instance_id = ? AND execution_id = ? AND sequence_id > ? ORDER BY sequence_id",
+			"SELECT h.event_id, h.sequence_id, h.event_type, h.timestamp, h.schedule_event_id, a.data, h.visible_at FROM `history` h JOIN `attributes` a ON h.event_id = a.event_id AND a.instance_id = h.instance_id AND a.execution_id = h.execution_id WHERE h.instance_id = ? AND h.execution_id = ? AND h.sequence_id > ? ORDER BY h.sequence_id",
 			instance.InstanceID,
 			instance.ExecutionID,
 			*lastSequenceID,
 		)
 	} else {
 		historyEvents, err = tx.QueryContext(
 			ctx,
-			"SELECT event_id, sequence_id, event_type, timestamp, schedule_event_id, attributes, visible_at FROM `history` WHERE instance_id = ? AND execution_id = ? ORDER BY sequence_id",
+			"SELECT h.event_id, h.sequence_id, h.event_type, h.timestamp, h.schedule_event_id, a.data, h.visible_at FROM `history` h JOIN `attributes` a ON h.event_id = a.event_id AND a.instance_id = h.instance_id AND a.execution_id = h.execution_id WHERE h.instance_id = ? AND h.execution_id = ? ORDER BY h.sequence_id",
 			instance.InstanceID,
 			instance.ExecutionID,
 		)
@@ -459,7 +463,7 @@ func (b *mysqlBackend) GetWorkflowTask(ctx context.Context) (*backend.WorkflowTa
 	// Get new events
 	events, err := tx.QueryContext(
 		ctx,
-		"SELECT event_id, sequence_id, event_type, timestamp, schedule_event_id, attributes, visible_at FROM `pending_events` WHERE instance_id = ? AND execution_id = ? AND (`visible_at` IS NULL OR `visible_at` <= ?) ORDER BY id",
+		"SELECT pe.event_id, pe.sequence_id, pe.event_type, pe.timestamp, pe.schedule_event_id, a.data, pe.visible_at FROM `pending_events` pe LEFT JOIN `attributes` a ON pe.instance_id = a.instance_id AND pe.execution_id = a.execution_id AND pe.event_id = a.event_id WHERE pe.instance_id = ? AND pe.execution_id = ? AND (pe.visible_at IS NULL OR pe.visible_at <= ?) ORDER BY pe.id",
 		instanceID,
 		executionID,
 		now,
@@ -503,15 +507,20 @@ func (b *mysqlBackend) GetWorkflowTask(ctx context.Context) (*backend.WorkflowTa
 	}
 
 	// Get most recent sequence id
-	row = tx.QueryRowContext(ctx, "SELECT sequence_id FROM `history` WHERE instance_id = ? AND execution_id = ? ORDER BY id DESC LIMIT 1", instanceID, executionID)
+	var lastSequenceID sql.NullInt64
+	row = tx.QueryRowContext(ctx, "SELECT MAX(sequence_id) FROM `history` WHERE instance_id = ? AND execution_id = ?", instanceID, executionID)
 	if err := row.Scan(
-		&t.LastSequenceID,
+		&lastSequenceID,
 	); err != nil {
 		if err != sql.ErrNoRows {
 			return nil, fmt.Errorf("getting most recent sequence id: %w", err)
 		}
 	}
 
+	if lastSequenceID.Valid {
+		t.LastSequenceID = lastSequenceID.Int64
+	}
+
 	if err := tx.Commit(); err != nil {
 		return nil, err
 	}
@@ -687,10 +696,11 @@ func (b *mysqlBackend) GetActivityTask(ctx context.Context) (*backend.ActivityTa
 	now := time.Now()
 	res := tx.QueryRowContext(
 		ctx,
-		`SELECT activities.id, activity_id, activities.instance_id, activities.execution_id,
-			event_type, timestamp, schedule_event_id, attributes, visible_at
-			FROM activities
-			WHERE activities.locked_until IS NULL OR activities.locked_until < ?
+		`SELECT a.id, a.activity_id, a.instance_id, a.execution_id,
+			a.event_type, a.timestamp, a.schedule_event_id, at.data, a.visible_at
+			FROM activities a
+			JOIN attributes at ON at.event_id = a.activity_id AND at.instance_id = a.instance_id AND at.execution_id = a.execution_id
+			WHERE a.locked_until IS NULL OR a.locked_until < ?
 			LIMIT 1
 			FOR UPDATE SKIP LOCKED`,
 		now,
@@ -815,22 +825,18 @@ func (b *mysqlBackend) ExtendActivityTask(ctx context.Context, activityID string
 }
 
 func scheduleActivity(ctx context.Context, tx *sql.Tx, instance *core.WorkflowInstance, event *history.Event) error {
-	a, err := history.SerializeAttributes(event.Attributes)
-	if err != nil {
-		return err
-	}
+	// Attributes are already persisted via the history, we do not need to add them again.
 
-	_, err = tx.ExecContext(
+	_, err := tx.ExecContext(
 		ctx,
 		`INSERT INTO activities
-			(activity_id, instance_id, execution_id, event_type, timestamp, schedule_event_id, attributes, visible_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
+			(activity_id, instance_id, execution_id, event_type, timestamp, schedule_event_id, visible_at) VALUES (?, ?, ?, ?, ?, ?, ?)`,
 		event.ID,
 		instance.InstanceID,
 		instance.ExecutionID,
 		event.Type,
 		event.Timestamp,
 		event.ScheduleEventID,
-		a,
 		event.VisibleAt,
 	)
 

backend/mysql/mysql_test.go

@@ -121,7 +121,7 @@ func (mb *mysqlBackend) GetFutureEvents(ctx context.Context) ([]*history.Event,
 	// There is no index on `visible_at`, but this is okay for test only usage.
 	futureEvents, err := tx.QueryContext(
 		ctx,
-		"SELECT event_id, sequence_id, instance_id, event_type, timestamp, schedule_event_id, attributes, visible_at FROM `pending_events` WHERE visible_at IS NOT NULL",
+		"SELECT pe.id, pe.sequence_id, pe.instance_id, pe.execution_id, pe.event_type, pe.timestamp, pe.schedule_event_id, pe.visible_at, a.data FROM `pending_events` pe JOIN `attributes` a ON a.id = pe.id AND a.instance_id = pe.instance_id AND a.execution_id = pe.execution_id WHERE pe.visible_at IS NOT NULL",
 	)
 	if err != nil {
 		return nil, fmt.Errorf("getting history: %w", err)

backend/sqlite/db/migrations/000002_add_attributes_table.down.sql

@@ -1,12 +1,9 @@
--- Move activity attributes to attributes table
 ALTER TABLE `activities` ADD COLUMN `attributes` BLOB NULL;
 UPDATE `activities` SET `attributes` = `attributes`.`data` FROM `attributes` WHERE `activities`.`id` = `attributes`.`id` AND `activities`.`instance_id` = `attributes`.`instance_id` AND `activities`.`execution_id` = `attributes`.`execution_id`;
 
--- Move history attributes to attributes table
 ALTER TABLE `history` ADD COLUMN `attributes` BLOB NULL;
 UPDATE `history` SET `attributes` = `attributes`.`data` FROM `attributes` WHERE `history`.`id` = `attributes`.`id` AND `history`.`instance_id` = `attributes`.`instance_id` AND `history`.`execution_id` = `attributes`.`execution_id`;
 
--- Move pending_events attributes to attributes table
 ALTER TABLE `pending_events` ADD COLUMN `attributes` BLOB NULL;
 UPDATE `pending_events` SET `attributes` = `attributes`.`data` FROM `attributes` WHERE `pending_events`.`id` = `attributes`.`id` AND `pending_events`.`instance_id` = `attributes`.`instance_id` AND `pending_events`.`execution_id` = `attributes`.`execution_id`;
 

backend/sqlite/db/migrations/000002_add_attributes_table.up.sql

@@ -3,17 +3,17 @@ CREATE TABLE IF NOT EXISTS `attributes` (
   `instance_id` TEXT NOT NULL,
   `execution_id` TEXT NOT NULL,
   `data` BLOB NOT NULL,
-  PRIMARY KEY(`id`, `instance_id`)
+  PRIMARY KEY(`id`, `instance_id`, `execution_id`)
 );
 
--- Move activity attributes to payloads table
+-- Move activity attributes to attributes table
 INSERT OR IGNORE INTO `attributes` (`id`, `instance_id`, `execution_id`, `data`) SELECT `id`, `instance_id`, `execution_id`, `attributes` FROM `activities`;
 ALTER TABLE `activities` DROP COLUMN `attributes`;
 
--- Move history attributes to payloads table
+-- Move history attributes to attributes table
 INSERT OR IGNORE INTO `attributes` (`id`, `instance_id`, `execution_id`, `data`) SELECT `id`, `instance_id`, `execution_id`, `attributes` FROM `history`;
 ALTER TABLE `history` DROP COLUMN `attributes`;
 
--- Move pending_events attributes to payloads table
+-- Move pending_events attributes to attributes table
 INSERT OR IGNORE INTO `attributes` (`id`, `instance_id`, `execution_id`, `data`) SELECT `id`, `instance_id`, `execution_id`, `attributes` FROM `pending_events`;
 ALTER TABLE `pending_events` DROP COLUMN `attributes`;