Add CHASM visibility task handling of alias to field mapping for custom search attributes (#8558)

Add CHASM visibility task handling of alias to field mapping for custom
search attributes. The Visibility component stores aliased custom search
attributes and their corresponding payloads.

On the WRITE path, Visibility Store CURRENTLY assumes search attributes
are schema field names, and does not apply any mappings from alias ->
field. This is because FrontEnd "Unaliases" to its field names before
sending to History.

## What changed?
Before calling the visibility manager `UpsertWorkflowExecution`, the
search attributes need to be mapped from their aliases to their field
names.

## Why?
Support Visibility component compatibility with current Visibility Store
implementation.

## How did you test it?
- [X] built
- [X] run locally and tested manually
- [X] covered by existing tests
- [X] added new unit test(s)
- [ ] added new functional test(s)

Author: awln-temporal

common/persistence/visibility/store/elasticsearch/visibility_store_read_test.go

@@ -336,7 +336,7 @@ func (s *ESVisibilitySuite) Test_convertQueryLegacy() {
 	query = `StartTime = "2018-06-07T15:04:05.123456789-08:00"`
 	queryParams, err = s.visibilityStore.convertQueryLegacy(testNamespace, testNamespaceID, query)
 	s.NoError(err)
-	s.Equal(`{"bool":{"filter":[{"term":{"NamespaceId":"bfd5c907-f899-4baf-a7b2-2ab85e623ebd"}},{"bool":{"filter":{"match":{"StartTime":{"query":"2018-06-07T15:04:05.123456789-08:00"}}}}}],"must_not":{"exists":{"field":"TemporalNamespaceDivision"}}}}`, s.queryToJSON(queryParams.Query))
+	s.JSONEq(`{"bool":{"filter":[{"term":{"NamespaceId":"bfd5c907-f899-4baf-a7b2-2ab85e623ebd"}},{"bool":{"filter":{"match":{"StartTime":{"query":"2018-06-07T15:04:05.123456789-08:00"}}}}}],"must_not":{"exists":{"field":"TemporalNamespaceDivision"}}}}`, s.queryToJSON(queryParams.Query))
 	s.Nil(queryParams.Sorter)
 
 	query = `WorkflowId = 'wid' and StartTime > "2018-06-07T15:04:05+00:00"`

service/history/visibility_queue_task_executor.go

@@ -389,13 +389,33 @@ func (t *visibilityQueueTaskExecutor) processChasmTask(
 		return serviceerror.NewInternalf("expected visibility component, but got %T", visComponent)
 	}
 
-	searchattributes, err := visComponent.GetSearchAttributes(visTaskContext)
+	namespaceEntry, err := t.shardContext.GetNamespaceRegistry().
+		GetNamespaceByID(namespace.ID(task.GetNamespaceID()))
+	if err != nil {
+		return err
+	}
+
+	searchattributesMapperProvider := t.shardContext.GetSearchAttributesMapperProvider()
+	searchAttributesMapper, err := searchattributesMapperProvider.GetMapper(namespaceEntry.Name())
 	if err != nil {
 		return err
 	}
-	if searchattributes == nil {
-		searchattributes = make(map[string]*commonpb.Payload)
+
+	searchattributes := make(map[string]*commonpb.Payload)
+
+	aliasedSearchAttributes, err := visComponent.GetSearchAttributes(visTaskContext)
+	if err != nil {
+		return err
 	}
+
+	for alias, value := range aliasedSearchAttributes {
+		fieldName, err := searchAttributesMapper.GetFieldName(alias, namespaceEntry.Name().String())
+		if err != nil {
+			return err
+		}
+		searchattributes[fieldName] = value
+	}
+
 	memo, err := visComponent.GetMemo(visTaskContext)
 	if err != nil {
 		return err
@@ -419,11 +439,6 @@ func (t *visibilityQueueTaskExecutor) processChasmTask(
 		}
 	}
 
-	namespaceEntry, err := t.shardContext.GetNamespaceRegistry().
-		GetNamespaceByID(namespace.ID(task.GetNamespaceID()))
-	if err != nil {
-		return err
-	}
 	requestBase := t.getVisibilityRequestBase(
 		task,
 		namespaceEntry,

service/history/visibility_queue_task_executor_test.go

@@ -107,6 +107,17 @@ func (s *visibilityQueueTaskExecutorSuite) SetupTest() {
 		config,
 	)
 
+	// Set up expectations on the SearchAttributesMapper mocks created by NewTestContext
+	mockMapper := searchattribute.NewMockMapper(s.controller)
+	mockMapper.EXPECT().GetFieldName(gomock.Any(), gomock.Any()).DoAndReturn(
+		func(alias string, _ string) (string, error) {
+			return alias, nil
+		},
+	).AnyTimes()
+
+	mockMapperProvider := s.mockShard.Resource.SearchAttributesMapperProvider
+	mockMapperProvider.EXPECT().GetMapper(gomock.Any()).Return(mockMapper, nil).AnyTimes()
+
 	reg := hsm.NewRegistry()
 	err := workflow.RegisterStateMachine(reg)
 	s.NoError(err)