Remove Proto Double Serialization, VersionHistories Column Cassandra (#624)

Author: shawnhathaway

api/persistenceblobs/v1/message.pb.go

@@ -172,12 +172,6 @@ workflow_state = ? ` +
 		`VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?` +
 		`, ?, ?, ?, ?, ?) IF NOT EXISTS `
 
-	templateCreateWorkflowExecutionWithVersionHistoriesQuery = `INSERT INTO executions (` +
-		`shard_id, namespace_id, workflow_id, run_id, type, ` +
-		`execution, execution_encoding, execution_state, execution_state_encoding, next_event_id, ` +
-		`visibility_ts, task_id, version_histories, version_histories_encoding, checksum, checksum_encoding) ` +
-		`VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) IF NOT EXISTS `
-
 	templateCreateTransferTaskQuery = `INSERT INTO executions (` +
 		`shard_id, type, namespace_id, workflow_id, run_id, transfer, transfer_encoding, visibility_ts, task_id) ` +
 		`VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)`
@@ -203,7 +197,7 @@ workflow_state = ? ` +
 
 	templateGetWorkflowExecutionQuery = `SELECT execution, execution_encoding, execution_state, execution_state_encoding, next_event_id, replication_metadata, replication_metadata_encoding, activity_map, activity_map_encoding, timer_map, timer_map_encoding, ` +
 		`child_executions_map, child_executions_map_encoding, request_cancel_map, request_cancel_map_encoding, signal_map, signal_map_encoding, signal_requested, buffered_events_list, ` +
-		`version_histories, version_histories_encoding, checksum, checksum_encoding ` +
+		`checksum, checksum_encoding ` +
 		`FROM executions ` +
 		`WHERE shard_id = ? ` +
 		`and type = ? ` +
