Refactor peristence layer to support inserting history tasks of new categories (#6671)

**Detailed Description**
Introduce data and data_encoding columns to the executions table of Cassandra. These field are used to replace other columns in `executions` table. We'll start by replacing the history task columns (`transfer`, `timer` and `replication`).

**Impact Analysis**
- **Backward Compatibility**: Yes. It only changes the schema, no data is written to the new columns in this PR.
- **Forward Compatibility**: Yes. It only changes the schema, no data is written to the new columns in this PR.

**Testing Plan**
- **Unit Tests**: Yes.
- **Persistence Tests**: No. The code implementation is not fully done in this PR.
- **Integration Tests**: No.
- **Compatibility Tests**: No.

**Rollout Plan**
- What is the rollout plan? No special rollout plan.
- Does the order of deployment matter? Doesn't matter.
- Is it safe to rollback? Does the order of rollback matter? Yes. The order doesn't matter.
- Is there a kill switch to mitigate the impact immediately? No.

<!-- Describe what has changed in this PR -->
**What changed?**
- Introduce `data` and `data_encoding` columns to the executions table of Cassandra. (NOTE: no data is written to Cassandra in this PR, it will be introduced in a separate PR)
- Refactor persistence data type to replace transfer tasks, timer tasks, and replication tasks with tasksByCategory map of tasks
- Remove cross cluster tasks logic from persistence layer

<!-- Tell your future self why have you made these changes -->
**Why?**
- The new columns will be used to store data in `executions` table. A migration is required later to migrate transfer, timer and replication tasks data from their corresponding columns to `data` and `data_encoding` columns. The migration will be done in several phases:
  1. Dual writes: data is written in both `transfer/timer/replication` column and `data`, `data_encoding` columns
  2. Read from old columns
  3. Read from new columns
Several other migrations will be initiated to migrate data from other columns of other types to these 2 columns.
- The refactoring is to make it easier to introduce new categories of history tasks.
- Cross cluster task is deprecated.
- This change should only affect cadence-history

<!-- How have you verified this change? Tested locally? Added a unit test? Checked in staging env? -->
**How did you test it?**
Unit tests

<!-- Assuming the worst case, what can be broken when deploying this change to production? -->
**Potential risks**

<!-- Is it notable for release? e.g. schema updates, configuration or data migration required? If so, please mention it, and also update CHANGELOG.md -->
**Release notes**

<!-- Is there any documentation updates should be made for config, https://cadenceworkflow.io/docs/operation-guide/setup/ ? If so, please open an PR in https://github.com/cadence-workflow/cadence-docs -->
**Documentation Changes**

Author: Shaddoll

common/persistence/data_manager_interfaces.go

@@ -174,6 +174,46 @@ const (
 	TaskListKindSticky
 )
 
+// HistoryTaskCategory represents various categories of history tasks
+type HistoryTaskCategory struct {
+	categoryType int
+	categoryID   int
+}
+
+func (c *HistoryTaskCategory) Type() int {
+	return c.categoryType
+}
+
+func (c *HistoryTaskCategory) ID() int {
+	return c.categoryID
+}
+
+const (
+	HistoryTaskCategoryTypeImmediate = iota + 1
+	HistoryTaskCategoryTypeScheduled
+)
+
+const (
+	HistoryTaskCategoryIDTransfer    = 1
+	HistoryTaskCategoryIDTimer       = 2
+	HistoryTaskCategoryIDReplication = 3
+)
+
+var (
+	HistoryTaskCategoryTransfer = HistoryTaskCategory{
+		categoryType: HistoryTaskCategoryTypeImmediate,
+		categoryID:   HistoryTaskCategoryIDTransfer,
+	}
+	HistoryTaskCategoryTimer = HistoryTaskCategory{
+		categoryType: HistoryTaskCategoryTypeScheduled,
+		categoryID:   HistoryTaskCategoryIDTimer,
+	}
+	HistoryTaskCategoryReplication = HistoryTaskCategory{
+		categoryType: HistoryTaskCategoryTypeImmediate,
+		categoryID:   HistoryTaskCategoryIDReplication,
+	}
+)
+
 // Transfer task types
 const (
 	TransferTaskTypeDecisionTask = iota
@@ -806,10 +846,7 @@ type (
 		NewBufferedEvents         []*types.HistoryEvent
 		ClearBufferedEvents       bool
 
-		TransferTasks     []Task
-		CrossClusterTasks []Task
-		ReplicationTasks  []Task
-		TimerTasks        []Task
+		TasksByCategory map[HistoryTaskCategory][]Task
 
 		WorkflowRequests []*WorkflowRequest
 
@@ -830,10 +867,7 @@ type (
 		SignalInfos         []*SignalInfo
 		SignalRequestedIDs  []string
 
-		TransferTasks     []Task
-		CrossClusterTasks []Task
-		ReplicationTasks  []Task
-		TimerTasks        []Task
+		TasksByCategory map[HistoryTaskCategory][]Task
 
 		WorkflowRequests []*WorkflowRequest
 
@@ -1326,10 +1360,7 @@ type (
 		DeleteSignalInfoCount        int
 		DeleteRequestCancelInfoCount int
 
-		TransferTasksCount    int
-		CrossClusterTaskCount int
-		TimerTasksCount       int
-		ReplicationTasksCount int
+		TaskCountByCategory map[HistoryTaskCategory]int
 	}
 
 	// UpdateWorkflowExecutionResponse is response for UpdateWorkflowExecutionRequest

common/persistence/data_store_interfaces.go

@@ -466,10 +466,7 @@ type (
 		NewBufferedEvents         *DataBlob
 		ClearBufferedEvents       bool
 
-		TransferTasks     []Task
-		CrossClusterTasks []Task
-		TimerTasks        []Task
-		ReplicationTasks  []Task
+		TasksByCategory map[HistoryTaskCategory][]Task
 
 		WorkflowRequests []*WorkflowRequest
 
@@ -493,10 +490,7 @@ type (
 		SignalInfos         []*SignalInfo
 		SignalRequestedIDs  []string
 
-		TransferTasks     []Task
-		CrossClusterTasks []Task
-		TimerTasks        []Task
-		ReplicationTasks  []Task
+		TasksByCategory map[HistoryTaskCategory][]Task
 
 		WorkflowRequests []*WorkflowRequest
 

common/persistence/execution_manager.go

@@ -676,10 +676,7 @@ func (m *executionManagerImpl) SerializeWorkflowMutation(
 		NewBufferedEvents:         serializedNewBufferedEvents,
 		ClearBufferedEvents:       input.ClearBufferedEvents,
 
-		TransferTasks:     input.TransferTasks,
-		CrossClusterTasks: input.CrossClusterTasks,
-		ReplicationTasks:  input.ReplicationTasks,
-		TimerTasks:        input.TimerTasks,
+		TasksByCategory: input.TasksByCategory,
 
 		WorkflowRequests: input.WorkflowRequests,
 
@@ -742,10 +739,7 @@ func (m *executionManagerImpl) SerializeWorkflowSnapshot(
 		SignalInfos:         input.SignalInfos,
 		SignalRequestedIDs:  input.SignalRequestedIDs,
 
-		TransferTasks:     input.TransferTasks,
-		CrossClusterTasks: input.CrossClusterTasks,
-		ReplicationTasks:  input.ReplicationTasks,
-		TimerTasks:        input.TimerTasks,
+		TasksByCategory: input.TasksByCategory,
 
 		WorkflowRequests: input.WorkflowRequests,
 

common/persistence/execution_manager_test.go

@@ -426,14 +426,15 @@ func TestExecutionManager_UpdateWorkflowExecution(t *testing.T) {
 	res, err := manager.UpdateWorkflowExecution(context.Background(), request)
 	assert.NoError(t, err)
 	stats := &MutableStateUpdateSessionStats{
-		MutableStateSize:  90,
-		ExecutionInfoSize: 40,
-		ActivityInfoSize:  20,
-		TimerInfoSize:     10,
-		ChildInfoSize:     20,
-		ActivityInfoCount: 1,
-		TimerInfoCount:    2,
-		ChildInfoCount:    1,
+		MutableStateSize:    90,
+		ExecutionInfoSize:   40,
+		ActivityInfoSize:    20,
+		TimerInfoSize:       10,
+		ChildInfoSize:       20,
+		ActivityInfoCount:   1,
+		TimerInfoCount:      1,
+		ChildInfoCount:      1,
+		TaskCountByCategory: map[HistoryTaskCategory]int{},
 	}
 	assert.Equal(t, stats, res.MutableStateUpdateSessionStats)
 }
@@ -1016,14 +1017,15 @@ func TestCreateWorkflowExecution(t *testing.T) {
 			checkRes: func(t *testing.T, response *CreateWorkflowExecutionResponse, err error) {
 				assert.Equal(t, &CreateWorkflowExecutionResponse{
 					MutableStateUpdateSessionStats: &MutableStateUpdateSessionStats{
-						MutableStateSize:  91,
-						ExecutionInfoSize: 20,
-						ActivityInfoSize:  29,
-						TimerInfoSize:     22,
-						ChildInfoSize:     20,
-						ActivityInfoCount: 1,
-						TimerInfoCount:    2,
-						ChildInfoCount:    1,
+						MutableStateSize:    91,
+						ExecutionInfoSize:   20,
+						ActivityInfoSize:    29,
+						TimerInfoSize:       22,
+						ChildInfoSize:       20,
+						ActivityInfoCount:   1,
+						TimerInfoCount:      2,
+						ChildInfoCount:      1,
+						TaskCountByCategory: map[HistoryTaskCategory]int{},
 					},
 				}, response)
 			},
@@ -1118,14 +1120,15 @@ func TestConflictResolveWorkflowExecution(t *testing.T) {
 				assert.NoError(t, err)
 				assert.Equal(t, &ConflictResolveWorkflowExecutionResponse{
 					MutableStateUpdateSessionStats: &MutableStateUpdateSessionStats{
-						MutableStateSize:  91,
-						ExecutionInfoSize: 20,
-						ActivityInfoSize:  29,
-						TimerInfoSize:     22,
-						ChildInfoSize:     20,
-						ActivityInfoCount: 1,
-						TimerInfoCount:    2,
-						ChildInfoCount:    1,
+						MutableStateSize:    91,
+						ExecutionInfoSize:   20,
+						ActivityInfoSize:    29,
+						TimerInfoSize:       22,
+						ChildInfoSize:       20,
+						ActivityInfoCount:   1,
+						TimerInfoCount:      2,
+						ChildInfoCount:      1,
+						TaskCountByCategory: map[HistoryTaskCategory]int{},
 					},
 				}, response)
 			},
@@ -1184,14 +1187,15 @@ func TestConflictResolveWorkflowExecution(t *testing.T) {
 				assert.NoError(t, err)
 				assert.Equal(t, &ConflictResolveWorkflowExecutionResponse{
 					MutableStateUpdateSessionStats: &MutableStateUpdateSessionStats{
-						MutableStateSize:  161,
-						ExecutionInfoSize: 40,
-						ActivityInfoSize:  49,
-						TimerInfoSize:     32,
-						ChildInfoSize:     40,
-						ActivityInfoCount: 2,
-						TimerInfoCount:    6,
-						ChildInfoCount:    2,
+						MutableStateSize:    161,
+						ExecutionInfoSize:   40,
+						ActivityInfoSize:    49,
+						TimerInfoSize:       32,
+						ChildInfoSize:       40,
+						ActivityInfoCount:   2,
+						TimerInfoCount:      3,
+						ChildInfoCount:      2,
+						TaskCountByCategory: map[HistoryTaskCategory]int{},
 					},
 				}, response)
 			},

common/persistence/nosql/nosql_execution_store.go

@@ -90,10 +90,11 @@ func (d *nosqlExecutionStore) CreateWorkflowExecution(
 		return nil, err
 	}
 
-	transferTasks, crossClusterTasks, replicationTasks, timerTasks, err := d.prepareNoSQLTasksForWorkflowTxn(
+	tasksByCategory := map[persistence.HistoryTaskCategory][]*nosqlplugin.HistoryMigrationTask{}
+	err = d.prepareNoSQLTasksForWorkflowTxn(
 		domainID, workflowID, runID,
-		newWorkflow.TransferTasks, newWorkflow.CrossClusterTasks, newWorkflow.ReplicationTasks, newWorkflow.TimerTasks,
-		nil, nil, nil, nil,
+		newWorkflow.TasksByCategory,
+		tasksByCategory,
 	)
 	if err != nil {
 		return nil, err
@@ -115,7 +116,7 @@ func (d *nosqlExecutionStore) CreateWorkflowExecution(
 		ctx,
 		workflowRequestsWriteRequest,
 		currentWorkflowWriteReq, workflowExecutionWriteReq,
-		transferTasks, crossClusterTasks, replicationTasks, timerTasks,
+		tasksByCategory,
 		shardCondition,
 	)
 	if err != nil {
@@ -287,21 +288,18 @@ func (d *nosqlExecutionStore) UpdateWorkflowExecution(
 	}
 
 	var mutateExecution, insertExecution *nosqlplugin.WorkflowExecutionRequest
-	var nosqlTransferTasks []*nosqlplugin.TransferTask
-	var nosqlCrossClusterTasks []*nosqlplugin.CrossClusterTask
-	var nosqlReplicationTasks []*nosqlplugin.ReplicationTask
-	var nosqlTimerTasks []*nosqlplugin.TimerTask
 	var workflowRequests []*nosqlplugin.WorkflowRequestRow
+	tasksByCategory := map[persistence.HistoryTaskCategory][]*nosqlplugin.HistoryMigrationTask{}
 
 	// 1. current
 	mutateExecution, err = d.prepareUpdateWorkflowExecutionRequestWithMapsAndEventBuffer(&updateWorkflow)
 	if err != nil {
 		return err
 	}
-	nosqlTransferTasks, nosqlCrossClusterTasks, nosqlReplicationTasks, nosqlTimerTasks, err = d.prepareNoSQLTasksForWorkflowTxn(
+	err = d.prepareNoSQLTasksForWorkflowTxn(
 		domainID, workflowID, updateWorkflow.ExecutionInfo.RunID,
-		updateWorkflow.TransferTasks, updateWorkflow.CrossClusterTasks, updateWorkflow.ReplicationTasks, updateWorkflow.TimerTasks,
-		nosqlTransferTasks, nosqlCrossClusterTasks, nosqlReplicationTasks, nosqlTimerTasks,
+		updateWorkflow.TasksByCategory,
+		tasksByCategory,
 	)
 	if err != nil {
 		return err
@@ -315,10 +313,10 @@ func (d *nosqlExecutionStore) UpdateWorkflowExecution(
 			return err
 		}
 
-		nosqlTransferTasks, nosqlCrossClusterTasks, nosqlReplicationTasks, nosqlTimerTasks, err = d.prepareNoSQLTasksForWorkflowTxn(
+		err = d.prepareNoSQLTasksForWorkflowTxn(
 			domainID, workflowID, newWorkflow.ExecutionInfo.RunID,
-			newWorkflow.TransferTasks, newWorkflow.CrossClusterTasks, newWorkflow.ReplicationTasks, newWorkflow.TimerTasks,
-			nosqlTransferTasks, nosqlCrossClusterTasks, nosqlReplicationTasks, nosqlTimerTasks,
+			newWorkflow.TasksByCategory,
+			tasksByCategory,
 		)
 		if err != nil {
 			return err
@@ -339,7 +337,7 @@ func (d *nosqlExecutionStore) UpdateWorkflowExecution(
 	err = d.db.UpdateWorkflowExecutionWithTasks(
 		ctx, workflowRequestsWriteRequest, currentWorkflowWriteReq,
 		mutateExecution, insertExecution, nil, // no workflow to reset here
-		nosqlTransferTasks, nosqlCrossClusterTasks, nosqlReplicationTasks, nosqlTimerTasks,
+		tasksByCategory,
 		shardCondition)
 
 	return d.processUpdateWorkflowResult(err, request.RangeID)
@@ -422,22 +420,19 @@ func (d *nosqlExecutionStore) ConflictResolveWorkflowExecution(
 	}
 
 	var mutateExecution, insertExecution, resetExecution *nosqlplugin.WorkflowExecutionRequest
-	var nosqlTransferTasks []*nosqlplugin.TransferTask
-	var nosqlCrossClusterTasks []*nosqlplugin.CrossClusterTask
-	var nosqlReplicationTasks []*nosqlplugin.ReplicationTask
-	var nosqlTimerTasks []*nosqlplugin.TimerTask
 	var workflowRequests []*nosqlplugin.WorkflowRequestRow
+	tasksByCategory := map[persistence.HistoryTaskCategory][]*nosqlplugin.HistoryMigrationTask{}
 
 	// 1. current
 	if currentWorkflow != nil {
 		mutateExecution, err = d.prepareUpdateWorkflowExecutionRequestWithMapsAndEventBuffer(currentWorkflow)
 		if err != nil {
 			return err
 		}
-		nosqlTransferTasks, nosqlCrossClusterTasks, nosqlReplicationTasks, nosqlTimerTasks, err = d.prepareNoSQLTasksForWorkflowTxn(
+		err = d.prepareNoSQLTasksForWorkflowTxn(
 			domainID, workflowID, currentWorkflow.ExecutionInfo.RunID,
-			currentWorkflow.TransferTasks, currentWorkflow.CrossClusterTasks, currentWorkflow.ReplicationTasks, currentWorkflow.TimerTasks,
-			nosqlTransferTasks, nosqlCrossClusterTasks, nosqlReplicationTasks, nosqlTimerTasks,
+			currentWorkflow.TasksByCategory,
+			tasksByCategory,
 		)
 		if err != nil {
 			return err
@@ -450,10 +445,10 @@ func (d *nosqlExecutionStore) ConflictResolveWorkflowExecution(
 	if err != nil {
 		return err
 	}
-	nosqlTransferTasks, nosqlCrossClusterTasks, nosqlReplicationTasks, nosqlTimerTasks, err = d.prepareNoSQLTasksForWorkflowTxn(
+	err = d.prepareNoSQLTasksForWorkflowTxn(
 		domainID, workflowID, resetWorkflow.ExecutionInfo.RunID,
-		resetWorkflow.TransferTasks, resetWorkflow.CrossClusterTasks, resetWorkflow.ReplicationTasks, resetWorkflow.TimerTasks,
-		nosqlTransferTasks, nosqlCrossClusterTasks, nosqlReplicationTasks, nosqlTimerTasks,
+		resetWorkflow.TasksByCategory,
+		tasksByCategory,
 	)
 	if err != nil {
 		return err
@@ -467,10 +462,10 @@ func (d *nosqlExecutionStore) ConflictResolveWorkflowExecution(
 			return err
 		}
 
-		nosqlTransferTasks, nosqlCrossClusterTasks, nosqlReplicationTasks, nosqlTimerTasks, err = d.prepareNoSQLTasksForWorkflowTxn(
+		err = d.prepareNoSQLTasksForWorkflowTxn(
 			domainID, workflowID, newWorkflow.ExecutionInfo.RunID,
-			newWorkflow.TransferTasks, newWorkflow.CrossClusterTasks, newWorkflow.ReplicationTasks, newWorkflow.TimerTasks,
-			nosqlTransferTasks, nosqlCrossClusterTasks, nosqlReplicationTasks, nosqlTimerTasks,
+			newWorkflow.TasksByCategory,
+			tasksByCategory,
 		)
 		if err != nil {
 			return err
@@ -491,7 +486,7 @@ func (d *nosqlExecutionStore) ConflictResolveWorkflowExecution(
 	err = d.db.UpdateWorkflowExecutionWithTasks(
 		ctx, workflowRequestsWriteRequest, currentWorkflowWriteReq,
 		mutateExecution, insertExecution, resetExecution,
-		nosqlTransferTasks, nosqlCrossClusterTasks, nosqlReplicationTasks, nosqlTimerTasks,
+		tasksByCategory,
 		shardCondition)
 	return d.processUpdateWorkflowResult(err, request.RangeID)
 }
@@ -759,7 +754,10 @@ func (d *nosqlExecutionStore) PutReplicationTaskToDLQ(
 	ctx context.Context,
 	request *persistence.InternalPutReplicationTaskToDLQRequest,
 ) error {
-	err := d.db.InsertReplicationDLQTask(ctx, d.shardID, request.SourceClusterName, *request.TaskInfo)
+	err := d.db.InsertReplicationDLQTask(ctx, d.shardID, request.SourceClusterName, &nosqlplugin.HistoryMigrationTask{
+		Replication: request.TaskInfo,
+		Task:        nil, // TODO: encode task infor into datablob
+	})
 	if err != nil {
 		return convertCommonErrors(d.db, "PutReplicationTaskToDLQ", err)
 	}
@@ -829,15 +827,20 @@ func (d *nosqlExecutionStore) CreateFailoverMarkerTasks(
 	request *persistence.CreateFailoverMarkersRequest,
 ) error {
 
-	var nosqlTasks []*nosqlplugin.ReplicationTask
+	var nosqlTasks []*nosqlplugin.HistoryMigrationTask
 	for _, task := range request.Markers {
 		ts := []persistence.Task{task}
 
 		tasks, err := d.prepareReplicationTasksForWorkflowTxn(task.DomainID, rowTypeReplicationWorkflowID, rowTypeReplicationRunID, ts)
 		if err != nil {
 			return err
 		}
-		nosqlTasks = append(nosqlTasks, tasks...)
+		for _, task := range tasks {
+			nosqlTasks = append(nosqlTasks, &nosqlplugin.HistoryMigrationTask{
+				Replication: task,
+				Task:        nil, // TODO: encode replication task into datablob
+			})
+		}
 	}
 
 	err := d.db.InsertReplicationTask(ctx, nosqlTasks, nosqlplugin.ShardCondition{