enable namespace mapper with fallbacks (#9664)

## What changed?
enable namespace mapper with fallbacks to the cluster metadata.

## Why?
Allow backwards compatible rollout of search attribute mapper
unification for ElasticSearch backed visibility stores.

## 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/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,12 @@ func SearchAttributeMapperProviderProvider(
 	searchAttributeProvider searchattribute.Provider,
 	persistenceConfig *config.Persistence,
 ) searchattribute.MapperProvider {
+	primaryVisibilityStoreConfig := persistenceConfig.GetVisibilityStoreConfig()
 	return searchattribute.NewMapperProvider(
 		saMapper,
 		namespaceRegistry,
 		searchAttributeProvider,
-		persistenceConfig.IsSQLVisibilityStore() || persistenceConfig.IsCustomVisibilityStore(),
+		primaryVisibilityStoreConfig.GetIndexName(),
 	)
 }
 

common/searchattribute/mapper.go

@@ -4,8 +4,10 @@ package searchattribute
 
 import (
 	"errors"
+	"fmt"
 
 	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.temporal.io/server/common/searchattribute/sadefs"
@@ -20,101 +22,106 @@ 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
+	// This mapper preserves legacy custom search attribute behavior by falling back
+	// to identity mapping when the wrapped mapper misses but cluster metadata still
+	// recognizes the name as a legacy custom search attribute.
+	backCompMapper struct {
+		mapper              Mapper
+		fallbackNameTypeMap 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
+		fallbackIndexName        string
 	}
 )
 
-var _ Mapper = (*noopMapper)(nil)
-var _ Mapper = (*backCompMapper_v1_20)(nil)
+var _ Mapper = (*backCompMapper)(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) {
+func (m *backCompMapper) 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
+		// this is a custom search attribute registered through cluster metadata.
 		return fieldName, nil
 	}
 	return alias, nil
 }
 
-func (m *backCompMapper_v1_20) GetFieldName(alias string, namespaceName string) (string, error) {
+func (m *backCompMapper) 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
+		// this is a custom search attribute registered through cluster metadata.
 		return alias, nil
 	}
 	return fieldName, nil
 }
 
