enable namespace mapper with fallbacks

common/persistence/visibility/store/query/converter_test.go

@@ -23,6 +23,16 @@ const (
 	testNamespaceID   = namespace.ID("test-namespace-id")
 )
 
+type identityMapper struct{}
+
+func (identityMapper) GetAlias(fieldName string, _ string) (string, error) {
+	return fieldName, nil
+}
+
+func (identityMapper) GetFieldName(alias string, _ string) (string, error) {
+	return alias, nil
+}
+
 func TestWithSearchAttributeInterceptor(t *testing.T) {
 	t.Parallel()
 	r := require.New(t)
@@ -1917,7 +1927,7 @@ func TestQueryConverter_ResolveSearchAttributeAlias(t *testing.T) {
 			)
 
 			if tc.useNoopMapper {
-				queryConverter.saMapper = searchattribute.NewNoopMapper()
+				queryConverter.saMapper = identityMapper{}
 			}
 
 			fn, ft, err := queryConverter.resolveSearchAttributeAlias(tc.in)

common/resource/fx.go

@@ -156,11 +156,14 @@ func SearchAttributeMapperProviderProvider(
 	searchAttributeProvider searchattribute.Provider,
 	persistenceConfig *config.Persistence,
 ) searchattribute.MapperProvider {
+	primaryVisibilityStoreConfig := persistenceConfig.GetVisibilityStoreConfig()
+	secondaryVisibilityStoreConfig := persistenceConfig.GetSecondaryVisibilityStoreConfig()
 	return searchattribute.NewMapperProvider(
 		saMapper,
 		namespaceRegistry,
 		searchAttributeProvider,
-		persistenceConfig.IsSQLVisibilityStore() || persistenceConfig.IsCustomVisibilityStore(),
+		primaryVisibilityStoreConfig.GetIndexName(),
+		secondaryVisibilityStoreConfig.GetIndexName(),
 	)
 }
 

common/searchattribute/mapper.go

@@ -20,47 +20,35 @@ type (
 		GetFieldName(alias string, namespace string) (string, error)
 	}
 
-	noopMapper struct{}
-
 	// This mapper is to be backwards compatible with versions before v1.20.
 	// Users using standard visibility might have registered custom search attributes.
 	// Those search attributes won't be searchable, as they weren't before version v1.20.
 	// Thus, this mapper will allow those search attributes to be used without being alised.
 	backCompMapper_v1_20 struct {
-		mapper                 Mapper
-		emptyStringNameTypeMap NameTypeMap
+		mapper               Mapper
+		fallbackNameTypeMaps []NameTypeMap
 	}
 
 	MapperProvider interface {
 		GetMapper(nsName namespace.Name) (Mapper, error)
 	}
 
 	mapperProviderImpl struct {
-		customMapper              Mapper
-		namespaceRegistry         namespace.Registry
-		searchAttributesProvider  Provider
-		enableMapperFromNamespace bool
+		customMapper             Mapper
+		namespaceRegistry        namespace.Registry
+		searchAttributesProvider Provider
+		fallbackIndexNames       []string
 	}
 )
 
-var _ Mapper = (*noopMapper)(nil)
 var _ Mapper = (*backCompMapper_v1_20)(nil)
 var _ Mapper = (*namespace.CustomSearchAttributesMapper)(nil)
 var _ MapperProvider = (*mapperProviderImpl)(nil)
 
