Merge branch 'master' into master

cmd/server/go.mod

@@ -40,7 +40,7 @@ require (
 	github.com/startreedata/pinot-client-go v0.2.0 // latest release supports pinot v0.12.0 which is also internal version
 	github.com/stretchr/testify v1.8.3
 	github.com/uber-go/tally v3.3.15+incompatible // indirect
-	github.com/uber/cadence-idl v0.0.0-20240326234200-e3a59cdd3c36
+	github.com/uber/cadence-idl v0.0.0-20240416202333-83d5cae7fc51
 	github.com/uber/ringpop-go v0.8.5 // indirect
 	github.com/uber/tchannel-go v1.22.2 // indirect
 	github.com/urfave/cli v1.22.4

cmd/server/go.sum

@@ -398,8 +398,8 @@ github.com/uber-go/tally v3.3.12+incompatible/go.mod h1:YDTIBxdXyOU/sCWilKB4bgyu
 github.com/uber-go/tally v3.3.15+incompatible h1:9hLSgNBP28CjIaDmAuRTq9qV+UZY+9PcvAkXO4nNMwg=
 github.com/uber-go/tally v3.3.15+incompatible/go.mod h1:YDTIBxdXyOU/sCWilKB4bgyufu1cEi0jdVnRdxvjnmU=
 github.com/uber/cadence-idl v0.0.0-20211111101836-d6b70b60eb8c/go.mod h1:oyUK7GCNCRHCCyWyzifSzXpVrRYVBbAMHAzF5dXiKws=
-github.com/uber/cadence-idl v0.0.0-20240326234200-e3a59cdd3c36 h1:75Uvw7a4Pqo5LrpAFfJehKSJT50JcF7dQolJj5/bUc4=
-github.com/uber/cadence-idl v0.0.0-20240326234200-e3a59cdd3c36/go.mod h1:oyUK7GCNCRHCCyWyzifSzXpVrRYVBbAMHAzF5dXiKws=
+github.com/uber/cadence-idl v0.0.0-20240416202333-83d5cae7fc51 h1:gAdDymipzIzh7PTPAW7DjFT2BOEwY7hwC0COsZIulsI=
+github.com/uber/cadence-idl v0.0.0-20240416202333-83d5cae7fc51/go.mod h1:oyUK7GCNCRHCCyWyzifSzXpVrRYVBbAMHAzF5dXiKws=
 github.com/uber/jaeger-client-go v2.22.1+incompatible h1:NHcubEkVbahf9t3p75TOCR83gdUHXjRJvjoBh1yACsM=
 github.com/uber/jaeger-client-go v2.22.1+incompatible/go.mod h1:WVhlPFC8FDjOFMMWRy2pZqQJSXxYSwNYOkTr/Z6d3Kk=
 github.com/uber/jaeger-lib v2.2.0+incompatible h1:MxZXOiR2JuoANZ3J6DE/U0kSFv/eJ/GfSYVCjK7dyaw=

common/persistence/nosql/nosql_history_store_test.go

@@ -456,3 +456,121 @@ func TestForkHistoryBranch_AllAncestors(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, expecedResp, resp)
 }
