Add SQL specific tests for replication tasks (#859)

* Some minor code structure change
* Add replication tasks tests
* Add replication DLQ tasks tests

Author: wxing1292

common/persistence/sql/sqlExecutionManager.go

@@ -763,7 +763,7 @@ func (m *sqlExecutionManager) RangeCompleteTransferTask(
 }
 
 func (m *sqlExecutionManager) GetReplicationTask(request *persistence.GetReplicationTaskRequest) (*persistence.GetReplicationTaskResponse, error) {
-	rows, err := m.db.SelectFromReplicationTasks(&sqlplugin.ReplicationTasksFilter{ShardID: request.ShardID, TaskID: request.TaskID})
+	rows, err := m.db.SelectFromReplicationTasks(sqlplugin.ReplicationTasksFilter{ShardID: request.ShardID, TaskID: request.TaskID})
 	if err != nil {
 		if err == sql.ErrNoRows {
 			return nil, serviceerror.NewNotFound(fmt.Sprintf("GetReplicationTask operation failed. Task with ID %v not found. Error: %v", request.TaskID, err))
@@ -795,8 +795,8 @@ func (m *sqlExecutionManager) GetReplicationTasks(
 		return nil, err
 	}
 
-	rows, err := m.db.SelectFromReplicationTasks(
-		&sqlplugin.ReplicationTasksFilter{
+	rows, err := m.db.RangeSelectFromReplicationTasks(
+		sqlplugin.ReplicationTasksRangeFilter{
 			ShardID:   m.shardID,
 			MinTaskID: readLevel,
 			MaxTaskID: maxReadLevelInclusive,
@@ -854,11 +854,39 @@ func (m *sqlExecutionManager) populateGetReplicationTasksResponse(
 	}, nil
 }
 
+func (m *sqlExecutionManager) populateGetReplicationDLQTasksResponse(
+	rows []sqlplugin.ReplicationDLQTasksRow,
+	requestMaxReadLevel int64,
+) (*p.GetReplicationTasksResponse, error) {
+	if len(rows) == 0 {
+		return &p.GetReplicationTasksResponse{}, nil
+	}
+
+	var tasks = make([]*persistenceblobs.ReplicationTaskInfo, len(rows))
+	for i, row := range rows {
+		info, err := serialization.ReplicationTaskInfoFromBlob(row.Data, row.DataEncoding)
+		if err != nil {
+			return nil, err
+		}
+
+		tasks[i] = info
+	}
+	var nextPageToken []byte
+	lastTaskID := rows[len(rows)-1].TaskID
+	if lastTaskID < requestMaxReadLevel {
+		nextPageToken = serializePageToken(lastTaskID)
+	}
+	return &p.GetReplicationTasksResponse{
+		Tasks:         tasks,
+		NextPageToken: nextPageToken,
+	}, nil
+}
+
 func (m *sqlExecutionManager) CompleteReplicationTask(
 	request *p.CompleteReplicationTaskRequest,
 ) error {
 
-	if _, err := m.db.DeleteFromReplicationTasks(&sqlplugin.ReplicationTasksFilter{
+	if _, err := m.db.DeleteFromReplicationTasks(sqlplugin.ReplicationTasksFilter{
 		ShardID: m.shardID,
 		TaskID:  request.TaskID,
 	}); err != nil {
@@ -871,9 +899,10 @@ func (m *sqlExecutionManager) RangeCompleteReplicationTask(
 	request *p.RangeCompleteReplicationTaskRequest,
 ) error {
 
-	if _, err := m.db.RangeDeleteFromReplicationTasks(&sqlplugin.ReplicationTasksFilter{
-		ShardID: m.shardID,
-		TaskID:  request.InclusiveEndTaskID,
+	if _, err := m.db.RangeDeleteFromReplicationTasks(sqlplugin.ReplicationTasksRangeFilter{
+		ShardID:   m.shardID,
+		MinTaskID: 0,
+		MaxTaskID: request.InclusiveEndTaskID,
 	}); err != nil {
 		return serviceerror.NewInternal(fmt.Sprintf("RangeCompleteReplicationTask operation failed. Error: %v", err))
 	}
@@ -889,20 +918,17 @@ func (m *sqlExecutionManager) GetReplicationTasksFromDLQ(
 		return nil, err
 	}
 
-	filter := sqlplugin.ReplicationTasksFilter{
-		ShardID:   m.shardID,
-		MinTaskID: readLevel,
-		MaxTaskID: maxReadLevelInclusive,
-		PageSize:  request.BatchSize,
-	}
-	rows, err := m.db.SelectFromReplicationTasksDLQ(&sqlplugin.ReplicationTasksDLQFilter{
-		ReplicationTasksFilter: filter,
-		SourceClusterName:      request.SourceClusterName,
+	rows, err := m.db.RangeSelectFromReplicationDLQTasks(sqlplugin.ReplicationDLQTasksRangeFilter{
+		ShardID:           m.shardID,
+		MinTaskID:         readLevel,
+		MaxTaskID:         maxReadLevelInclusive,
+		PageSize:          request.BatchSize,
+		SourceClusterName: request.SourceClusterName,
 	})
 
 	switch err {
 	case nil:
-		return m.populateGetReplicationTasksResponse(rows, request.MaxReadLevel)
+		return m.populateGetReplicationDLQTasksResponse(rows, request.MaxReadLevel)
 	case sql.ErrNoRows:
 		return &p.GetReplicationTasksResponse{}, nil
 	default:
@@ -914,14 +940,10 @@ func (m *sqlExecutionManager) DeleteReplicationTaskFromDLQ(
 	request *p.DeleteReplicationTaskFromDLQRequest,
 ) error {
 
-	filter := sqlplugin.ReplicationTasksFilter{
-		ShardID: m.shardID,
-		TaskID:  request.TaskID,
-	}
-
-	if _, err := m.db.DeleteMessageFromReplicationTasksDLQ(&sqlplugin.ReplicationTasksDLQFilter{
-		ReplicationTasksFilter: filter,
-		SourceClusterName:      request.SourceClusterName,
+	if _, err := m.db.DeleteFromReplicationDLQTasks(sqlplugin.ReplicationDLQTasksFilter{
+		ShardID:           m.shardID,
+		TaskID:            request.TaskID,
+		SourceClusterName: request.SourceClusterName,
 	}); err != nil {
 		return err
 	}
@@ -932,15 +954,11 @@ func (m *sqlExecutionManager) RangeDeleteReplicationTaskFromDLQ(
 	request *p.RangeDeleteReplicationTaskFromDLQRequest,
 ) error {
 
-	filter := sqlplugin.ReplicationTasksFilter{
-		ShardID:            m.shardID,
-		TaskID:             request.ExclusiveBeginTaskID,
-		InclusiveEndTaskID: request.InclusiveEndTaskID,
-	}
-
-	if _, err := m.db.RangeDeleteMessageFromReplicationTasksDLQ(&sqlplugin.ReplicationTasksDLQFilter{
-		ReplicationTasksFilter: filter,
-		SourceClusterName:      request.SourceClusterName,
+	if _, err := m.db.RangeDeleteFromReplicationDLQTasks(sqlplugin.ReplicationDLQTasksRangeFilter{
+		ShardID:           m.shardID,
+		SourceClusterName: request.SourceClusterName,
+		MinTaskID:         request.ExclusiveBeginTaskID,
+		MaxTaskID:         request.InclusiveEndTaskID,
 	}); err != nil {
 		return err
 	}
@@ -1075,15 +1093,13 @@ func (m *sqlExecutionManager) PutReplicationTaskToDLQ(request *p.PutReplicationT
 		return err
 	}
 
-	row := &sqlplugin.ReplicationTaskDLQRow{
+	_, err = m.db.InsertIntoReplicationDLQTasks([]sqlplugin.ReplicationDLQTasksRow{{
 		SourceClusterName: request.SourceClusterName,
 		ShardID:           m.shardID,
 		TaskID:            replicationTask.GetTaskId(),
 		Data:              blob.Data,
 		DataEncoding:      blob.Encoding.String(),
-	}
-
-	_, err = m.db.InsertIntoReplicationTasksDLQ(row)
+	}})
 
 	// Tasks are immutable. So it's fine if we already persisted it before.
 	// This can happen when tasks are retried (ack and cleanup can have lag on source side).

common/persistence/sql/sqlplugin/definition.go

@@ -176,41 +176,6 @@ type (
 		PageSize                    *int
 	}
 
-	// ReplicationTasksRow represents a row in replication_tasks table
-	ReplicationTasksRow struct {
-		ShardID      int32
-		TaskID       int64
-		Data         []byte
-		DataEncoding string
-	}
-
-	// ReplicationTaskDLQRow represents a row in replication_tasks_dlq table
-	ReplicationTaskDLQRow struct {
-		SourceClusterName string
-		ShardID           int32
-		TaskID            int64
-		Data              []byte
-		DataEncoding      string
-	}
-
-	// ReplicationTasksFilter contains the column names within replication_tasks table that
-	// can be used to filter results through a WHERE clause
-	ReplicationTasksFilter struct {
-		ShardID            int32
-		TaskID             int64
-		InclusiveEndTaskID int64
-		MinTaskID          int64
-		MaxTaskID          int64
-		PageSize           int
-	}
-
-	// ReplicationTasksDLQFilter contains the column names within replication_tasks_dlq table that
-	// can be used to filter results through a WHERE clause
-	ReplicationTasksDLQFilter struct {
-		ReplicationTasksFilter
-		SourceClusterName string
-	}
-
 	// EventsRow represents a row in events table
 	EventsRow struct {
 		NamespaceID  primitives.UUID

common/persistence/sql/sqlplugin/history.go

@@ -42,30 +42,4 @@ type (
 		SelectFromBufferedEvents(filter *BufferedEventsFilter) ([]BufferedEventsRow, error)
 		DeleteFromBufferedEvents(filter *BufferedEventsFilter) (sql.Result, error)
 	}
-
-	// HistoryReplicationTask is the SQL persistence interface for history nodes and history replication tasks
-	HistoryReplicationTask interface {
-		InsertIntoReplicationTasks(rows []ReplicationTasksRow) (sql.Result, error)
-		// SelectFromReplicationTasks returns one or more rows from replication_tasks table
-		// Required filter params - {shardID, minTaskID, maxTaskID, pageSize}
-		SelectFromReplicationTasks(filter *ReplicationTasksFilter) ([]ReplicationTasksRow, error)
-		// DeleteFromReplicationTasks deletes a row from replication_tasks table
-		// Required filter params - {shardID, inclusiveEndTaskID}
-		DeleteFromReplicationTasks(filter *ReplicationTasksFilter) (sql.Result, error)
-		// DeleteFromReplicationTasks deletes multi rows from replication_tasks table
-		// Required filter params - {shardID, inclusiveEndTaskID}
-		RangeDeleteFromReplicationTasks(filter *ReplicationTasksFilter) (sql.Result, error)
-
-		// InsertIntoReplicationTasksDLQ puts the replication task into DLQ
-		InsertIntoReplicationTasksDLQ(row *ReplicationTaskDLQRow) (sql.Result, error)
-		// SelectFromReplicationTasksDLQ returns one or more rows from replication_tasks_dlq table
-		// Required filter params - {sourceClusterName, shardID, minTaskID, pageSize}
-		SelectFromReplicationTasksDLQ(filter *ReplicationTasksDLQFilter) ([]ReplicationTasksRow, error)
-		// DeleteMessageFromReplicationTasksDLQ deletes one row from replication_tasks_dlq table
-		// Required filter params - {sourceClusterName, shardID, taskID}
-		DeleteMessageFromReplicationTasksDLQ(filter *ReplicationTasksDLQFilter) (sql.Result, error)
-		// RangeDeleteMessageFromReplicationTasksDLQ deletes one or more rows from replication_tasks_dlq table
-		// Required filter params - {sourceClusterName, shardID, taskID, inclusiveTaskID}
-		RangeDeleteMessageFromReplicationTasksDLQ(filter *ReplicationTasksDLQFilter) (sql.Result, error)
-	}
 )

common/persistence/sql/sqlplugin/history_activity.go

@@ -1,3 +1,27 @@
+// The MIT License
+//
+// Copyright (c) 2020 Temporal Technologies Inc.  All rights reserved.
+//
+// Copyright (c) 2020 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
 package sqlplugin
 
 import (

common/persistence/sql/sqlplugin/history_child_workflow.go

@@ -1,3 +1,27 @@
+// The MIT License
+//
+// Copyright (c) 2020 Temporal Technologies Inc.  All rights reserved.
+//
+// Copyright (c) 2020 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
 package sqlplugin
 
 import (
@@ -33,7 +57,7 @@ type (
 		InitiatedID *int64
 	}
 
-	// HistoryExecutionChildWorkflow is the SQL persistence interface for history nodes and history execution child workflows
+	// HistoryExecutionChildWorkflow is the SQL persistence interface for history execution child workflows
 	HistoryExecutionChildWorkflow interface {
 		ReplaceIntoChildExecutionInfoMaps(rows []ChildExecutionInfoMapsRow) (sql.Result, error)
 		// SelectFromChildExecutionInfoMaps returns one or more rows form child_execution_info_maps table

common/persistence/sql/sqlplugin/history_execution.go

@@ -79,7 +79,7 @@ type (
 		RunID       primitives.UUID
 	}
 
-	// HistoryExecution is the SQL persistence interface for history nodes and history executions
+	// HistoryExecution is the SQL persistence interface for history executions
 	HistoryExecution interface {
 		InsertIntoExecutions(row *ExecutionsRow) (sql.Result, error)
 		UpdateExecutions(row *ExecutionsRow) (sql.Result, error)

common/persistence/sql/sqlplugin/history_replication_dlq_tasks.go

@@ -0,0 +1,70 @@
+// The MIT License
+//
+// Copyright (c) 2020 Temporal Technologies Inc.  All rights reserved.
+//
+// Copyright (c) 2020 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package sqlplugin
+
+import "database/sql"
+
+type (
+	// ReplicationDLQTasksRow represents a row in replication_tasks_dlq table
+	ReplicationDLQTasksRow struct {
+		SourceClusterName string
+		ShardID           int32
+		TaskID            int64
+		Data              []byte
+		DataEncoding      string
+	}
+
+	// ReplicationDLQTasksFilter contains the column names within replication_tasks_dlq table that
+	// can be used to filter results through a WHERE clause
+	ReplicationDLQTasksFilter struct {
+		ShardID           int32
+		SourceClusterName string
+		TaskID            int64
+	}
+
+	// ReplicationDLQTasksRangeFilter
+	ReplicationDLQTasksRangeFilter struct {
+		ShardID           int32
+		SourceClusterName string
+		MinTaskID         int64
+		MaxTaskID         int64
+		PageSize          int
+	}
+
+	// HistoryReplicationDLQTask is the SQL persistence interface for history replication tasks DLQ
+	HistoryReplicationDLQTask interface {
+		// InsertIntoReplicationDLQTasks puts the replication task into DLQ
+		InsertIntoReplicationDLQTasks(row []ReplicationDLQTasksRow) (sql.Result, error)
+		// SelectFromReplicationDLQTasks returns one or more rows from replication_tasks_dlq table
+		SelectFromReplicationDLQTasks(filter ReplicationDLQTasksFilter) ([]ReplicationDLQTasksRow, error)
+		// RangeSelectFromReplicationDLQTasks returns one or more rows from replication_tasks_dlq table
+		RangeSelectFromReplicationDLQTasks(filter ReplicationDLQTasksRangeFilter) ([]ReplicationDLQTasksRow, error)
+		// DeleteFromReplicationDLQTasks deletes one row from replication_tasks_dlq table
+		DeleteFromReplicationDLQTasks(filter ReplicationDLQTasksFilter) (sql.Result, error)
+		// RangeDeleteFromReplicationDLQTasks deletes one or more rows from replication_tasks_dlq table
+		//  ReplicationDLQTasksRangeFilter - {PageSize} will be ignored
+		RangeDeleteFromReplicationDLQTasks(filter ReplicationDLQTasksRangeFilter) (sql.Result, error)
+	}
+)