-func (m *noopMapper) GetAlias(fieldName string, _ string) (string, error) {
-	return fieldName, nil
-}
-
-func (m *noopMapper) GetFieldName(alias string, _ string) (string, error) {
-	return alias, nil
-}
-
 func (m *backCompMapper_v1_20) GetAlias(fieldName string, namespaceName string) (string, error) {
 	alias, firstErr := m.mapper.GetAlias(fieldName, namespaceName)
 	if firstErr != nil {
-		_, err := m.emptyStringNameTypeMap.getType(fieldName, customCategory)
-		if err != nil {
+		if !m.isLegacyCustomSearchAttribute(fieldName) {
 			return "", firstErr
 		}
 		// this is custom search attribute registered in pre-v1.20
@@ -72,8 +60,7 @@ func (m *backCompMapper_v1_20) GetAlias(fieldName string, namespaceName string)
 func (m *backCompMapper_v1_20) GetFieldName(alias string, namespaceName string) (string, error) {
 	fieldName, firstErr := m.mapper.GetFieldName(alias, namespaceName)
 	if firstErr != nil {
-		_, err := m.emptyStringNameTypeMap.getType(alias, customCategory)
-		if err != nil {
+		if !m.isLegacyCustomSearchAttribute(alias) {
 			return "", firstErr
 		}
 		// this is custom search attribute registered in pre-v1.20
@@ -82,39 +69,65 @@ func (m *backCompMapper_v1_20) GetFieldName(alias string, namespaceName string)
 	return fieldName, nil
 }
 
+func (m *backCompMapper_v1_20) isLegacyCustomSearchAttribute(name string) bool {
+	for _, nameTypeMap := range m.fallbackNameTypeMaps {
+		if _, err := nameTypeMap.getType(name, customCategory); err == nil {
+			return true
+		}
+	}
+	return false
+}
+
 func NewMapperProvider(
 	customMapper Mapper,
 	namespaceRegistry namespace.Registry,
 	searchAttributesProvider Provider,
-	enableMapperFromNamespace bool,
+	fallbackIndexNames ...string,
 ) MapperProvider {
 	return &mapperProviderImpl{
-		customMapper:              customMapper,
-		namespaceRegistry:         namespaceRegistry,
-		searchAttributesProvider:  searchAttributesProvider,
-		enableMapperFromNamespace: enableMapperFromNamespace,
+		customMapper:             customMapper,
+		namespaceRegistry:        namespaceRegistry,
+		searchAttributesProvider: searchAttributesProvider,
+		fallbackIndexNames:       fallbackIndexNames,
 	}
 }
 
 func (m *mapperProviderImpl) GetMapper(nsName namespace.Name) (Mapper, error) {
 	if m.customMapper != nil {
 		return m.customMapper, nil
 	}
-	if !m.enableMapperFromNamespace {
-		return &noopMapper{}, nil
+	if m.namespaceRegistry == nil {
+		return nil, nil
 	}
 	saMapper, err := m.namespaceRegistry.GetCustomSearchAttributesMapper(nsName)
 	if err != nil {
 		return nil, err
 	}
-	// if there's an error, it returns an empty object, which is expected here
-	emptyStringNameTypeMap, _ := m.searchAttributesProvider.GetSearchAttributes("", false)
+	fallbackNameTypeMaps := make([]NameTypeMap, 0, len(m.fallbackIndexNames)+1)
+	for _, indexName := range uniqueFallbackIndexNames(m.fallbackIndexNames) {
+		// If there is an error, it returns an empty object, which is expected here.
+		nameTypeMap, _ := m.searchAttributesProvider.GetSearchAttributes(indexName, false)
+		fallbackNameTypeMaps = append(fallbackNameTypeMaps, nameTypeMap)
+	}
 	return &backCompMapper_v1_20{
-		mapper:                 &saMapper,
-		emptyStringNameTypeMap: emptyStringNameTypeMap,
+		mapper:               &saMapper,
+		fallbackNameTypeMaps: fallbackNameTypeMaps,
 	}, nil
 }
 
+func uniqueFallbackIndexNames(indexNames []string) []string {
+	seen := map[string]struct{}{}
+	result := make([]string, 0, len(indexNames)+1)
+	for _, indexName := range append(indexNames, "") {
+		if _, ok := seen[indexName]; ok {
+			continue
+		}
+		seen[indexName] = struct{}{}
+		result = append(result, indexName)
+	}
+	return result
+}
+
 // AliasFields returns SearchAttributes struct where each custom search attribute name is replaced with alias.
 // If no replacement where made, it returns nil which means that original SearchAttributes struct should be used.
 func AliasFields(

common/searchattribute/mapper_test.go

@@ -5,7 +5,10 @@ import (
 
 	"github.com/stretchr/testify/require"
 	commonpb "go.temporal.io/api/common/v1"
+	enumspb "go.temporal.io/api/enums/v1"
 	"go.temporal.io/api/serviceerror"
+	"go.temporal.io/server/common/namespace"
+	"go.uber.org/mock/gomock"
 )
 
 func Test_AliasFields(t *testing.T) {
@@ -137,3 +140,61 @@ func Test_UnaliasFields(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, sa, sb, "when there is nothin to unalias should return received attributes")
 }
+
+type staticSearchAttributesProvider struct {
+	nameTypeMaps map[string]NameTypeMap
+}
+
+func (s staticSearchAttributesProvider) GetSearchAttributes(indexName string, _ bool) (NameTypeMap, error) {
+	if nameTypeMap, ok := s.nameTypeMaps[indexName]; ok {
+		return nameTypeMap, nil
+	}
+	return NameTypeMap{}, nil
+}
+
+func Test_BackCompMapperFallsBackToClusterMetadataFields(t *testing.T) {
+	mapper := &backCompMapper_v1_20{
+		mapper:               &TestMapper{},
+		fallbackNameTypeMaps: []NameTypeMap{TestNameTypeMap()},
+	}
+
+	alias, err := mapper.GetAlias("Keyword02", "error-namespace")
+	require.NoError(t, err)
+	require.Equal(t, "Keyword02", alias)
+
+	fieldName, err := mapper.GetFieldName("Keyword02", "error-namespace")
+	require.NoError(t, err)
+	require.Equal(t, "Keyword02", fieldName)
+}
+
+func TestMapperProviderUsesConfiguredVisibilityIndexForBackCompatFallback(t *testing.T) {
+	controller := gomock.NewController(t)
+	nsRegistry := namespace.NewMockRegistry(controller)
+	nsRegistry.EXPECT().
+		GetCustomSearchAttributesMapper(namespace.Name("test-namespace")).
+		Return(namespace.CustomSearchAttributesMapper{}, nil)
+
+	mapperProvider := NewMapperProvider(
+		nil,
+		nsRegistry,
+		staticSearchAttributesProvider{
+			nameTypeMaps: map[string]NameTypeMap{
+				"test-visibility-index": NewNameTypeMap(map[string]enumspb.IndexedValueType{
+					"Keyword41": enumspb.INDEXED_VALUE_TYPE_KEYWORD,
+				}),
+			},
+		},
+		"test-visibility-index",
+	)
+
+	mapper, err := mapperProvider.GetMapper(namespace.Name("test-namespace"))
+	require.NoError(t, err)
+
+	alias, err := mapper.GetAlias("Keyword41", "error-namespace")
+	require.NoError(t, err)
+	require.Equal(t, "Keyword41", alias)
+
+	fieldName, err := mapper.GetFieldName("Keyword41", "error-namespace")
+	require.NoError(t, err)
+	require.Equal(t, "Keyword41", fieldName)
+}

common/searchattribute/test_provider.go

@@ -61,6 +61,10 @@ func TestEsNameTypeMap() NameTypeMap {
 	return NewNameTypeMap(csa)
 }
 
+func TestSearchAttributesToRegister() map[string]enumspb.IndexedValueType {
+	return maps.Clone(esCustomSearchAttributes)
+}
+
 func TestEsNameTypeMapWithScheduleID() NameTypeMap {
 	res := TestEsNameTypeMap()
 	res.customSearchAttributes[sadefs.ScheduleID] = enumspb.INDEXED_VALUE_TYPE_KEYWORD
@@ -134,12 +138,8 @@ func (t *TestMapper) GetFieldName(alias string, namespace string) (string, error
 	return "", serviceerror.NewInvalidArgument("unknown namespace")
 }
 
-func NewNoopMapper() Mapper {
-	return &noopMapper{}
-}
-
 func NewTestMapperProvider(customMapper Mapper) MapperProvider {
-	return NewMapperProvider(customMapper, nil, NewTestProvider(), false)
+	return NewMapperProvider(customMapper, nil, NewTestProvider())
 }
 
 func NewNameTypeMapStub(attributes map[string]enumspb.IndexedValueType) NameTypeMap {

temporal/fx.go

@@ -664,9 +664,7 @@ func ApplyClusterMetadataConfigProvider(
 	}
 	indexSearchAttributes := make(map[string]*persistencespb.IndexSearchAttributes)
 	for _, ds := range visDataStores {
-		if ds.SQL != nil || ds.CustomDataStoreConfig != nil {
-			indexSearchAttributes[ds.GetIndexName()] = sadefs.GetDBIndexSearchAttributes(visCSAOverride)
-		}
+		indexSearchAttributes[ds.GetIndexName()] = sadefs.GetDBIndexSearchAttributes(visCSAOverride)
 	}
 
 	clusterMetadata := svc.ClusterMetadata

tests/advanced_visibility_test.go

@@ -1635,23 +1635,13 @@ func (s *AdvancedVisibilitySuite) TestUpsertWorkflowExecution_InvalidKey() {
 		WorkflowId: id,
 		RunId:      we.RunId,
 	})
-	if !testcore.UseSQLVisibility() {
-		s.ErrorContains(err, "BadSearchAttributes: search attribute INVALIDKEY is not defined")
-		s.EqualHistoryEvents(`
-  1 WorkflowExecutionStarted
-  2 WorkflowTaskScheduled
-  3 WorkflowTaskStarted
-  4 WorkflowTaskFailed {"Cause":23,"Failure":{"Message":"BadSearchAttributes: search attribute INVALIDKEY is not defined"}}
-  5 WorkflowTaskScheduled`, historyEvents)
-	} else {
-		s.ErrorContains(err, fmt.Sprintf("BadSearchAttributes: Namespace %s has no mapping defined for search attribute INVALIDKEY", s.Namespace().String()))
-		s.EqualHistoryEvents(fmt.Sprintf(`
+	s.ErrorContains(err, fmt.Sprintf("BadSearchAttributes: Namespace %s has no mapping defined for search attribute INVALIDKEY", s.Namespace().String()))
+	s.EqualHistoryEvents(fmt.Sprintf(`
   1 WorkflowExecutionStarted
   2 WorkflowTaskScheduled
   3 WorkflowTaskStarted
   4 WorkflowTaskFailed {"Cause":23,"Failure":{"Message":"BadSearchAttributes: Namespace %s has no mapping defined for search attribute INVALIDKEY"}}
   5 WorkflowTaskScheduled`, s.Namespace().String()), historyEvents)
-	}
 }
 
 func (s *AdvancedVisibilitySuite) TestChildWorkflow_ParentWorkflow() {