+
+func getValidInternalDeleteHistoryBranchRequest() *persistence.InternalDeleteHistoryBranchRequest {
+	return &persistence.InternalDeleteHistoryBranchRequest{
+		BranchInfo: types.HistoryBranch{
+			TreeID:   "TestTreeID",
+			BranchID: "TestBranchID",
+			Ancestors: []*types.HistoryBranchRange{
+				{
+					BranchID:    "TestAncestorBranchID",
+					BeginNodeID: 0,
+					EndNodeID:   5,
+				},
+				{
+					BranchID:    "TestAncestorBranchID",
+					BeginNodeID: 6,
+					EndNodeID:   10,
+				},
+			},
+		},
+		ShardID: testShardID,
+	}
+}
+
+func TestDeleteHistoryBranch_unusedBranch(t *testing.T) {
+	store, dbMock := setUpMocks(t)
+
+	request := getValidInternalDeleteHistoryBranchRequest()
+
+	expecedTreeFilter := &nosqlplugin.HistoryTreeFilter{
+		ShardID:  testShardID,
+		TreeID:   "TestTreeID",
+		BranchID: common.Ptr("TestBranchID"),
+	}
+
+	// Delete in reverse order, add 0 in the end
+	expectedNodeFilters := []*nosqlplugin.HistoryNodeFilter{
+		{
+			ShardID:   testShardID,
+			TreeID:    "TestTreeID",
+			BranchID:  "TestBranchID",
+			MinNodeID: 10,
+		},
+		{
+			ShardID:   testShardID,
+			TreeID:    "TestTreeID",
+			BranchID:  "TestAncestorBranchID",
+			MinNodeID: 6,
+		},
+		{
+			ShardID:   testShardID,
+			TreeID:    "TestTreeID",
+			BranchID:  "TestAncestorBranchID",
+			MinNodeID: 0,
+		},
+	}
+
+	// Expect to delete the history branch
+	dbMock.EXPECT().DeleteFromHistoryTreeAndNode(gomock.Any(), expecedTreeFilter, expectedNodeFilters).
+		Return(nil).Times(1)
+	dbMock.EXPECT().SelectFromHistoryTree(gomock.Any(), gomock.Any()).
+		Return(nil, nil).Times(1)
+
+	err := store.DeleteHistoryBranch(ctx.Background(), request)
+	assert.NoError(t, err)
+}
+
+func TestDeleteHistoryBranch_usedBranch(t *testing.T) {
+	store, dbMock := setUpMocks(t)
+
+	request := getValidInternalDeleteHistoryBranchRequest()
+
+	expecedTreeFilter := &nosqlplugin.HistoryTreeFilter{
+		ShardID:  testShardID,
+		TreeID:   "TestTreeID",
+		BranchID: common.Ptr("TestBranchID"),
+	}
+
+	// Delete in reverse order, add 0 in the end
+	expectedNodeFilters := []*nosqlplugin.HistoryNodeFilter{
+		{
+			ShardID:   testShardID,
+			TreeID:    "TestTreeID",
+			BranchID:  "TestBranchID",
+			MinNodeID: 10,
+		},
+		{
+			ShardID:   testShardID,
+			TreeID:    "TestTreeID",
+			BranchID:  "TestAncestorBranchID",
+			MinNodeID: 7,
+		},
+	}
+
+	// Expect to delete the history branch
+	dbMock.EXPECT().DeleteFromHistoryTreeAndNode(gomock.Any(), expecedTreeFilter, expectedNodeFilters).
+		Return(nil).Times(1)
+
+	historyTree := []*nosqlplugin.HistoryTreeRow{
+		{
+			ShardID:  testShardID,
+			TreeID:   "TestTreeID",
+			BranchID: "TestBranchID",
+			Ancestors: []*types.HistoryBranchRange{
+				{
+					BranchID:    "TestAncestorBranchID",
+					BeginNodeID: 0,
+					EndNodeID:   7,
+				},
+			},
+		},
+	}
+
+	dbMock.EXPECT().SelectFromHistoryTree(gomock.Any(), gomock.Any()).
+		Return(historyTree, nil).Times(1)
+
+	err := store.DeleteHistoryBranch(ctx.Background(), request)
+	assert.NoError(t, err)
+}

common/pinot/pinotQueryValidator.go

@@ -386,6 +386,12 @@ func processCustomString(comparisonExpr *sqlparser.ComparisonExpr, colNameStr st
 		Type: sqlparser.StrVal,
 		Val:  []byte("%" + colValStr + "%"),
 	}
+
+	if colValStr == "" {
+		return fmt.Sprintf("(JSON_MATCH(Attr, '\"$.%s\" is not null') "+
+			"AND REGEXP_LIKE(JSON_EXTRACT_SCALAR(Attr, '$.%s', 'string'), '^$'))", colNameStr, colNameStr)
+	}
+
 	return fmt.Sprintf("(JSON_MATCH(Attr, '\"$.%s\" is not null') "+
 		"AND REGEXP_LIKE(JSON_EXTRACT_SCALAR(Attr, '$.%s', 'string'), '%s*'))", colNameStr, colNameStr, colValStr)
 }

common/pinot/pinotQueryValidator_test.go

@@ -46,10 +46,14 @@ func TestValidateQuery(t *testing.T) {
 			query:     "WorkflowID = 'wid'",
 			validated: "WorkflowID = 'wid'",
 		},