@@ -275,25 +269,6 @@ workflow_state = ? ` +
 		`and task_id = ? ` +
 		`IF next_event_id = ? `
 
-	templateUpdateWorkflowExecutionWithVersionHistoriesQuery = `UPDATE executions ` +
-		`SET execution = ?` +
-		`, execution_encoding = ?` +
-		`, execution_state = ? ` +
-		`, execution_state_encoding = ? ` +
-		`, next_event_id = ? ` +
-		`, version_histories = ? ` +
-		`, version_histories_encoding = ? ` +
-		`, checksum = ? ` +
-		`, checksum_encoding = ? ` +
-		`WHERE shard_id = ? ` +
-		`and type = ? ` +
-		`and namespace_id = ? ` +
-		`and workflow_id = ? ` +
-		`and run_id = ? ` +
-		`and visibility_ts = ? ` +
-		`and task_id = ? ` +
-		`IF next_event_id = ? `
-
 	templateUpdateActivityInfoQuery = `UPDATE executions ` +
 		`SET activity_map[ ? ] = ?, activity_map_encoding = ? ` +
 		`WHERE shard_id = ? ` +
@@ -1109,16 +1084,11 @@ func (d *cassandraPersistence) GetWorkflowExecution(request *p.GetWorkflowExecut
 		return nil, convertCommonErrors("GetWorkflowExecution", err)
 	}
 
-	info, replicationState, err := workflowExecutionFromRow(result)
+	state, err := mutableStateFromRow(result)
 	if err != nil {
 		return nil, serviceerror.NewInternal(fmt.Sprintf("GetWorkflowExecution operation failed. Error: %v", err))
 	}
 
-	state := &p.InternalWorkflowMutableState{
-		ExecutionInfo:    info,
-		ReplicationState: replicationState,
-		VersionHistories: p.NewDataBlob(result["version_histories"].([]byte), common.EncodingType(result["version_histories_encoding"].(string))),
-	}
 
 	if state.VersionHistories != nil && state.ReplicationState != nil {
 		return nil, serviceerror.NewInternal(fmt.Sprintf("GetWorkflowExecution operation failed. VersionHistories and ReplicationState both are set."))
@@ -1929,7 +1899,8 @@ func (d *cassandraPersistence) ListConcreteExecutions(
 			continue
 		}
 		if _, ok := result["execution"]; ok {
-			wfInfo, _, _ := workflowExecutionFromRow(result)
+			state, _ := mutableStateFromRow(result)
+			wfInfo := state.ExecutionInfo
 			response.ExecutionInfos = append(response.ExecutionInfos, wfInfo)
 		}
 		result = make(map[string]interface{})
@@ -2806,30 +2777,30 @@ func (d *cassandraPersistence) RangeDeleteReplicationTaskFromDLQ(
 	return nil
 }
 
-func workflowExecutionFromRow(result map[string]interface{}) (*p.InternalWorkflowExecutionInfo, *persistenceblobs.ReplicationState, error) {
+func mutableStateFromRow(result map[string]interface{}) (*p.InternalWorkflowMutableState, error) {
 	eiBytes, ok := result["execution"].([]byte)
 	if !ok {
-		return nil, nil, newPersistedTypeMismatchError("execution", "", eiBytes, result)
+		return nil, newPersistedTypeMismatchError("execution", "", eiBytes, result)
 	}
 
 	eiEncoding, ok := result["execution_encoding"].(string)
 	if !ok {
-		return nil, nil, newPersistedTypeMismatchError("execution_encoding", "", eiEncoding, result)
+		return nil, newPersistedTypeMismatchError("execution_encoding", "", eiEncoding, result)
 	}
 
 	protoInfo, err := serialization.WorkflowExecutionInfoFromBlob(eiBytes, eiEncoding)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
 	nextEventID, ok := result["next_event_id"].(int64)
 	if !ok {
-		return nil, nil, newPersistedTypeMismatchError("next_event_id", "", nextEventID, result)
+		return nil, newPersistedTypeMismatchError("next_event_id", "", nextEventID, result)
 	}
 
 	protoState, err := protoExecutionStateFromRow(result)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
 	info := p.ProtoWorkflowExecutionToPartialInternalExecution(protoInfo, protoState, nextEventID)
@@ -2838,13 +2809,18 @@ func workflowExecutionFromRow(result map[string]interface{}) (*p.InternalWorkflo
 	if protoInfo.ReplicationData != nil {
 		protoReplVersions, err := ProtoReplicationVersionsFromResultMap(result)
 		if err != nil {
-			return nil, nil, err
+			return nil, err
 		}
 
 		state = ReplicationStateFromProtos(protoInfo, protoReplVersions)
 	}
 
-	return info, state, nil
+	mutableState := &p.InternalWorkflowMutableState{
+		ExecutionInfo:    info,
+		ReplicationState: state,
+		VersionHistories: protoInfo.VersionHistories,
+	}
+	return mutableState, nil
 }
 
 func ProtoReplicationVersionsFromResultMap(result map[string]interface{}) (*persistenceblobs.ReplicationVersions, error) {

common/persistence/cassandra/cassandraPersistence.go

@@ -34,6 +34,7 @@ import (
 	"go.temporal.io/api/serviceerror"
 
 	enumsspb "go.temporal.io/server/api/enums/v1"
+	"go.temporal.io/server/api/history/v1"
 	"go.temporal.io/server/api/persistenceblobs/v1"
 	replicationspb "go.temporal.io/server/api/replication/v1"
 	"go.temporal.io/server/common"
@@ -426,7 +427,7 @@ func createExecution(
 	shardID int,
 	executionInfo *p.InternalWorkflowExecutionInfo,
 	replicationState *persistenceblobs.ReplicationState,
-	versionHistories *serialization.DataBlob,
+	versionHistories *history.VersionHistories,
 	checksum checksum.Checksum,
 	cqlNowTimestampMillis int64,
 	startVersion int64,
@@ -487,8 +488,7 @@ func createExecution(
 			checksumDatablob.Encoding.String())
 	} else if versionHistories != nil {
 		// TODO also need to set the start / current / last write version
-		versionHistoriesData, versionHistoriesEncoding := p.FromDataBlob(versionHistories)
-		batch.Query(templateCreateWorkflowExecutionWithVersionHistoriesQuery,
+		batch.Query(templateCreateWorkflowExecutionQuery,
 			shardID,
 			namespaceID,
 			workflowID,
@@ -501,8 +501,6 @@ func createExecution(
 			executionInfo.NextEventID,
 			defaultVisibilityTimestamp,
 			rowTypeExecutionTaskID,
-			versionHistoriesData,
-			versionHistoriesEncoding,
 			checksumDatablob.Data,
 			checksumDatablob.Encoding.String())
 	} else if replicationState != nil {
@@ -538,7 +536,7 @@ func updateExecution(
 	shardID int,
 	executionInfo *p.InternalWorkflowExecutionInfo,
 	replicationState *persistenceblobs.ReplicationState,
-	versionHistories *serialization.DataBlob,
+	versionHistories *history.VersionHistories,
 	cqlNowTimestampMillis int64,
 	condition int64,
 	checksum checksum.Checksum,
@@ -600,15 +598,12 @@ func updateExecution(
 			condition)
 	} else if versionHistories != nil {
 		// TODO also need to set the start / current / last write version
-		versionHistoriesData, versionHistoriesEncoding := p.FromDataBlob(versionHistories)
-		batch.Query(templateUpdateWorkflowExecutionWithVersionHistoriesQuery,
+		batch.Query(templateUpdateWorkflowExecutionQuery,
 			executionDatablob.Data,
 			executionDatablob.Encoding.String(),
 			executionStateDatablob.Data,
 			executionStateDatablob.Encoding.String(),
 			executionInfo.NextEventID,
-			versionHistoriesData,
-			versionHistoriesEncoding,
 			checksumDatablob.Data,
 			checksumDatablob.Encoding.String(),
 			shardID,

common/persistence/cassandra/cassandraPersistenceUtil.go

@@ -27,6 +27,7 @@ package persistence
 import (
 	historypb "go.temporal.io/api/history/v1"
 	"go.temporal.io/api/serviceerror"
+	workflowpb "go.temporal.io/api/workflow/v1"
 
 	"go.temporal.io/server/api/persistenceblobs/v1"
 	"go.temporal.io/server/common"
@@ -87,6 +88,7 @@ func (m *executionManagerImpl) GetWorkflowExecution(
 			ReplicationState:    response.State.ReplicationState,
 			Checksum:            response.State.Checksum,
 			ChildExecutionInfos: response.State.ChildExecutionInfos,
+			VersionHistories:    NewVersionHistoriesFromProto(response.State.VersionHistories),
 		},
 	}
 
@@ -98,11 +100,7 @@ func (m *executionManagerImpl) GetWorkflowExecution(
 	if err != nil {
 		return nil, err
 	}
-	versionHistories, err := m.DeserializeVersionHistories(response.State.VersionHistories)
-	if err != nil {
-		return nil, err
-	}
-	newResponse.State.VersionHistories = versionHistories
+
 	newResponse.MutableStateStats = m.statsComputer.computeMutableStateStats(response)
 
 	return newResponse, nil
@@ -111,20 +109,8 @@ func (m *executionManagerImpl) GetWorkflowExecution(
 func (m *executionManagerImpl) DeserializeExecutionInfo(
 	info *InternalWorkflowExecutionInfo,
 ) (*WorkflowExecutionInfo, *persistenceblobs.ExecutionStats, error) {
-
-	completionEvent, err := m.serializer.DeserializeEvent(info.CompletionEvent)
-	if err != nil {
-		return nil, nil, err
-	}
-
-	autoResetPoints, err := m.serializer.DeserializeResetPoints(info.AutoResetPoints)
-	if err != nil {
-		return nil, nil, err
-	}
-
 	newInfo := &WorkflowExecutionInfo{
-		CompletionEvent: completionEvent,
-
+		CompletionEvent:                        info.CompletionEvent,
 		NamespaceID:                            info.NamespaceID,
 		WorkflowID:                             info.WorkflowID,
 		RunID:                                  info.RunID,
@@ -174,10 +160,15 @@ func (m *executionManagerImpl) DeserializeExecutionInfo(
 		NonRetryableErrorTypes:                 info.NonRetryableErrorTypes,
 		BranchToken:                            info.BranchToken,
 		CronSchedule:                           info.CronSchedule,
-		AutoResetPoints:                        autoResetPoints,
+		AutoResetPoints:                        info.AutoResetPoints,
 		SearchAttributes:                       info.SearchAttributes,
 		Memo:                                   info.Memo,
 	}
+
+	if newInfo.AutoResetPoints == nil {
+		newInfo.AutoResetPoints = &workflowpb.ResetPoints{}
+	}
+
 	newStats := &persistenceblobs.ExecutionStats{
 		HistorySize: info.HistorySize,
 	}
@@ -238,15 +229,6 @@ func (m *executionManagerImpl) SerializeExecutionInfo(
 	if info == nil {
 		return &InternalWorkflowExecutionInfo{}, nil
 	}
-	completionEvent, err := m.serializer.SerializeEvent(info.CompletionEvent, encoding)
-	if err != nil {
-		return nil, err
-	}
-
-	resetPoints, err := m.serializer.SerializeResetPoints(info.AutoResetPoints, encoding)
-	if err != nil {
-		return nil, err
-	}
 
 	return &InternalWorkflowExecutionInfo{
 		NamespaceID:                            info.NamespaceID,
@@ -257,7 +239,7 @@ func (m *executionManagerImpl) SerializeExecutionInfo(
 		ParentRunID:                            info.ParentRunID,
 		InitiatedID:                            info.InitiatedID,
 		CompletionEventBatchID:                 info.CompletionEventBatchID,
-		CompletionEvent:                        completionEvent,
+		CompletionEvent:                        info.CompletionEvent,
 		TaskQueue:                              info.TaskQueue,
 		WorkflowTypeName:                       info.WorkflowTypeName,
 		WorkflowRunTimeout:                     info.WorkflowRunTimeout,
@@ -289,7 +271,7 @@ func (m *executionManagerImpl) SerializeExecutionInfo(
 		ClientLibraryVersion:                   info.ClientLibraryVersion,
 		ClientFeatureVersion:                   info.ClientFeatureVersion,
 		ClientImpl:                             info.ClientImpl,
-		AutoResetPoints:                        resetPoints,
+		AutoResetPoints:                        info.AutoResetPoints,
 		Attempt:                                info.Attempt,
 		HasRetryPolicy:                         info.HasRetryPolicy,
 		InitialInterval:                        info.InitialInterval,
@@ -423,10 +405,7 @@ func (m *executionManagerImpl) SerializeWorkflowMutation(
 	if err != nil {
 		return nil, err
 	}
-	serializedVersionHistories, err := m.SerializeVersionHistories(input.VersionHistories, encoding)
-	if err != nil {
-		return nil, err
-	}
+
 	var serializedNewBufferedEvents *serialization.DataBlob
 	if input.NewBufferedEvents != nil {
 		serializedNewBufferedEvents, err = m.serializer.SerializeBatchEvents(input.NewBufferedEvents, encoding)
@@ -447,7 +426,7 @@ func (m *executionManagerImpl) SerializeWorkflowMutation(
 	return &InternalWorkflowMutation{
 		ExecutionInfo:    serializedExecutionInfo,
 		ReplicationState: input.ReplicationState,
-		VersionHistories: serializedVersionHistories,
+		VersionHistories: input.VersionHistories.ToProto(),
 		StartVersion:     startVersion,
 		LastWriteVersion: lastWriteVersion,
 
@@ -488,10 +467,6 @@ func (m *executionManagerImpl) SerializeWorkflowSnapshot(
 	if err != nil {
 		return nil, err
 	}
-	serializedVersionHistories, err := m.SerializeVersionHistories(input.VersionHistories, encoding)
-	if err != nil {
-		return nil, err
-	}
 
 	startVersion, err := getStartVersion(input.VersionHistories, input.ReplicationState)
 	if err != nil {
@@ -505,7 +480,7 @@ func (m *executionManagerImpl) SerializeWorkflowSnapshot(
 	return &InternalWorkflowSnapshot{
 		ExecutionInfo:    serializedExecutionInfo,
 		ReplicationState: input.ReplicationState,
-		VersionHistories: serializedVersionHistories,
+		VersionHistories: input.VersionHistories.ToProto(),
 		StartVersion:     startVersion,
 		LastWriteVersion: lastWriteVersion,