+func (m *backCompMapper) isLegacyCustomSearchAttribute(name string) bool {
+	_, err := m.fallbackNameTypeMap.getType(name, customCategory)
+	return err == nil
+}
+
 func NewMapperProvider(
 	customMapper Mapper,
 	namespaceRegistry namespace.Registry,
 	searchAttributesProvider Provider,
-	enableMapperFromNamespace bool,
+	fallbackIndexName string,
 ) MapperProvider {
 	return &mapperProviderImpl{
-		customMapper:              customMapper,
-		namespaceRegistry:         namespaceRegistry,
-		searchAttributesProvider:  searchAttributesProvider,
-		enableMapperFromNamespace: enableMapperFromNamespace,
+		customMapper:             customMapper,
+		namespaceRegistry:        namespaceRegistry,
+		searchAttributesProvider: searchAttributesProvider,
+		fallbackIndexName:        fallbackIndexName,
 	}
 }
 
 func (m *mapperProviderImpl) GetMapper(nsName namespace.Name) (Mapper, error) {
 	if m.customMapper != nil {
 		return m.customMapper, nil
 	}
-	if !m.enableMapperFromNamespace {
-		return &noopMapper{}, 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)
-	return &backCompMapper_v1_20{
-		mapper:                 &saMapper,
-		emptyStringNameTypeMap: emptyStringNameTypeMap,
+	fallbackNameTypeMap := NameTypeMap{}
+	if m.fallbackIndexName != "" {
+		nameTypeMap, err := m.searchAttributesProvider.GetSearchAttributes(m.fallbackIndexName, false)
+		if err != nil {
+			return nil, fmt.Errorf("failed to load search attributes for fallback index %q: %w", m.fallbackIndexName, err)
+		}
+		fallbackNameTypeMap = legacyCustomSearchAttributes(nameTypeMap)
+	}
+	return &backCompMapper{
+		mapper:              &saMapper,
+		fallbackNameTypeMap: fallbackNameTypeMap,
 	}, nil
 }
 
+func legacyCustomSearchAttributes(nameTypeMap NameTypeMap) NameTypeMap {
+	legacyCustomSearchAttributes := make(map[string]enumspb.IndexedValueType)
+	for name, valueType := range nameTypeMap.Custom() {
+		if sadefs.IsPreallocatedCSAFieldName(name, valueType) {
+			continue
+		}
+		legacyCustomSearchAttributes[name] = valueType
+	}
+	return NewNameTypeMap(legacyCustomSearchAttributes)
+}
+
 // 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

@@ -1,11 +1,15 @@
 package searchattribute
 
 import (
+	"errors"
 	"testing"
 
 	"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 +141,115 @@ 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
+	err          error
+}
+
+func (s staticSearchAttributesProvider) GetSearchAttributes(indexName string, _ bool) (NameTypeMap, error) {
+	if s.err != nil {
+		return NameTypeMap{}, s.err
+	}
+	if nameTypeMap, ok := s.nameTypeMaps[indexName]; ok {
+		return nameTypeMap, nil
+	}
+	return NameTypeMap{}, nil
+}
+
+func Test_BackCompMapperFallsBackToClusterMetadataFields(t *testing.T) {
+	mapper := &backCompMapper{
+		mapper:              &TestMapper{},
+		fallbackNameTypeMap: 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{
+					"LegacyKeyword": enumspb.INDEXED_VALUE_TYPE_KEYWORD,
+				}),
+			},
+		},
+		"test-visibility-index",
+	)
+
+	mapper, err := mapperProvider.GetMapper(namespace.Name("test-namespace"))
+	require.NoError(t, err)
+
+	alias, err := mapper.GetAlias("LegacyKeyword", "error-namespace")
+	require.NoError(t, err)
+	require.Equal(t, "LegacyKeyword", alias)
+
+	fieldName, err := mapper.GetFieldName("LegacyKeyword", "error-namespace")
+	require.NoError(t, err)
+	require.Equal(t, "LegacyKeyword", fieldName)
+}
+
+func TestMapperProviderDoesNotTreatPreallocatedFieldsAsLegacyCustomAttributes(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": TestNameTypeMap(),
+			},
+		},
+		"test-visibility-index",
+	)
+
+	mapper, err := mapperProvider.GetMapper(namespace.Name("test-namespace"))
+	require.NoError(t, err)
+
+	_, err = mapper.GetFieldName("Text01", "error-namespace")
+	require.Error(t, err)
+	var invalidArgumentErr *serviceerror.InvalidArgument
+	require.ErrorAs(t, err, &invalidArgumentErr)
+}
+
+func TestMapperProviderReturnsFallbackLookupError(t *testing.T) {
+	controller := gomock.NewController(t)
+	nsRegistry := namespace.NewMockRegistry(controller)
+	nsRegistry.EXPECT().
+		GetCustomSearchAttributesMapper(namespace.Name("test-namespace")).
+		Return(namespace.CustomSearchAttributesMapper{}, nil)
+
+	expectedErr := errors.New("boom")
+	mapperProvider := NewMapperProvider(
+		nil,
+		nsRegistry,
+		staticSearchAttributesProvider{
+			err: expectedErr,
+		},
+		"test-visibility-index",
+	)
+
+	_, err := mapperProvider.GetMapper(namespace.Name("test-namespace"))
+	require.Error(t, err)
+	require.ErrorContains(t, err, `failed to load search attributes for fallback index "test-visibility-index"`)
+	require.ErrorContains(t, err, expectedErr.Error())
+}

common/searchattribute/test_provider.go

@@ -7,6 +7,7 @@ import (
 
 	enumspb "go.temporal.io/api/enums/v1"
 	"go.temporal.io/api/serviceerror"
+	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/searchattribute/sadefs"
 )
 
@@ -15,6 +16,10 @@ type (
 		es bool
 	}
 
+	testMapperProvider struct {
+		mapper Mapper
+	}
+
 	TestMapper struct {
 		Namespace            string
 		WithCustomScheduleID bool
@@ -23,6 +28,7 @@ type (
 
 var _ Provider = (*TestProvider)(nil)
 var _ Mapper = (*TestMapper)(nil)
+var _ MapperProvider = (*testMapperProvider)(nil)
 
 var (
 	esCustomSearchAttributes = map[string]enumspb.IndexedValueType{
@@ -61,6 +67,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 +144,12 @@ 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 &testMapperProvider{mapper: customMapper}
 }
 
-func NewTestMapperProvider(customMapper Mapper) MapperProvider {
-	return NewMapperProvider(customMapper, nil, NewTestProvider(), false)
+func (p *testMapperProvider) GetMapper(namespace.Name) (Mapper, error) {
+	return p.mapper, nil
 }
 
 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