-		"Case3: query with custom field": {
+		"Case3-1: query with custom field": {
 			query:     "CustomStringField = 'custom'",
 			validated: "(JSON_MATCH(Attr, '\"$.CustomStringField\" is not null') AND REGEXP_LIKE(JSON_EXTRACT_SCALAR(Attr, '$.CustomStringField', 'string'), 'custom*'))",
 		},
+		"Case3-2: query with custom field value is empty": {
+			query:     "CustomStringField = ''",
+			validated: "(JSON_MATCH(Attr, '\"$.CustomStringField\" is not null') AND REGEXP_LIKE(JSON_EXTRACT_SCALAR(Attr, '$.CustomStringField', 'string'), '^$'))",
+		},
 		"Case4: custom field query with or in string": {
 			query:     "CustomStringField='Or'",
 			validated: "(JSON_MATCH(Attr, '\"$.CustomStringField\" is not null') AND REGEXP_LIKE(JSON_EXTRACT_SCALAR(Attr, '$.CustomStringField', 'string'), 'Or*'))",

common/types/history.go

@@ -521,6 +521,7 @@ type RecordChildExecutionCompletedRequest struct {
 	InitiatedID        int64              `json:"initiatedId,omitempty"`
 	CompletedExecution *WorkflowExecution `json:"completedExecution,omitempty"`
 	CompletionEvent    *HistoryEvent      `json:"completionEvent,omitempty"`
+	StartedID          int64              `json:"startedId,omitempty"`
 }
 
 // GetDomainUUID is an internal getter (TBD...)

common/types/mapper/proto/api.go

@@ -332,6 +332,8 @@ func FromCancelExternalWorkflowExecutionFailedCause(t *types.CancelExternalWorkf
 	switch *t {
 	case types.CancelExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution:
 		return apiv1.CancelExternalWorkflowExecutionFailedCause_CANCEL_EXTERNAL_WORKFLOW_EXECUTION_FAILED_CAUSE_UNKNOWN_EXTERNAL_WORKFLOW_EXECUTION
+	case types.CancelExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted:
+		return apiv1.CancelExternalWorkflowExecutionFailedCause_CANCEL_EXTERNAL_WORKFLOW_EXECUTION_FAILED_CAUSE_WORKFLOW_ALREADY_COMPLETED
 	}
 	panic("unexpected enum value")
 }
@@ -342,6 +344,8 @@ func ToCancelExternalWorkflowExecutionFailedCause(t apiv1.CancelExternalWorkflow
 		return nil
 	case apiv1.CancelExternalWorkflowExecutionFailedCause_CANCEL_EXTERNAL_WORKFLOW_EXECUTION_FAILED_CAUSE_UNKNOWN_EXTERNAL_WORKFLOW_EXECUTION:
 		return types.CancelExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution.Ptr()
+	case apiv1.CancelExternalWorkflowExecutionFailedCause_CANCEL_EXTERNAL_WORKFLOW_EXECUTION_FAILED_CAUSE_WORKFLOW_ALREADY_COMPLETED:
+		return types.CancelExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted.Ptr()
 	}
 	panic("unexpected enum value")
 }
@@ -3283,6 +3287,8 @@ func FromSignalExternalWorkflowExecutionFailedCause(t *types.SignalExternalWorkf
 	switch *t {
 	case types.SignalExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution:
 		return apiv1.SignalExternalWorkflowExecutionFailedCause_SIGNAL_EXTERNAL_WORKFLOW_EXECUTION_FAILED_CAUSE_UNKNOWN_EXTERNAL_WORKFLOW_EXECUTION
+	case types.SignalExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted:
+		return apiv1.SignalExternalWorkflowExecutionFailedCause_SIGNAL_EXTERNAL_WORKFLOW_EXECUTION_FAILED_CAUSE_WORKFLOW_ALREADY_COMPLETED
 	}
 	panic("unexpected enum value")
 }
@@ -3293,6 +3299,8 @@ func ToSignalExternalWorkflowExecutionFailedCause(t apiv1.SignalExternalWorkflow
 		return nil
 	case apiv1.SignalExternalWorkflowExecutionFailedCause_SIGNAL_EXTERNAL_WORKFLOW_EXECUTION_FAILED_CAUSE_UNKNOWN_EXTERNAL_WORKFLOW_EXECUTION:
 		return types.SignalExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution.Ptr()
+	case apiv1.SignalExternalWorkflowExecutionFailedCause_SIGNAL_EXTERNAL_WORKFLOW_EXECUTION_FAILED_CAUSE_WORKFLOW_ALREADY_COMPLETED:
+		return types.SignalExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted.Ptr()
 	}
 	panic("unexpected enum value")
 }

common/types/mapper/proto/enum_test.go

