Sql task queue visibility (#2537)

* Adding task_queue to read/write paths for postgresql executions_visibility * Adding task_queue for mysql and sqlite
2026-08-30 18:41:49 -07:00 · 2022-02-24 18:22:46 -08:00
parent 9e47a655fc
commit f92cc93828
7 changed files with 58 additions and 32 deletions
--- a/.vscode/launch.json
+++ b/.vscode/launch.json
@@ -38,6 +38,19 @@
 				"start",
 			]
 		},
+		{
+			"name": "Debug Server with PostgreSQL",
+			"type": "go",
+			"request": "launch",
+			"mode": "debug",
+			"program": "${workspaceFolder}/cmd/server",
+			"cwd": "${workspaceFolder}",
+			"args": [
+				"--env",
+				"development_postgres",
+				"start",
+			]
+		},
 		{
 			"name": "Debug CLI Namespace Describe",
 			"type": "go",
--- a/common/persistence/sql/sqlplugin/mysql/visibility.go
+++ b/common/persistence/sql/sqlplugin/mysql/visibility.go
@@ -35,16 +35,16 @@ import (

 const (
 	templateCreateWorkflowExecutionStarted = `INSERT INTO executions_visibility (` +
-		`namespace_id, workflow_id, run_id, start_time, execution_time, workflow_type_name, status, memo, encoding) ` +
-		`VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ` +
+		`namespace_id, workflow_id, run_id, start_time, execution_time, workflow_type_name, status, memo, encoding, task_queue) ` +
+		`VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ` +
 		`ON DUPLICATE KEY UPDATE ` +
 		`run_id=VALUES(run_id)`

 	templateCreateWorkflowExecutionClosed = `INSERT INTO executions_visibility (` +
-		`namespace_id, workflow_id, run_id, start_time, execution_time, workflow_type_name, close_time, status, history_length, memo, encoding) ` +
-		`VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ` +
+		`namespace_id, workflow_id, run_id, start_time, execution_time, workflow_type_name, close_time, status, history_length, memo, encoding, task_queue) ` +
+		`VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ` +
 		`ON DUPLICATE KEY UPDATE workflow_id = VALUES(workflow_id), start_time = VALUES(start_time), execution_time = VALUES(execution_time), workflow_type_name = VALUES(workflow_type_name), ` +
-		`close_time = VALUES(close_time), status = VALUES(status), history_length = VALUES(history_length), memo = VALUES(memo), encoding = VALUES(encoding)`
+		`close_time = VALUES(close_time), status = VALUES(status), history_length = VALUES(history_length), memo = VALUES(memo), encoding = VALUES(encoding), task_queue = VALUES(task_queue)`

 	// RunID condition is needed for correct pagination
 	templateConditions = ` AND namespace_id = ?
@@ -61,7 +61,7 @@ const (
 	ORDER BY close_time DESC, run_id
 	LIMIT ?`

-	templateOpenFieldNames = `workflow_id, run_id, start_time, execution_time, workflow_type_name, status, memo, encoding`
+	templateOpenFieldNames = `workflow_id, run_id, start_time, execution_time, workflow_type_name, status, memo, encoding, task_queue`
 	templateOpenSelect     = `SELECT ` + templateOpenFieldNames + ` FROM executions_visibility WHERE status = 1 `

 	templateClosedSelect = `SELECT ` + templateOpenFieldNames + `, close_time, history_length
@@ -81,7 +81,7 @@ const (

 	templateGetClosedWorkflowExecutionsByStatus = templateClosedSelect + `AND status = ?` + templateConditionsClosedWorkflows

-	templateGetClosedWorkflowExecution = `SELECT workflow_id, run_id, start_time, execution_time, memo, encoding, close_time, workflow_type_name, status, history_length 
+	templateGetClosedWorkflowExecution = `SELECT workflow_id, run_id, start_time, execution_time, memo, encoding, close_time, workflow_type_name, status, history_length, task_queue 
 		 FROM executions_visibility
 		 WHERE namespace_id = ? AND status != 1
 		 AND run_id = ?`
@@ -110,6 +110,7 @@ func (mdb *db) InsertIntoVisibility(
 		row.Status,
 		row.Memo,
 		row.Encoding,
+		row.TaskQueue,
 	)
 }

@@ -136,6 +137,7 @@ func (mdb *db) ReplaceIntoVisibility(
 			*row.HistoryLength,
 			row.Memo,
 			row.Encoding,
+			row.TaskQueue,
 		)
 	default:
 		return nil, errCloseParams
--- a/common/persistence/sql/sqlplugin/postgresql/visibility.go
+++ b/common/persistence/sql/sqlplugin/postgresql/visibility.go
@@ -36,13 +36,13 @@ import (

 const (
 	templateCreateWorkflowExecutionStarted = `INSERT INTO executions_visibility (` +
-		`namespace_id, workflow_id, run_id, start_time, execution_time, workflow_type_name, status, memo, encoding) ` +
-		`VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+		`namespace_id, workflow_id, run_id, start_time, execution_time, workflow_type_name, status, memo, encoding, task_queue) ` +
+		`VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
         ON CONFLICT (namespace_id, run_id) DO NOTHING`

 	templateCreateWorkflowExecutionClosed = `INSERT INTO executions_visibility (` +
-		`namespace_id, workflow_id, run_id, start_time, execution_time, workflow_type_name, close_time, status, history_length, memo, encoding) ` +
-		`VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
+		`namespace_id, workflow_id, run_id, start_time, execution_time, workflow_type_name, close_time, status, history_length, memo, encoding, task_queue) ` +
+		`VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
 		ON CONFLICT (namespace_id, run_id) DO UPDATE 
 		  SET workflow_id = excluded.workflow_id,
 		      start_time = excluded.start_time,
@@ -52,7 +52,8 @@ const (
 			  status = excluded.status,
 			  history_length = excluded.history_length,
 			  memo = excluded.memo,
-			  encoding = excluded.encoding`
+			  encoding = excluded.encoding,
+			  task_queue = excluded.task_queue`

 	// RunID condition is needed for correct pagination
 	templateConditions1 = ` AND namespace_id = $1
@@ -83,7 +84,7 @@ const (
         ORDER BY close_time DESC, run_id
         LIMIT $8`

-	templateOpenFieldNames = `workflow_id, run_id, start_time, execution_time, workflow_type_name, status, memo, encoding`
+	templateOpenFieldNames = `workflow_id, run_id, start_time, execution_time, workflow_type_name, status, memo, encoding, task_queue`
 	templateOpenSelect     = `SELECT ` + templateOpenFieldNames + ` FROM executions_visibility WHERE status = 1 `

 	templateClosedSelect = `SELECT ` + templateOpenFieldNames + `, close_time, history_length
@@ -103,7 +104,7 @@ const (

 	templateGetClosedWorkflowExecutionsByStatus = templateClosedSelect + `AND status = $1` + templateConditionsClosedWorkflow2

-	templateGetClosedWorkflowExecution = `SELECT workflow_id, run_id, start_time, execution_time, memo, encoding, close_time, workflow_type_name, status, history_length 
+	templateGetClosedWorkflowExecution = `SELECT workflow_id, run_id, start_time, execution_time, memo, encoding, close_time, workflow_type_name, status, history_length, task_queue
 		 FROM executions_visibility
 		 WHERE namespace_id = $1 AND status != 1
 		 AND run_id = $2`
@@ -131,6 +132,7 @@ func (pdb *db) InsertIntoVisibility(
 		row.Status,
 		row.Memo,
 		row.Encoding,
+		row.TaskQueue,
 	)
 }

@@ -156,6 +158,7 @@ func (pdb *db) ReplaceIntoVisibility(
 			*row.HistoryLength,
 			row.Memo,
 			row.Encoding,
+			row.TaskQueue,
 		)
 	default:
 		return nil, errCloseParams
--- a/common/persistence/sql/sqlplugin/sqlite/visibility.go
+++ b/common/persistence/sql/sqlplugin/sqlite/visibility.go
@@ -39,13 +39,13 @@ import (

 const (
 	templateCreateWorkflowExecutionStarted = `INSERT INTO executions_visibility (` +
-		`namespace_id, workflow_id, run_id, start_time, execution_time, workflow_type_name, status, memo, encoding) ` +
-		`VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ` +
+		`namespace_id, workflow_id, run_id, start_time, execution_time, workflow_type_name, status, memo, encoding, task_queue) ` +
+		`VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ` +
 		`ON CONFLICT (namespace_id, run_id) DO NOTHING`

 	templateCreateWorkflowExecutionClosed = `REPLACE INTO executions_visibility (` +
-		`namespace_id, workflow_id, run_id, start_time, execution_time, workflow_type_name, close_time, status, history_length, memo, encoding) ` +
-		`VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `
+		`namespace_id, workflow_id, run_id, start_time, execution_time, workflow_type_name, close_time, status, history_length, memo, encoding, task_queue) ` +
+		`VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `

 	// RunID condition is needed for correct pagination
 	templateConditions = ` AND namespace_id = ?
@@ -62,7 +62,7 @@ const (
 	ORDER BY close_time DESC, run_id
 	LIMIT ?`

-	templateOpenFieldNames = `workflow_id, run_id, start_time, execution_time, workflow_type_name, status, memo, encoding`
+	templateOpenFieldNames = `workflow_id, run_id, start_time, execution_time, workflow_type_name, status, memo, encoding, task_queue`
 	templateOpenSelect     = `SELECT ` + templateOpenFieldNames + ` FROM executions_visibility WHERE status = 1 `

 	templateClosedSelect = `SELECT ` + templateOpenFieldNames + `, close_time, history_length
@@ -82,7 +82,7 @@ const (

 	templateGetClosedWorkflowExecutionsByStatus = templateClosedSelect + `AND status = ?` + templateConditionsClosedWorkflows

-	templateGetClosedWorkflowExecution = `SELECT workflow_id, run_id, start_time, execution_time, memo, encoding, close_time, workflow_type_name, status, history_length 
+	templateGetClosedWorkflowExecution = `SELECT workflow_id, run_id, start_time, execution_time, memo, encoding, close_time, workflow_type_name, status, history_length, task_queue 
 		 FROM executions_visibility
 		 WHERE namespace_id = ? AND status != 1
 		 AND run_id = ?`
@@ -110,6 +110,7 @@ func (mdb *db) InsertIntoVisibility(
 		row.Status,
 		row.Memo,
 		row.Encoding,
+		row.TaskQueue,
 	)
 }

@@ -135,6 +136,7 @@ func (mdb *db) ReplaceIntoVisibility(
 			*row.HistoryLength,
 			row.Memo,
 			row.Encoding,
+			row.TaskQueue,
 		)
 	default:
 		return nil, errCloseParams
--- a/common/persistence/sql/sqlplugin/visibility.go
+++ b/common/persistence/sql/sqlplugin/visibility.go
@@ -44,6 +44,7 @@ type (
 		HistoryLength    *int64
 		Memo             []byte
 		Encoding         string
+		TaskQueue        string
 	}

 	// VisibilitySelectFilter contains the column names within executions_visibility table that
--- a/common/persistence/visibility/persistence-tests/visibility_persistence_suite_test.go
+++ b/common/persistence/visibility/persistence-tests/visibility_persistence_suite_test.go
@@ -96,7 +96,7 @@ func (s *VisibilityPersistenceSuite) TearDownSuite() {
 func (s *VisibilityPersistenceSuite) TestBasicVisibility() {
 	testNamespaceUUID := namespace.ID(uuid.New())
 	startTime := time.Now().UTC().Add(time.Second * -5)
-	startReq := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-workflow-test", "visibility-workflow", startTime)
+	startReq := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-workflow-test", "visibility-workflow", startTime, "test-queue")

 	resp, err1 := s.VisibilityMgr.ListOpenWorkflowExecutions(&manager.ListWorkflowExecutionsRequest{
 		NamespaceID:       testNamespaceUUID,
@@ -135,7 +135,7 @@ func (s *VisibilityPersistenceSuite) TestBasicVisibilityTimeSkew() {
 	testNamespaceUUID := namespace.ID(uuid.New())

 	startTime := time.Now()
-	openRecord := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-workflow-test-time-skew", "visibility-workflow", startTime)
+	openRecord := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-workflow-test-time-skew", "visibility-workflow", startTime, "test-queue")

 	resp, err1 := s.VisibilityMgr.ListOpenWorkflowExecutions(&manager.ListWorkflowExecutionsRequest{
 		NamespaceID:       testNamespaceUUID,
@@ -173,7 +173,7 @@ func (s *VisibilityPersistenceSuite) TestBasicVisibilityShortWorkflow() {
 	testNamespaceUUID := namespace.ID(uuid.New())

 	startTime := time.Now().UTC()
-	openRecord := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-workflow-test-short-workflow", "visibility-workflow", startTime)
+	openRecord := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-workflow-test-short-workflow", "visibility-workflow", startTime, "test-queue")
 	closedRecord := s.createClosedWorkflowRecord(openRecord, startTime.Add(10*time.Millisecond))

 	resp, err3 := s.VisibilityMgr.ListOpenWorkflowExecutions(&manager.ListWorkflowExecutionsRequest{
@@ -202,10 +202,10 @@ func (s *VisibilityPersistenceSuite) TestVisibilityPagination() {

 	// Create 2 executions
 	startTime1 := time.Now().UTC()
-	openRecord1 := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-pagination-test1", "visibility-workflow", startTime1)
+	openRecord1 := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-pagination-test1", "visibility-workflow", startTime1, "test-queue")

 	startTime2 := startTime1.Add(time.Second)
-	openRecord2 := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-pagination-test2", "visibility-workflow", startTime2)
+	openRecord2 := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-pagination-test2", "visibility-workflow", startTime2, "test-queue")

 	// Get the first one
 	resp, err2 := s.VisibilityMgr.ListOpenWorkflowExecutions(&manager.ListWorkflowExecutionsRequest{
@@ -251,8 +251,8 @@ func (s *VisibilityPersistenceSuite) TestFilteringByStartTime() {
 	startTime := time.Now()

 	// Create 2 open workflows, one started 2hrs ago, the other started just now.
-	openRecord1 := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-filtering-test1", "visibility-workflow-1", startTime.Add(-2*time.Hour))
-	openRecord2 := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-filtering-test2", "visibility-workflow-2", startTime)
+	openRecord1 := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-filtering-test1", "visibility-workflow-1", startTime.Add(-2*time.Hour), "test-queue")
+	openRecord2 := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-filtering-test2", "visibility-workflow-2", startTime, "test-queue")

 	// List open workflows with start time filter
 	resp, err := s.VisibilityMgr.ListOpenWorkflowExecutions(&manager.ListWorkflowExecutionsRequest{
@@ -301,8 +301,8 @@ func (s *VisibilityPersistenceSuite) TestFilteringByType() {
 	startTime := time.Now()

 	// Create 2 executions
-	openRecord1 := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-filtering-test1", "visibility-workflow-1", startTime)
-	openRecord2 := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-filtering-test2", "visibility-workflow-2", startTime)
+	openRecord1 := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-filtering-test1", "visibility-workflow-1", startTime, "test-queue")
+	openRecord2 := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-filtering-test2", "visibility-workflow-2", startTime, "test-queue")

 	// List open with filtering
 	resp, err2 := s.VisibilityMgr.ListOpenWorkflowExecutionsByType(&manager.ListWorkflowExecutionsByTypeRequest{
@@ -363,8 +363,8 @@ func (s *VisibilityPersistenceSuite) TestFilteringByWorkflowID() {
 	startTime := time.Now()

 	// Create 2 executions
-	openRecord1 := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-filtering-test1", "visibility-workflow", startTime)
-	openRecord2 := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-filtering-test2", "visibility-workflow", startTime)
+	openRecord1 := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-filtering-test1", "visibility-workflow", startTime, "test-queue")
+	openRecord2 := s.createOpenWorkflowRecord(testNamespaceUUID, "visibility-filtering-test2", "visibility-workflow", startTime, "test-queue")

 	// List open with filtering
 	resp, err2 := s.VisibilityMgr.ListOpenWorkflowExecutionsByWorkflowID(&manager.ListWorkflowExecutionsByWorkflowIDRequest{
@@ -631,7 +631,7 @@ func (s *VisibilityPersistenceSuite) TestAdvancedVisibilityPagination() {
 	var startReqs []*manager.RecordWorkflowExecutionStartedRequest
 	var closeReqs []*manager.RecordWorkflowExecutionClosedRequest
 	for i := 0; i < 5; i++ {
-		startReq := s.createOpenWorkflowRecord(testNamespaceUUID, fmt.Sprintf("advanced-visibility-%v", i), "visibility-workflow", time.Now())
+		startReq := s.createOpenWorkflowRecord(testNamespaceUUID, fmt.Sprintf("advanced-visibility-%v", i), "visibility-workflow", time.Now(), "test-queue")
 		if i <= 1 {
 			startReqs = append([]*manager.RecordWorkflowExecutionStartedRequest{startReq}, startReqs...)
 		} else {
@@ -701,6 +701,7 @@ func (s *VisibilityPersistenceSuite) createOpenWorkflowRecord(
 	workflowID string,
 	workflowType string,
 	startTime time.Time,
+	taskQueue string,
 ) *manager.RecordWorkflowExecutionStartedRequest {
 	workflowExecution := commonpb.WorkflowExecution{
 		WorkflowId: workflowID,
@@ -712,6 +713,7 @@ func (s *VisibilityPersistenceSuite) createOpenWorkflowRecord(
 			Execution:        workflowExecution,
 			WorkflowTypeName: workflowType,
 			StartTime:        startTime,
+			TaskQueue:        taskQueue,
 		},
 	}
 	err := s.VisibilityMgr.RecordWorkflowExecutionStarted(startReq)
--- a/common/persistence/visibility/store/standard/sql/visibility_store.go
+++ b/common/persistence/visibility/store/standard/sql/visibility_store.go
@@ -105,6 +105,7 @@ func (s *visibilityStore) RecordWorkflowExecutionStarted(
 		Status:           int32(enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING), // Underlying value (1) is hardcoded in SQL queries.
 		Memo:             request.Memo.Data,
 		Encoding:         request.Memo.EncodingType.String(),
+		TaskQueue:        request.TaskQueue,
 	})

 	return err
@@ -125,6 +126,7 @@ func (s *visibilityStore) RecordWorkflowExecutionClosed(request *store.InternalR
 		HistoryLength:    &request.HistoryLength,
 		Memo:             request.Memo.Data,
 		Encoding:         request.Memo.EncodingType.String(),
+		TaskQueue:        request.TaskQueue,
 	})
 	if err != nil {
 		return err
@@ -349,6 +351,7 @@ func (s *visibilityStore) rowToInfo(
 		ExecutionTime: row.ExecutionTime,
 		Memo:          persistence.NewDataBlob(row.Memo, row.Encoding),
 		Status:        enumspb.WorkflowExecutionStatus(row.Status),
+		TaskQueue:     row.TaskQueue,
 	}
 	if row.CloseTime != nil {
 		info.CloseTime = *row.CloseTime