mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
visibility(chasm): enable querying w/out Temporal prefix (#8370)
## What changed? When resolving search attributes allow users to query for `TemporalXXX` search attributes w/out the `Temporal` prefix. The server will default to the user namespace search attributes, then system/predefined search attributes, then system/predefined `TemporalXXX` search attributes. ## Why? This will unblock future chasm workstreams and allow users to create search attributes that might conflict with existing system/predefined SAs. ## How did you test it? - [ ] built - [ ] run locally and tested manually - [ ] covered by existing tests - [ ] added new unit test(s) - [ ] added new functional test(s) ## Potential risks ... --------- Co-authored-by: Rodrigo Zhou <[email protected]>
This commit is contained in:
committed by
GitHub
co-authored by
Rodrigo Zhou
parent
d89cc3285b
commit
d3955dfc56
@@ -50,35 +50,13 @@ func NewValuesInterceptor(
|
||||
|
||||
// TODO: this is invoked for non-ES validation code flow. Needs refactoring
|
||||
func (ni *nameInterceptor) Name(name string, usage query.FieldNameUsage) (string, error) {
|
||||
fieldName := name
|
||||
if searchattribute.IsMappable(name) {
|
||||
mapper, err := ni.searchAttributesMapperProvider.GetMapper(ni.namespace)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
if mapper != nil {
|
||||
fieldName, err = mapper.GetFieldName(name, ni.namespace.String())
|
||||
if err != nil {
|
||||
if name != searchattribute.ScheduleID {
|
||||
return "", err
|
||||
}
|
||||
|
||||
// ScheduleId is a fake SA -- convert to WorkflowId
|
||||
fieldName = searchattribute.WorkflowID
|
||||
} else if name == searchattribute.ScheduleID && name == fieldName {
|
||||
_, isCustom := ni.searchAttributesTypeMap.Custom()[fieldName]
|
||||
if !isCustom {
|
||||
// ScheduleId is a fake SA -- convert to WorkflowId
|
||||
fieldName = searchattribute.WorkflowID
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fieldType, err := ni.searchAttributesTypeMap.GetType(fieldName)
|
||||
mapper, err := ni.searchAttributesMapperProvider.GetMapper(ni.namespace)
|
||||
if err != nil {
|
||||
return "", query.NewConverterError("invalid search attribute: %s", name)
|
||||
return "", err
|
||||
}
|
||||
fieldName, fieldType, err := query.ResolveSearchAttributeAlias(name, ni.namespace, mapper, ni.searchAttributesTypeMap)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
switch usage {
|
||||
|
||||
@@ -408,15 +408,15 @@ func (s *ESVisibilitySuite) Test_convertQuery_Mapper() {
|
||||
s.Nil(queryParams.Sorter)
|
||||
|
||||
query = `CustomKeywordField = 'pid'`
|
||||
_, err = s.visibilityStore.convertQuery(testNamespace, testNamespaceID, query)
|
||||
s.Error(err)
|
||||
var invalidArgumentErr *serviceerror.InvalidArgument
|
||||
s.ErrorAs(err, &invalidArgumentErr)
|
||||
s.EqualError(err, "mapper error")
|
||||
queryParams, err = s.visibilityStore.convertQuery(testNamespace, testNamespaceID, query)
|
||||
s.NoError(err)
|
||||
s.JSONEq(`{"bool":{"filter":[{"term":{"NamespaceId":"bfd5c907-f899-4baf-a7b2-2ab85e623ebd"}},{"bool":{"filter":{"term":{"CustomKeywordField":"pid"}}}}],"must_not":{"exists":{"field":"TemporalNamespaceDivision"}}}}`, s.queryToJSON(queryParams.Query))
|
||||
s.Nil(queryParams.Sorter)
|
||||
|
||||
query = `AliasForUnknownField = 'pid'`
|
||||
_, err = s.visibilityStore.convertQuery(testNamespace, testNamespaceID, query)
|
||||
s.Error(err)
|
||||
var invalidArgumentErr *serviceerror.InvalidArgument
|
||||
s.ErrorAs(err, &invalidArgumentErr)
|
||||
s.EqualError(err, "invalid query: unable to convert filter expression: unable to convert left side of \"AliasForUnknownField = 'pid'\": invalid search attribute: AliasForUnknownField")
|
||||
|
||||
@@ -433,10 +433,10 @@ func (s *ESVisibilitySuite) Test_convertQuery_Mapper() {
|
||||
s.Equal(`[{"CustomKeywordField":{"order":"asc"}}]`, s.sorterToJSON(queryParams.Sorter))
|
||||
|
||||
query = `order by CustomKeywordField asc`
|
||||
_, err = s.visibilityStore.convertQuery(testNamespace, testNamespaceID, query)
|
||||
s.Error(err)
|
||||
s.ErrorAs(err, &invalidArgumentErr)
|
||||
s.EqualError(err, "mapper error")
|
||||
queryParams, err = s.visibilityStore.convertQuery(testNamespace, testNamespaceID, query)
|
||||
s.NoError(err)
|
||||
s.JSONEq(`{"bool":{"filter":{"term":{"NamespaceId":"bfd5c907-f899-4baf-a7b2-2ab85e623ebd"}},"must_not":{"exists":{"field":"TemporalNamespaceDivision"}}}}`, s.queryToJSON(queryParams.Query))
|
||||
s.NotNil(queryParams.Sorter)
|
||||
|
||||
query = `order by AliasForUnknownField asc`
|
||||
_, err = s.visibilityStore.convertQuery(testNamespace, testNamespaceID, query)
|
||||
@@ -463,7 +463,7 @@ func (s *ESVisibilitySuite) Test_convertQuery_Mapper_Error() {
|
||||
s.Error(err)
|
||||
var invalidArgumentErr *serviceerror.InvalidArgument
|
||||
s.ErrorAs(err, &invalidArgumentErr)
|
||||
s.EqualError(err, "mapper error")
|
||||
s.EqualError(err, "invalid query: unable to convert filter expression: unable to convert left side of \"ProductId = 'pid'\": invalid search attribute: ProductId")
|
||||
|
||||
query = `order by ExecutionTime`
|
||||
queryParams, err = s.visibilityStore.convertQuery(testNamespace, testNamespaceID, query)
|
||||
@@ -471,12 +471,6 @@ func (s *ESVisibilitySuite) Test_convertQuery_Mapper_Error() {
|
||||
s.Equal(`{"bool":{"filter":{"term":{"NamespaceId":"bfd5c907-f899-4baf-a7b2-2ab85e623ebd"}},"must_not":{"exists":{"field":"TemporalNamespaceDivision"}}}}`, s.queryToJSON(queryParams.Query))
|
||||
s.Equal(`[{"ExecutionTime":{"order":"asc"}}]`, s.sorterToJSON(queryParams.Sorter))
|
||||
|
||||
query = `order by CustomIntField asc`
|
||||
_, err = s.visibilityStore.convertQuery(testNamespace, testNamespaceID, query)
|
||||
s.Error(err)
|
||||
s.ErrorAs(err, &invalidArgumentErr)
|
||||
s.EqualError(err, "mapper error")
|
||||
|
||||
s.visibilityStore.searchAttributesMapperProvider = nil
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,93 @@
|
||||
package query
|
||||
|
||||
import (
|
||||
"strings"
|
||||
|
||||
enumspb "go.temporal.io/api/enums/v1"
|
||||
"go.temporal.io/server/common/namespace"
|
||||
"go.temporal.io/server/common/searchattribute"
|
||||
)
|
||||
|
||||
// ResolveSearchAttributeAlias resolves the search attribute alias for the given name. The process is:
|
||||
// 1. If the name has the "Temporal", skip mapping to a custom search attribute.
|
||||
// 2. If the search attribute exists in the visibility mapper, pass it through.
|
||||
// 3. If it exists as a system / predefined attribute, map it.
|
||||
// 3.1 Some pre-defined attributes are already defined with the Temporal prefix, so both options need to be checked.
|
||||
// 4. In the future, also need to lookup in the CHASM archetype search attribute mapping
|
||||
func ResolveSearchAttributeAlias(
|
||||
name string,
|
||||
ns namespace.Name,
|
||||
mapper searchattribute.Mapper,
|
||||
saTypeMap searchattribute.NameTypeMap,
|
||||
) (string, enumspb.IndexedValueType, error) {
|
||||
if searchattribute.IsMappable(name) {
|
||||
// First check if the visibility mapper can handle this field (e.g., custom search attributes)
|
||||
fieldName, fieldType := tryVisibilityMapper(name, ns, mapper, saTypeMap)
|
||||
if fieldName != "" {
|
||||
return fieldName, fieldType, nil
|
||||
}
|
||||
|
||||
// Handle ScheduleID → WorkflowID transformation, but only if ScheduleID is not defined as a custom search attribute
|
||||
// This fallback only applies when the visibility mapper doesn't handle the field
|
||||
if name == searchattribute.ScheduleID {
|
||||
// ScheduleID is not defined, transform to WorkflowID
|
||||
saType, _ := saTypeMap.GetType(searchattribute.WorkflowID)
|
||||
return searchattribute.WorkflowID, saType, nil
|
||||
}
|
||||
}
|
||||
|
||||
fieldName, fieldType, found := tryDirectAndPrefixedLookup(name, saTypeMap)
|
||||
if found {
|
||||
return fieldName, fieldType, nil
|
||||
}
|
||||
|
||||
return "", enumspb.INDEXED_VALUE_TYPE_UNSPECIFIED, NewConverterError("invalid search attribute: %s", name)
|
||||
}
|
||||
|
||||
// tryVisibilityMapper might find a successful match in which case we return it
|
||||
// otherwise we continue with the fallback logic.
|
||||
func tryVisibilityMapper(
|
||||
name string,
|
||||
ns namespace.Name,
|
||||
mapper searchattribute.Mapper,
|
||||
saTypeMap searchattribute.NameTypeMap,
|
||||
) (string, enumspb.IndexedValueType) {
|
||||
if mapper == nil {
|
||||
return "", enumspb.INDEXED_VALUE_TYPE_UNSPECIFIED
|
||||
}
|
||||
|
||||
fieldName, err := mapper.GetFieldName(name, ns.String())
|
||||
if err != nil {
|
||||
// If there is an error, we need to continue with the fallback logic
|
||||
// because this search attribute is not defined in the mapper, but might
|
||||
// exist with a different name in the namespace.
|
||||
return "", enumspb.INDEXED_VALUE_TYPE_UNSPECIFIED
|
||||
}
|
||||
|
||||
// Mapper successfully resolved the field name, now check if it exists in the type map
|
||||
fieldType, err := saTypeMap.GetType(fieldName)
|
||||
if err != nil {
|
||||
// If the mapped field doesn't exist in type map, allow fallback to direct/prefixed lookup.
|
||||
return "", enumspb.INDEXED_VALUE_TYPE_UNSPECIFIED
|
||||
}
|
||||
|
||||
return fieldName, fieldType
|
||||
}
|
||||
|
||||
func tryDirectAndPrefixedLookup(name string, saTypeMap searchattribute.NameTypeMap) (string, enumspb.IndexedValueType, bool) {
|
||||
if saType, err := saTypeMap.GetType(name); err == nil {
|
||||
return name, saType, true
|
||||
}
|
||||
|
||||
prefixedName := searchattribute.ReservedPrefix + name
|
||||
if saType, err := saTypeMap.GetType(prefixedName); err == nil {
|
||||
return prefixedName, saType, true
|
||||
}
|
||||
|
||||
strippedName := strings.TrimPrefix(name, searchattribute.ReservedPrefix)
|
||||
if saType, err := saTypeMap.GetType(strippedName); err == nil {
|
||||
return strippedName, saType, true
|
||||
}
|
||||
|
||||
return "", enumspb.INDEXED_VALUE_TYPE_UNSPECIFIED, false
|
||||
}
|
||||
@@ -0,0 +1,103 @@
|
||||
package query
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
enumspb "go.temporal.io/api/enums/v1"
|
||||
"go.temporal.io/api/serviceerror"
|
||||
"go.temporal.io/server/common/namespace"
|
||||
"go.temporal.io/server/common/searchattribute"
|
||||
)
|
||||
|
||||
func TestResolveSearchAttributeAlias(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
expectedFieldName string
|
||||
expectedFieldType enumspb.IndexedValueType
|
||||
expectedErr bool
|
||||
}{
|
||||
{name: "MyCustomField", expectedFieldName: "MyCustomField", expectedFieldType: enumspb.INDEXED_VALUE_TYPE_KEYWORD, expectedErr: false},
|
||||
{name: "ExecutionStatus", expectedFieldName: "ExecutionStatus", expectedFieldType: enumspb.INDEXED_VALUE_TYPE_KEYWORD, expectedErr: false},
|
||||
{name: "ScheduledStartTime", expectedFieldName: "TemporalScheduledStartTime", expectedFieldType: enumspb.INDEXED_VALUE_TYPE_DATETIME, expectedErr: false},
|
||||
{name: "SchedulePaused", expectedFieldName: "TemporalSchedulePaused", expectedFieldType: enumspb.INDEXED_VALUE_TYPE_BOOL, expectedErr: false},
|
||||
{name: "BuildIds", expectedFieldName: "BuildIds", expectedFieldType: enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST, expectedErr: false},
|
||||
{name: "TemporalBuildIds", expectedFieldName: "BuildIds", expectedFieldType: enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST, expectedErr: false},
|
||||
{name: "TemporalPauseInfo", expectedFieldName: "TemporalPauseInfo", expectedFieldType: enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST, expectedErr: false},
|
||||
{name: "NonExistentField", expectedFieldName: "", expectedFieldType: enumspb.INDEXED_VALUE_TYPE_UNSPECIFIED, expectedErr: true},
|
||||
{name: searchattribute.ScheduleID, expectedFieldName: searchattribute.WorkflowID, expectedFieldType: enumspb.INDEXED_VALUE_TYPE_KEYWORD, expectedErr: false},
|
||||
}
|
||||
|
||||
ns := namespace.Name("test-namespace")
|
||||
saTypeMap := searchattribute.NewNameTypeMapStub(map[string]enumspb.IndexedValueType{
|
||||
"MyCustomField": enumspb.INDEXED_VALUE_TYPE_KEYWORD,
|
||||
"ExecutionStatus": enumspb.INDEXED_VALUE_TYPE_KEYWORD,
|
||||
})
|
||||
mapper := customMapper{
|
||||
fieldToAlias: map[string]string{
|
||||
"MyCustomField": "some-alias",
|
||||
"ExecutionStatus": "some-alias2",
|
||||
},
|
||||
aliasToField: map[string]string{
|
||||
"some-alias": "MyCustomField",
|
||||
"some-alias2": "ExecutionStatus",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
fieldName, fieldType, err := ResolveSearchAttributeAlias(tc.name, ns, mapper, saTypeMap)
|
||||
require.Equal(t, tc.expectedFieldName, fieldName)
|
||||
require.Equal(t, tc.expectedFieldType, fieldType)
|
||||
require.Equal(t, tc.expectedErr, err != nil)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveSearchAttributeAlias_CustomScheduleID(t *testing.T) {
|
||||
ns := namespace.Name("test-namespace")
|
||||
|
||||
// Test case where ScheduleID is defined as a custom search attribute
|
||||
saTypeMapWithCustomScheduleID := searchattribute.NewNameTypeMapStub(map[string]enumspb.IndexedValueType{
|
||||
searchattribute.ScheduleID: enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST,
|
||||
})
|
||||
|
||||
mapper := customMapper{
|
||||
fieldToAlias: map[string]string{
|
||||
searchattribute.ScheduleID: searchattribute.ScheduleID,
|
||||
},
|
||||
aliasToField: map[string]string{
|
||||
searchattribute.ScheduleID: searchattribute.ScheduleID,
|
||||
},
|
||||
}
|
||||
|
||||
// When ScheduleID is a custom search attribute, it should use the custom attribute, not transform to WorkflowID
|
||||
fieldName, fieldType, err := ResolveSearchAttributeAlias(searchattribute.ScheduleID, ns, mapper, saTypeMapWithCustomScheduleID)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, searchattribute.ScheduleID, fieldName)
|
||||
require.Equal(t, enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST, fieldType)
|
||||
}
|
||||
|
||||
type customMapper struct {
|
||||
fieldToAlias map[string]string
|
||||
aliasToField map[string]string
|
||||
}
|
||||
|
||||
func (m customMapper) GetAlias(fieldName string, ns string) (string, error) {
|
||||
if fn, ok := m.fieldToAlias[fieldName]; ok {
|
||||
return fn, nil
|
||||
}
|
||||
return "", serviceerror.NewInvalidArgument(
|
||||
fmt.Sprintf("Namespace %s has no mapping defined for field name %s", ns, fieldName),
|
||||
)
|
||||
}
|
||||
|
||||
func (m customMapper) GetFieldName(alias string, ns string) (string, error) {
|
||||
if fn, ok := m.aliasToField[alias]; ok {
|
||||
return fn, nil
|
||||
}
|
||||
return "", serviceerror.NewInvalidArgument(
|
||||
fmt.Sprintf("Namespace %s has no mapping defined for search attribute %s", ns, alias),
|
||||
)
|
||||
}
|
||||
@@ -441,26 +441,8 @@ func (c *QueryConverter) convertColName(exprRef *sqlparser.Expr) (*saColName, er
|
||||
)
|
||||
}
|
||||
saAlias := strings.ReplaceAll(sqlparser.String(expr), "`", "")
|
||||
saFieldName := saAlias
|
||||
if searchattribute.IsMappable(saAlias) {
|
||||
var err error
|
||||
saFieldName, err = c.saMapper.GetFieldName(saAlias, c.namespaceName.String())
|
||||
if err != nil {
|
||||
if saAlias != searchattribute.ScheduleID {
|
||||
return nil, query.NewConverterError(
|
||||
"%s: column name '%s' is not a valid search attribute",
|
||||
query.InvalidExpressionErrMessage,
|
||||
saAlias,
|
||||
)
|
||||
}
|
||||
// ScheduleId is a fake SA -- convert to WorkflowId
|
||||
saFieldName = searchattribute.WorkflowID
|
||||
}
|
||||
}
|
||||
|
||||
saType, err := c.saTypeMap.GetType(saFieldName)
|
||||
saFieldName, saType, err := query.ResolveSearchAttributeAlias(saAlias, c.namespaceName, c.saMapper, c.saTypeMap)
|
||||
if err != nil {
|
||||
// This should never happen since it came from mapping.
|
||||
return nil, query.NewConverterError(
|
||||
"%s: column name '%s' is not a valid search attribute",
|
||||
query.InvalidExpressionErrMessage,
|
||||
|
||||
@@ -667,7 +667,7 @@ func (s *queryConverterSuite) TestConvertColName() {
|
||||
}
|
||||
} else {
|
||||
s.Error(err)
|
||||
s.Equal(err, tc.err)
|
||||
s.Equal(tc.err, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -235,7 +235,8 @@ func IsReserved(name string) bool {
|
||||
return strings.HasPrefix(name, ReservedPrefix)
|
||||
}
|
||||
|
||||
// IsMappable returns true if name can have be mapped tho the alias.
|
||||
// IsMappable returns true if name can be mapped to the alias.
|
||||
// Mappable search attributes are those that can be defined by the user.
|
||||
func IsMappable(name string) bool {
|
||||
if _, ok := system[name]; ok {
|
||||
return false
|
||||
|
||||
@@ -3,6 +3,8 @@
|
||||
package searchattribute
|
||||
|
||||
import (
|
||||
"errors"
|
||||
|
||||
commonpb "go.temporal.io/api/common/v1"
|
||||
"go.temporal.io/api/serviceerror"
|
||||
"go.temporal.io/server/common/namespace"
|
||||
@@ -138,10 +140,11 @@ func AliasFields(
|
||||
|
||||
aliasName, err := mapper.GetAlias(saName, namespaceName)
|
||||
if err != nil {
|
||||
if _, isInvalidArgument := err.(*serviceerror.InvalidArgument); isInvalidArgument {
|
||||
// Silently ignore serviceerror.InvalidArgument because it indicates unmapped field (alias was deleted, for example).
|
||||
// IMPORTANT: AliasFields should never return serviceerror.InvalidArgument because it is used by Poll API and the error
|
||||
// goes through up to SDK, which shutdowns worker when it receives serviceerror.InvalidArgument as poll response.
|
||||
// Silently ignore serviceerror.InvalidArgument because it indicates unmapped field (alias was deleted, for example).
|
||||
// IMPORTANT: AliasFields should never return serviceerror.InvalidArgument because it is used by Poll API and the error
|
||||
// goes through up to SDK, which shutdowns worker when it receives serviceerror.InvalidArgument as poll response.
|
||||
var invalidArgumentErr *serviceerror.InvalidArgument
|
||||
if errors.As(err, &invalidArgumentErr) {
|
||||
continue
|
||||
}
|
||||
return nil, err
|
||||
|
||||
@@ -3,7 +3,7 @@ package searchattribute
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
commonpb "go.temporal.io/api/common/v1"
|
||||
"go.temporal.io/api/serviceerror"
|
||||
)
|
||||
@@ -17,10 +17,9 @@ func Test_AliasFields(t *testing.T) {
|
||||
"wrong_field": {Data: []byte("data23")}, // Wrong unknown name must be ignored.
|
||||
},
|
||||
}
|
||||
_, err := AliasFields(mapperProvider, sa, "error-namespace")
|
||||
assert.Error(t, err)
|
||||
var internalErr *serviceerror.Internal
|
||||
assert.ErrorAs(t, err, &internalErr)
|
||||
sb, err := AliasFields(mapperProvider, sa, "error-namespace")
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, sa, sb)
|
||||
|
||||
sa = &commonpb.SearchAttributes{
|
||||
IndexedFields: map[string]*commonpb.Payload{
|
||||
@@ -28,9 +27,9 @@ func Test_AliasFields(t *testing.T) {
|
||||
"wrong_field": {Data: []byte("data23")}, // Wrong unknown name must be ignored.
|
||||
},
|
||||
}
|
||||
sb, err := AliasFields(mapperProvider, sa, "unknown-namespace")
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, sa, sb)
|
||||
sb, err = AliasFields(mapperProvider, sa, "unknown-namespace")
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, sa, sb)
|
||||
|
||||
sa = &commonpb.SearchAttributes{
|
||||
IndexedFields: map[string]*commonpb.Payload{
|
||||
@@ -40,22 +39,22 @@ func Test_AliasFields(t *testing.T) {
|
||||
},
|
||||
}
|
||||
sb, err = AliasFields(mapperProvider, sa, "test-namespace")
|
||||
assert.NoError(t, err)
|
||||
assert.NotEqual(t, sa, sb)
|
||||
assert.Len(t, sb.GetIndexedFields(), 2)
|
||||
assert.EqualValues(t, "data1", sb.GetIndexedFields()["AliasForField1"].GetData())
|
||||
assert.EqualValues(t, "data2", sb.GetIndexedFields()["AliasForField2"].GetData())
|
||||
require.NoError(t, err)
|
||||
require.NotEqual(t, sa, sb)
|
||||
require.Len(t, sb.GetIndexedFields(), 2)
|
||||
require.EqualValues(t, "data1", sb.GetIndexedFields()["AliasForField1"].GetData())
|
||||
require.EqualValues(t, "data2", sb.GetIndexedFields()["AliasForField2"].GetData())
|
||||
|
||||
// Empty search attributes are not validated with mapper.
|
||||
sa = &commonpb.SearchAttributes{
|
||||
IndexedFields: nil,
|
||||
}
|
||||
sb, err = AliasFields(mapperProvider, sa, "error-namespace")
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, sa, sb)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, sa, sb)
|
||||
sb, err = AliasFields(mapperProvider, sa, "unknown-namespace")
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, sa, sb)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, sa, sb)
|
||||
|
||||
// Pass through search attributes are not mapped.
|
||||
sa = &commonpb.SearchAttributes{
|
||||
@@ -64,8 +63,8 @@ func Test_AliasFields(t *testing.T) {
|
||||
},
|
||||
}
|
||||
sb, err = AliasFields(mapperProvider, sa, "test-namespace")
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, sa, sb)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, sa, sb)
|
||||
}
|
||||
|
||||
func Test_UnaliasFields(t *testing.T) {
|
||||
@@ -77,9 +76,9 @@ func Test_UnaliasFields(t *testing.T) {
|
||||
},
|
||||
}
|
||||
_, err := UnaliasFields(mapperProvider, sa, "error-namespace")
|
||||
assert.Error(t, err)
|
||||
var internalErr *serviceerror.Internal
|
||||
assert.ErrorAs(t, err, &internalErr)
|
||||
require.Error(t, err)
|
||||
var invalidArgumentErr2 *serviceerror.InvalidArgument
|
||||
require.ErrorAs(t, err, &invalidArgumentErr2)
|
||||
|
||||
sa = &commonpb.SearchAttributes{
|
||||
IndexedFields: map[string]*commonpb.Payload{
|
||||
@@ -88,9 +87,9 @@ func Test_UnaliasFields(t *testing.T) {
|
||||
},
|
||||
}
|
||||
_, err = UnaliasFields(mapperProvider, sa, "unknown-namespace")
|
||||
assert.Error(t, err)
|
||||
require.Error(t, err)
|
||||
var invalidArgumentErr *serviceerror.InvalidArgument
|
||||
assert.ErrorAs(t, err, &invalidArgumentErr)
|
||||
require.ErrorAs(t, err, &invalidArgumentErr)
|
||||
|
||||
sa = &commonpb.SearchAttributes{
|
||||
IndexedFields: map[string]*commonpb.Payload{
|
||||
@@ -99,11 +98,11 @@ func Test_UnaliasFields(t *testing.T) {
|
||||
},
|
||||
}
|
||||
sa, err = UnaliasFields(mapperProvider, sa, "test-namespace")
|
||||
assert.NoError(t, err)
|
||||
assert.NotNil(t, sa)
|
||||
assert.Len(t, sa.GetIndexedFields(), 2)
|
||||
assert.EqualValues(t, "data1", sa.GetIndexedFields()["Field1"].GetData())
|
||||
assert.EqualValues(t, "data2", sa.GetIndexedFields()["Field2"].GetData())
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, sa)
|
||||
require.Len(t, sa.GetIndexedFields(), 2)
|
||||
require.EqualValues(t, "data1", sa.GetIndexedFields()["Field1"].GetData())
|
||||
require.EqualValues(t, "data2", sa.GetIndexedFields()["Field2"].GetData())
|
||||
|
||||
sa = &commonpb.SearchAttributes{
|
||||
IndexedFields: map[string]*commonpb.Payload{
|
||||
@@ -113,20 +112,20 @@ func Test_UnaliasFields(t *testing.T) {
|
||||
},
|
||||
}
|
||||
_, err = UnaliasFields(mapperProvider, sa, "test-namespace")
|
||||
assert.Error(t, err)
|
||||
assert.ErrorAs(t, err, &invalidArgumentErr)
|
||||
require.Error(t, err)
|
||||
require.ErrorAs(t, err, &invalidArgumentErr)
|
||||
|
||||
// Empty search attributes are not validated with mapper.
|
||||
sa = &commonpb.SearchAttributes{
|
||||
IndexedFields: nil,
|
||||
}
|
||||
sb, err := UnaliasFields(mapperProvider, sa, "error-namespace")
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, sa, sb, "when there is nothin to unalias should return received attributes")
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, sa, sb, "when there is nothin to unalias should return received attributes")
|
||||
|
||||
sb, err = UnaliasFields(mapperProvider, sa, "unknown-namespace")
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, sa, sb, "when there is nothin to unalias should return received attributes")
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, sa, sb, "when there is nothin to unalias should return received attributes")
|
||||
|
||||
// Pass through aliases are not substituted.
|
||||
sa = &commonpb.SearchAttributes{
|
||||
@@ -135,6 +134,6 @@ func Test_UnaliasFields(t *testing.T) {
|
||||
},
|
||||
}
|
||||
sb, err = UnaliasFields(mapperProvider, sa, "test-namespace")
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, sb, sb, "when there is nothin to unalias should return received attributes")
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, sa, sb, "when there is nothin to unalias should return received attributes")
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ package searchattribute
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"maps"
|
||||
|
||||
enumspb "go.temporal.io/api/enums/v1"
|
||||
persistencespb "go.temporal.io/server/api/persistence/v1"
|
||||
@@ -34,12 +35,8 @@ func buildIndexNameTypeMap(indexSearchAttributes map[string]*persistencespb.Inde
|
||||
|
||||
func (m NameTypeMap) System() map[string]enumspb.IndexedValueType {
|
||||
allSystem := make(map[string]enumspb.IndexedValueType, len(system)+len(predefined))
|
||||
for saName, saType := range system {
|
||||
allSystem[saName] = saType
|
||||
}
|
||||
for saName, saType := range predefined {
|
||||
allSystem[saName] = saType
|
||||
}
|
||||
maps.Copy(allSystem, system)
|
||||
maps.Copy(allSystem, predefined)
|
||||
return allSystem
|
||||
}
|
||||
|
||||
@@ -49,15 +46,9 @@ func (m NameTypeMap) Custom() map[string]enumspb.IndexedValueType {
|
||||
|
||||
func (m NameTypeMap) All() map[string]enumspb.IndexedValueType {
|
||||
allSearchAttributes := make(map[string]enumspb.IndexedValueType, len(system)+len(m.customSearchAttributes)+len(predefined))
|
||||
for saName, saType := range system {
|
||||
allSearchAttributes[saName] = saType
|
||||
}
|
||||
for saName, saType := range predefined {
|
||||
allSearchAttributes[saName] = saType
|
||||
}
|
||||
for saName, saType := range m.customSearchAttributes {
|
||||
allSearchAttributes[saName] = saType
|
||||
}
|
||||
maps.Copy(allSearchAttributes, system)
|
||||
maps.Copy(allSearchAttributes, predefined)
|
||||
maps.Copy(allSearchAttributes, m.customSearchAttributes)
|
||||
return allSearchAttributes
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package searchattribute
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
enumspb "go.temporal.io/api/enums/v1"
|
||||
@@ -91,7 +92,9 @@ func (t *TestMapper) GetAlias(fieldName string, namespace string) (string, error
|
||||
return "", serviceerror.NewInvalidArgument("unmapped field")
|
||||
}
|
||||
if namespace == "error-namespace" {
|
||||
return "", serviceerror.NewInternal("mapper error")
|
||||
return "", serviceerror.NewInvalidArgument(
|
||||
fmt.Sprintf("Namespace %s has no mapping defined for field name %s", namespace, fieldName),
|
||||
)
|
||||
}
|
||||
if namespace == "test-namespace" || namespace == t.Namespace {
|
||||
if fieldName == "pass-through" {
|
||||
@@ -110,7 +113,9 @@ func (t *TestMapper) GetFieldName(alias string, namespace string) (string, error
|
||||
return "", serviceerror.NewInvalidArgument("unmapped alias")
|
||||
}
|
||||
if namespace == "error-namespace" {
|
||||
return "", serviceerror.NewInternal("mapper error")
|
||||
return "", serviceerror.NewInvalidArgument(
|
||||
fmt.Sprintf("Namespace %s has no mapping defined for search attribute %s", namespace, alias),
|
||||
)
|
||||
} else if namespace == "test-namespace" || namespace == t.Namespace {
|
||||
if alias == "pass-through" {
|
||||
return alias, nil
|
||||
|
||||
@@ -209,7 +209,7 @@ func (s *searchAttributesValidatorSuite) TestSearchAttributesValidate_Mapper() {
|
||||
|
||||
err = saValidator.Validate(attr, "error-namespace")
|
||||
s.Error(err)
|
||||
s.EqualError(err, "mapper error")
|
||||
s.Require().EqualError(err, "Namespace error-namespace has no mapping defined for field name InvalidKey")
|
||||
|
||||
fields = map[string]*commonpb.Payload{
|
||||
"CustomTextField": payload.EncodeString("1"),
|
||||
|
||||
@@ -0,0 +1,162 @@
|
||||
package tests
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"github.com/stretchr/testify/suite"
|
||||
commonpb "go.temporal.io/api/common/v1"
|
||||
enumspb "go.temporal.io/api/enums/v1"
|
||||
"go.temporal.io/api/operatorservice/v1"
|
||||
"go.temporal.io/api/workflowservice/v1"
|
||||
"go.temporal.io/sdk/workflow"
|
||||
"go.temporal.io/server/common/payload"
|
||||
"go.temporal.io/server/common/searchattribute"
|
||||
"go.temporal.io/server/common/testing/testvars"
|
||||
"go.temporal.io/server/tests/testcore"
|
||||
)
|
||||
|
||||
type WorkflowAliasSearchAttributeTestSuite struct {
|
||||
testcore.FunctionalTestBase
|
||||
}
|
||||
|
||||
func TestWorkflowAliasSearchAttributeTestSuite(t *testing.T) {
|
||||
s := new(WorkflowAliasSearchAttributeTestSuite)
|
||||
suite.Run(t, s)
|
||||
}
|
||||
|
||||
func (s *WorkflowAliasSearchAttributeTestSuite) SetupTest() {
|
||||
s.FunctionalTestBase.SetupTest()
|
||||
|
||||
s.Worker().RegisterWorkflow(s.workflowFunc)
|
||||
}
|
||||
|
||||
func (s *WorkflowAliasSearchAttributeTestSuite) workflowFunc(ctx workflow.Context) (string, error) {
|
||||
return "done!", nil
|
||||
}
|
||||
|
||||
func (s *WorkflowAliasSearchAttributeTestSuite) createWorkflow(
|
||||
ctx context.Context,
|
||||
tv *testvars.TestVars,
|
||||
sa *commonpb.SearchAttributes,
|
||||
) (*workflowservice.StartWorkflowExecutionResponse, error) {
|
||||
request := &workflowservice.StartWorkflowExecutionRequest{
|
||||
RequestId: tv.Any().String(),
|
||||
Namespace: s.Namespace().String(),
|
||||
WorkflowId: tv.WorkflowID(),
|
||||
WorkflowType: tv.WorkflowType(),
|
||||
TaskQueue: tv.TaskQueue(),
|
||||
Identity: tv.WorkerIdentity(),
|
||||
VersioningOverride: tv.VersioningOverridePinned(true),
|
||||
SearchAttributes: sa,
|
||||
}
|
||||
return s.FrontendClient().StartWorkflowExecution(ctx, request)
|
||||
}
|
||||
|
||||
func (s *WorkflowAliasSearchAttributeTestSuite) terminateWorkflow(
|
||||
ctx context.Context,
|
||||
tv *testvars.TestVars,
|
||||
) (*workflowservice.TerminateWorkflowExecutionResponse, error) {
|
||||
return s.FrontendClient().TerminateWorkflowExecution(ctx, &workflowservice.TerminateWorkflowExecutionRequest{
|
||||
Namespace: s.Namespace().String(),
|
||||
WorkflowExecution: &commonpb.WorkflowExecution{
|
||||
WorkflowId: tv.WorkflowID(),
|
||||
},
|
||||
Reason: "terminate reason",
|
||||
})
|
||||
}
|
||||
|
||||
func (s *WorkflowAliasSearchAttributeTestSuite) TestWorkflowAliasSearchAttribute() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
tv := testvars.New(s.T())
|
||||
|
||||
_, err := s.createWorkflow(ctx, tv, nil)
|
||||
s.Require().NoError(err)
|
||||
|
||||
s.EventuallyWithT(
|
||||
func(t *assert.CollectT) {
|
||||
resp, err := s.SdkClient().ListWorkflow(ctx, &workflowservice.ListWorkflowExecutionsRequest{
|
||||
Namespace: s.Namespace().String(),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, resp)
|
||||
require.Len(t, resp.GetExecutions(), 1)
|
||||
|
||||
queriedResp, err := s.SdkClient().ListWorkflow(ctx, &workflowservice.ListWorkflowExecutionsRequest{
|
||||
Namespace: s.Namespace().String(),
|
||||
Query: fmt.Sprintf("%s = 'Pinned'", searchattribute.TemporalWorkflowVersioningBehavior),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, resp)
|
||||
require.Len(t, queriedResp.GetExecutions(), 1)
|
||||
|
||||
queriedResp, err = s.SdkClient().ListWorkflow(ctx, &workflowservice.ListWorkflowExecutionsRequest{
|
||||
Namespace: s.Namespace().String(),
|
||||
Query: "WorkflowVersioningBehavior = 'Pinned'",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, resp)
|
||||
require.Len(t, queriedResp.GetExecutions(), 1)
|
||||
},
|
||||
testcore.WaitForESToSettle,
|
||||
100*time.Millisecond,
|
||||
)
|
||||
|
||||
_, err = s.terminateWorkflow(ctx, tv)
|
||||
s.Require().NoError(err)
|
||||
}
|
||||
|
||||
func (s *WorkflowAliasSearchAttributeTestSuite) TestWorkflowAliasSearchAttribute_CustomSearchAttributeOverride() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
tv := testvars.New(s.T())
|
||||
|
||||
_, err := s.SdkClient().OperatorService().AddSearchAttributes(ctx, &operatorservice.AddSearchAttributesRequest{
|
||||
Namespace: s.Namespace().String(),
|
||||
SearchAttributes: map[string]enumspb.IndexedValueType{
|
||||
"WorkflowVersioningBehavior": enumspb.INDEXED_VALUE_TYPE_KEYWORD,
|
||||
},
|
||||
})
|
||||
s.Require().NoError(err)
|
||||
|
||||
sa := &commonpb.SearchAttributes{
|
||||
IndexedFields: map[string]*commonpb.Payload{
|
||||
"WorkflowVersioningBehavior": payload.EncodeString("user-defined"),
|
||||
},
|
||||
}
|
||||
|
||||
_, err = s.createWorkflow(ctx, tv, sa)
|
||||
s.Require().NoError(err)
|
||||
|
||||
s.EventuallyWithT(
|
||||
func(t *assert.CollectT) {
|
||||
queriedResp, err := s.SdkClient().ListWorkflow(ctx, &workflowservice.ListWorkflowExecutionsRequest{
|
||||
Namespace: s.Namespace().String(),
|
||||
Query: fmt.Sprintf("%s = 'Pinned'", searchattribute.TemporalWorkflowVersioningBehavior),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, queriedResp)
|
||||
require.Len(t, queriedResp.GetExecutions(), 1)
|
||||
|
||||
queriedResp, err = s.SdkClient().ListWorkflow(ctx, &workflowservice.ListWorkflowExecutionsRequest{
|
||||
Namespace: s.Namespace().String(),
|
||||
Query: "WorkflowVersioningBehavior = 'user-defined'",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, queriedResp)
|
||||
require.Len(t, queriedResp.GetExecutions(), 1)
|
||||
},
|
||||
testcore.WaitForESToSettle,
|
||||
100*time.Millisecond,
|
||||
)
|
||||
|
||||
_, err = s.terminateWorkflow(ctx, tv)
|
||||
s.Require().NoError(err)
|
||||
}
|
||||
Reference in New Issue
Block a user