@@ -99,6 +99,7 @@ func TestCancelExternalWorkflowExecutionFailedCause(t *testing.T) {
 	for _, item := range []*types.CancelExternalWorkflowExecutionFailedCause{
 		nil,
 		types.CancelExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution.Ptr(),
+		types.CancelExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted.Ptr(),
 	} {
 		assert.Equal(t, item, ToCancelExternalWorkflowExecutionFailedCause(FromCancelExternalWorkflowExecutionFailedCause(item)))
 	}
@@ -297,6 +298,7 @@ func TestSignalExternalWorkflowExecutionFailedCause(t *testing.T) {
 	for _, item := range []*types.SignalExternalWorkflowExecutionFailedCause{
 		nil,
 		types.SignalExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution.Ptr(),
+		types.SignalExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted.Ptr(),
 	} {
 		assert.Equal(t, item, ToSignalExternalWorkflowExecutionFailedCause(FromSignalExternalWorkflowExecutionFailedCause(item)))
 	}

common/types/mapper/proto/history.go

@@ -833,6 +833,7 @@ func FromHistoryRecordChildExecutionCompletedRequest(t *types.RecordChildExecuti
 		InitiatedId:        t.InitiatedID,
 		CompletedExecution: FromWorkflowExecution(t.CompletedExecution),
 		CompletionEvent:    FromHistoryEvent(t.CompletionEvent),
+		StartedId:          t.StartedID,
 	}
 }
 
@@ -846,6 +847,7 @@ func ToHistoryRecordChildExecutionCompletedRequest(t *historyv1.RecordChildExecu
 		InitiatedID:        t.InitiatedId,
 		CompletedExecution: ToWorkflowExecution(t.CompletedExecution),
 		CompletionEvent:    ToHistoryEvent(t.CompletionEvent),
+		StartedID:          t.StartedId,
 	}
 }
 

common/types/mapper/thrift/history.go

@@ -629,6 +629,7 @@ func FromRecordChildExecutionCompletedRequest(t *types.RecordChildExecutionCompl
 		InitiatedId:        &t.InitiatedID,
 		CompletedExecution: FromWorkflowExecution(t.CompletedExecution),
 		CompletionEvent:    FromHistoryEvent(t.CompletionEvent),
+		StartedId:          &t.StartedID,
 	}
 }
 
@@ -643,6 +644,7 @@ func ToRecordChildExecutionCompletedRequest(t *history.RecordChildExecutionCompl
 		InitiatedID:        t.GetInitiatedId(),
 		CompletedExecution: ToWorkflowExecution(t.CompletedExecution),
 		CompletionEvent:    ToHistoryEvent(t.CompletionEvent),
+		StartedID:          t.GetStartedId(),
 	}
 }
 

common/types/mapper/thrift/shared.go

@@ -415,6 +415,9 @@ func FromCancelExternalWorkflowExecutionFailedCause(t *types.CancelExternalWorkf
 	case types.CancelExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution:
 		v := shared.CancelExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution
 		return &v
+	case types.CancelExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted:
+		v := shared.CancelExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted
+		return &v
 	}
 	panic("unexpected enum value")
 }
@@ -428,6 +431,9 @@ func ToCancelExternalWorkflowExecutionFailedCause(t *shared.CancelExternalWorkfl
 	case shared.CancelExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution:
 		v := types.CancelExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution
 		return &v
+	case shared.CancelExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted:
+		v := types.CancelExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted
+		return &v
 	}
 	panic("unexpected enum value")
 }
@@ -4931,6 +4937,9 @@ func FromSignalExternalWorkflowExecutionFailedCause(t *types.SignalExternalWorkf
 	case types.SignalExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution:
 		v := shared.SignalExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution
 		return &v
+	case types.SignalExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted:
+		v := shared.SignalExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted
+		return &v
 	}
 	panic("unexpected enum value")
 }
@@ -4944,6 +4953,9 @@ func ToSignalExternalWorkflowExecutionFailedCause(t *shared.SignalExternalWorkfl
 	case shared.SignalExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution:
 		v := types.SignalExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution
 		return &v
+	case shared.SignalExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted:
+		v := types.SignalExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted
+		return &v
 	}
 	panic("unexpected enum value")
 }

common/types/mapper/thrift/shared_test.go

@@ -476,6 +476,7 @@ func TestCancelExternalWorkflowExecutionFailedCauseConversion(t *testing.T) {
 	testCases := []*types.CancelExternalWorkflowExecutionFailedCause{
 		nil,
 		types.CancelExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution.Ptr(),
+		types.CancelExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted.Ptr(),
 	}
 
 	for _, original := range testCases {
@@ -2603,6 +2604,7 @@ func TestSignalExternalWorkflowExecutionFailedCauseConversion(t *testing.T) {
 	testCases := []*types.SignalExternalWorkflowExecutionFailedCause{
 		nil,
 		types.SignalExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution.Ptr(),
+		types.SignalExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted.Ptr(),
 	}
 
 	for _, original := range testCases {

common/types/shared.go

@@ -381,6 +381,8 @@ func (e CancelExternalWorkflowExecutionFailedCause) String() string {
 	switch w {
 	case 0:
 		return "UNKNOWN_EXTERNAL_WORKFLOW_EXECUTION"
+	case 1:
+		return "WORKFLOW_ALREADY_COMPLETED"
 	}
 	return fmt.Sprintf("CancelExternalWorkflowExecutionFailedCause(%d)", w)
 }
@@ -391,6 +393,9 @@ func (e *CancelExternalWorkflowExecutionFailedCause) UnmarshalText(value []byte)
 	case "UNKNOWN_EXTERNAL_WORKFLOW_EXECUTION":
 		*e = CancelExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution
 		return nil
+	case "WORKFLOW_ALREADY_COMPLETED":
+		*e = CancelExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted
+		return nil
 	default:
 		val, err := strconv.ParseInt(s, 10, 32)
 		if err != nil {
@@ -409,6 +414,7 @@ func (e CancelExternalWorkflowExecutionFailedCause) MarshalText() ([]byte, error
 const (
 	// CancelExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution is an option for CancelExternalWorkflowExecutionFailedCause
 	CancelExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution CancelExternalWorkflowExecutionFailedCause = iota
+	CancelExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted
 )
 
 // CancelTimerDecisionAttributes is an internal type (TBD...)
@@ -5739,6 +5745,8 @@ func (e SignalExternalWorkflowExecutionFailedCause) String() string {
 	switch w {
 	case 0:
 		return "UNKNOWN_EXTERNAL_WORKFLOW_EXECUTION"
+	case 1:
+		return "WORKFLOW_ALREADY_COMPLETED"
 	}
 	return fmt.Sprintf("SignalExternalWorkflowExecutionFailedCause(%d)", w)
 }
@@ -5749,6 +5757,9 @@ func (e *SignalExternalWorkflowExecutionFailedCause) UnmarshalText(value []byte)
 	case "UNKNOWN_EXTERNAL_WORKFLOW_EXECUTION":
 		*e = SignalExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution
 		return nil
+	case "WORKFLOW_ALREADY_COMPLETED":
+		*e = SignalExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted
+		return nil
 	default:
 		val, err := strconv.ParseInt(s, 10, 32)
 		if err != nil {
@@ -5767,6 +5778,7 @@ func (e SignalExternalWorkflowExecutionFailedCause) MarshalText() ([]byte, error
 const (
 	// SignalExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution is an option for SignalExternalWorkflowExecutionFailedCause
 	SignalExternalWorkflowExecutionFailedCauseUnknownExternalWorkflowExecution SignalExternalWorkflowExecutionFailedCause = iota
+	SignalExternalWorkflowExecutionFailedCauseWorkflowAlreadyCompleted
 )
 
 // SignalExternalWorkflowExecutionFailedEventAttributes is an internal type (TBD...)

common/types/testdata/service_history.go

@@ -147,6 +147,7 @@ var (
 		InitiatedID:        EventID1,
 		CompletedExecution: &WorkflowExecution,
 		CompletionEvent:    &HistoryEvent_WorkflowExecutionStarted,
+		StartedID:          EventID2,
 	}
 	HistoryRecordDecisionTaskStartedRequest = types.RecordDecisionTaskStartedRequest{
 		DomainUUID:        DomainID,

go.mod

@@ -40,7 +40,7 @@ require (
 	github.com/startreedata/pinot-client-go v0.2.0 // latest release supports pinot v0.12.0 which is also internal version
 	github.com/stretchr/testify v1.8.3
 	github.com/uber-go/tally v3.3.15+incompatible
-	github.com/uber/cadence-idl v0.0.0-20240326234200-e3a59cdd3c36
+	github.com/uber/cadence-idl v0.0.0-20240416202333-83d5cae7fc51
 	github.com/uber/ringpop-go v0.8.5
 	github.com/uber/tchannel-go v1.22.2
 	github.com/urfave/cli v1.22.4