Vihren Atlas

visibility_store.go ×17

Joint frontier

unlabeled · c_747c5eba8c5d

1 test · 44979 LOC · 761 files · introduces 1 test · 171 LOC · 15 files

Introduces — evidence that enters the hierarchy at this concept

Code
49 ranges171 lines · 15 files
Tests
1 test

Contains — complete concept membership

All code (extent)
10722 ranges44979 lines · 761 files · Browse complete extent
All tests (intent)
1 testBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

1 test introduced at this concept.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

15 files ranked by introduced lines: 171 introduced LOC across 49 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/common/persistence/visibility/store/sql/visibility_store.go 63 introduced LOC · 17 ranges

Open complete file

201 return s.listWorkflowExecutions(ctx, request)
202 }
203 > return s.listWorkflowExecutionsLegacy(ctx, request) visibility_store.go
204 }
205
455 ctx context.Context,
456 request *manager.ListWorkflowExecutionsRequestV2,
457 > ) (*store.InternalListExecutionsResponse, error) { visibility_store.go
458 > return s.listExecutionsInternalLegacy(ctx, &listExecutionsRequestInternal{
459 > NamespaceID: request.NamespaceID,
460 > Namespace: request.Namespace,
461 > Query: request.Query,
462 > PageSize: request.PageSize,
463 > NextPageToken: request.NextPageToken,
464 > ChasmMapper: nil,
465 > ArchetypeID: chasm.UnspecifiedArchetypeID,
466 > })
467 > }
468
469 func (s *VisibilityStore) listExecutionsInternalLegacy(
470 ctx context.Context,
471 request *listExecutionsRequestInternal,
472 > ) (*store.InternalListExecutionsResponse, error) { visibility_store.go
473 > saTypeMap, err := s.searchAttributesProvider.GetSearchAttributes(s.GetIndexName(), false)
474 > if err != nil {
475 return nil, err
476 }
477
478 > saMapper, err := s.searchAttributesMapperProvider.GetMapper(request.Namespace) visibility_store.go
479 > if err != nil {
480 return nil, err
481 }
482
483 > converter := NewQueryConverterLegacy( visibility_store.go
484 > s.GetName(),
485 > request.Namespace,
486 > request.NamespaceID,
487 > saTypeMap,
488 > saMapper,
489 > request.Query,
490 > request.ChasmMapper,
491 > request.ArchetypeID,
492 > )
493 > selectFilter, err := converter.BuildSelectStmt(request.PageSize, request.NextPageToken)
494 > if err != nil {
495 // Convert ConverterError to InvalidArgument and pass through all other errors (which should be only mapper errors).
496 var converterErr *query.ConverterError
501 }
502
503 > rows, err := s.sqlStore.DB.SelectFromVisibility(ctx, *selectFilter) visibility_store.go
504 > if err != nil {
505 return nil, convertSQLError("ListWorkflowExecutions operation failed.", err)
506 }
507 > if len(rows) == 0 { visibility_store.go
508 return &store.InternalListExecutionsResponse{}, nil
509 }
510
511 > var infos = make([]*store.InternalExecutionInfo, len(rows)) visibility_store.go
512 > for i, row := range rows {
513 > infos[i], err = s.rowToInfo(&row, request.ChasmMapper)
514 > if err != nil {
515 return nil, err
516 }
517 }
518
519 > var nextPageToken []byte visibility_store.go
520 > if len(rows) == request.PageSize {
521 lastRow := rows[len(rows)-1]
522 closeTime := maxDatetime
533 }
534 }
535 > return &store.InternalListExecutionsResponse{ visibility_store.go
536 > Executions: infos,
537 > NextPageToken: nextPageToken,
538 > }, nil
539 }
540
842 }
843 if row.SearchAttributes != nil && len(*row.SearchAttributes) > 0 {
844 > // Encode all search attributes together (both CHASM and custom) visibility_store.go
845 > encodedSAs, err := s.encodeRowSearchAttributes(*row.SearchAttributes, chasmMapper)
846 > if err != nil {
847 return nil, err
848 }
849 > info.SearchAttributes = encodedSAs visibility_store.go
850 }
851 if row.CloseTime != nil {
876 rowSearchAttributes sqlplugin.VisibilitySearchAttributes,
877 chasmMapper *chasm.VisibilitySearchAttributesMapper,
878 > ) (*commonpb.SearchAttributes, error) { visibility_store.go
879 > saTypeMap, err := s.searchAttributesProvider.GetSearchAttributes(s.GetIndexName(), false)
880 > if err != nil {
881 return nil, serviceerror.NewUnavailable(
882 fmt.Sprintf("Unable to read search attributes types: %v", err))
883 }
884
885 > combinedTypeMap := store.CombineTypeMaps(saTypeMap, chasmMapper) visibility_store.go
886 > registeredSearchAttributes := sqlplugin.VisibilitySearchAttributes{}
887 >
888 > // Fix SQLite keyword list handling (convert string to []string for keyword lists)
889 > for name, value := range rowSearchAttributes {
890 > tp, err := combinedTypeMap.GetType(name)
891 > if err != nil {
892 if errors.Is(err, sadefs.ErrInvalidName) {
893 continue
895 return nil, err
896 }
897 > registeredSearchAttributes[name] = value visibility_store.go
898 > if tp == enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST {
899 > switch v := value.(type) {
900 > case []string:
901 // no-op
902 case string:
911
912 // Encode all search attributes together
913 > encodedSAs, err := searchattribute.Encode(registeredSearchAttributes, &combinedTypeMap) visibility_store.go
914 > if err != nil {
915 return nil, err
916 }
917
918 > return encodedSAs, nil visibility_store.go
919 }
920
go.temporal.io/server/common/persistence/visibility/store/sql/query_converter_legacy_sqlite.go 40 introduced LOC · 4 ranges

Open complete file

33 chasmMapper *chasm.VisibilitySearchAttributesMapper,
34 archetypeID chasm.ArchetypeID,
35 > ) *QueryConverterLegacy { query_converter_legacy_sqlite.go
36 > return newQueryConverterInternal(
37 > &sqliteQueryConverter{},
38 > namespaceName,
39 > namespaceID,
40 > saTypeMap,
41 > saMapper,
42 > queryString,
43 > chasmMapper,
44 > archetypeID,
45 > )
46 > }
47
48 func (c *sqliteQueryConverter) getDatetimeFormat() string {
213 pageSize int,
214 token *pageTokenLegacy,
215 > ) (string, []any) { query_converter_legacy_sqlite.go
216 > var whereClauses []string
217 > var queryArgs []any
218 >
219 > whereClauses = append(
220 > whereClauses,
221 > fmt.Sprintf("%s = ?", sadefs.GetSqlDbColName(sadefs.NamespaceID)),
222 > )
223 > queryArgs = append(queryArgs, namespaceID.String())
224 >
225 > if len(queryString) > 0 {
226 > whereClauses = append(whereClauses, queryString)
227 > }
228
229 > if token != nil { query_converter_legacy_sqlite.go
230 whereClauses = append(
231 whereClauses,
251 }
252
253 > queryArgs = append(queryArgs, pageSize) query_converter_legacy_sqlite.go
254 >
255 > return fmt.Sprintf(
256 > `SELECT %s
257 > FROM executions_visibility
258 > WHERE %s
259 > ORDER BY %s DESC, %s DESC, %s
260 > LIMIT ?`,
261 > strings.Join(sqlplugin.DbFields, ", "),
262 > strings.Join(whereClauses, " AND "),
263 > sqlparser.String(c.getCoalesceCloseTimeExpr()),
264 > sadefs.GetSqlDbColName(sadefs.StartTime),
265 > sadefs.GetSqlDbColName(sadefs.RunID),
266 > ), queryArgs
267 }
268
go.temporal.io/server/common/persistence/visibility/store/sql/query_converter_legacy.go 13 introduced LOC · 4 ranges

Open complete file

123 pageSize int,
124 nextPageToken []byte,
125 > ) (*sqlplugin.VisibilitySelectFilter, error) { query_converter_legacy.go
126 > token, err := deserializePageTokenLegacy(nextPageToken)
127 > if err != nil {
128 return nil, err
129 }
130 > qp, err := c.convertWhereString(c.queryString) query_converter_legacy.go
131 > if err != nil {
132 return nil, err
133 }
134 > if len(qp.groupBy) > 0 { query_converter_legacy.go
135 return nil, query.NewConverterError("%s: 'GROUP BY' clause", query.NotSupportedErrMessage)
136 }
137 > queryString, queryArgs := c.buildSelectStmt( query_converter_legacy.go
138 > c.namespaceID,
139 > qp.queryString,
140 > pageSize,
141 > token,
142 > )
143 > return &sqlplugin.VisibilitySelectFilter{Query: queryString, QueryArgs: queryArgs}, nil
144 }
145
go.temporal.io/server/service/matching/pri_matcher.go 13 introduced LOC · 6 ranges

Open complete file

205 defer stop()
206 } else {
207 > // Task is from local backlog. pri_matcher.go
208 >
209 > // Before we forward, ask task validator. This will happen every BacklogTaskForwardTimeout
210 > // to the head of the backlog, which is what taskValidator expects.
211 > maybeValid := tm.validator.maybeValidate(task.event.AllocatedTaskInfo, tm.fwdr.partition.TaskType())
212 > if !maybeValid {
213 // consider this task expired while processing.
214 task.finish(taskFinishResult{dropReason: getDroppedTaskExpiryReason(task)})
222 // Add a timeout for forwarding.
223 // Note that this does not block local match of other local backlog tasks.
224 > ctx, cancel = context.WithTimeout(tm.tqCtx, tm.config.BacklogTaskForwardTimeout()) pri_matcher.go
225 > defer cancel()
226 }
227
398 return syncMatchSuccess, nil
399 }
400 > return syncMatchNoPoller, nil // forward error, give up here pri_matcher.go
401 }
402 // TODO(pri): can we just always do this on the parent and simplify this to:
430 }
431
432 > res := tm.data.EnqueueTaskAndWait([]context.Context{ctx, tm.tqCtx}, task) pri_matcher.go
433 > if res.ctxErr != nil {
434 return syncMatchNoPoller, res.ctxErr
435 }
436 > if !softassert.That(tm.logger, res.poller != nil, "expeced poller from match") { pri_matcher.go
437 return syncMatchNoPoller, nil
438 }
439
440 > return finish() pri_matcher.go
441 }
442
go.temporal.io/server/common/persistence/sql/sqlplugin/sqlite/visibility.go 10 introduced LOC · 2 ranges

Open complete file

171 case time.Time:
172 finalSearchAttributes[name] = v.Format(time.RFC3339Nano)
173 > default: visibility.go
174 > finalSearchAttributes[name] = v
175 }
176 }
191 }
192 if row.SearchAttributes != nil {
193 > for saName, saValue := range *row.SearchAttributes { visibility.go
194 > switch typedSaValue := saValue.(type) {
195 > case string:
196 > if strings.Contains(typedSaValue, keywordListSeparator) {
197 > // If the string contains the keywordListSeparator, then we need to split it
198 > // into a list of keywords.
199 > (*row.SearchAttributes)[saName] = strings.Split(typedSaValue, keywordListSeparator)
200 > }
201 default:
202 // no-op
go.temporal.io/server/service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler.go 6 introduced LOC · 1 range

Open complete file

1328 }
1329 if unaliasedSas != attr.GetSearchAttributes() {
1330 > // Create a copy of the `attr` to avoid modification of original `attr`, workflow_task_completed_handler.go
1331 > // which can be needed again in case of retry.
1332 > newAttr := common.CloneProto(attr)
1333 > newAttr.SearchAttributes = unaliasedSas
1334 > attr = newAttr
1335 > }
1336
1337 // valid search attributes for upsert
go.temporal.io/server/common/persistence/sql/sqlplugin/visibility.go 5 introduced LOC · 3 ranges

Open complete file

116 var DbFields = getDbFields()
117
118 > func (vsa *VisibilitySearchAttributes) Scan(src any) error { visibility.go
119 > if src == nil {
120 return nil
121 }
122 > switch v := src.(type) { visibility.go
123 case []byte:
124 return json.Unmarshal(v, &vsa)
125 > case string: visibility.go
126 > return json.Unmarshal([]byte(v), &vsa)
127 default:
128 return fmt.Errorf("unsupported type for VisibilitySearchAttributes: %T", v)
go.temporal.io/server/common/persistence/visibility/store/sql/query_converter_legacy_factory.go 4 introduced LOC · 2 ranges

Open complete file

19 chasmMapper *chasm.VisibilitySearchAttributesMapper,
20 archetypeID chasm.ArchetypeID,
21 > ) *QueryConverterLegacy { query_converter_legacy_factory.go
22 > switch pluginName {
23 case mysql.PluginName:
24 return newMySQLQueryConverter(namespaceName, namespaceID, saTypeMap, saMapper, queryString, chasmMapper, archetypeID)
25 case postgresql.PluginName, postgresql.PluginNamePGX:
26 return newPostgreSQLQueryConverter(namespaceName, namespaceID, saTypeMap, saMapper, queryString, chasmMapper, archetypeID)
27 > case sqlite.PluginName: query_converter_legacy_factory.go
28 > return newSqliteQueryConverter(namespaceName, namespaceID, saTypeMap, saMapper, queryString, chasmMapper, archetypeID)
29 default:
30 return nil
go.temporal.io/server/common/persistence/visibility/visibility_manager_impl.go 4 introduced LOC · 2 ranges

Open complete file

274 customSAs = &commonpb.SearchAttributes{IndexedFields: make(map[string]*commonpb.Payload)}
275 for name, payloadValue := range searchAttributes.GetIndexedFields() {
276 > if sadefs.IsChasmSearchAttribute(name) { visibility_manager_impl.go
277 chasmSAs.IndexedFields[name] = payloadValue
278 > } else { visibility_manager_impl.go
279 > customSAs.IndexedFields[name] = payloadValue
280 > }
281 }
282 return
go.temporal.io/server/temporaltest/server.go 3 introduced LOC · 1 range

Open complete file

74
75 // GetDefaultNamespace returns the randomly generated namespace which has been pre-registered with the test server.
76 > func (ts *TestServer) GetDefaultNamespace() string { server.go
77 > return ts.defaultTestNamespace
78 > }
79
80 // GetFrontendHostPort returns the host:port for this server.
go.temporal.io/server/api/matchingservice/v1/service_grpc.pb.go 2 introduced LOC · 1 range

Open complete file

285 err := c.cc.Invoke(ctx, MatchingService_AddWorkflowTask_FullMethodName, in, out, opts...)
286 if err != nil {
287 > return nil, err service_grpc.pb.go
288 > }
289 return out, nil
290 }
go.temporal.io/server/common/client_cache.go 2 introduced LOC · 1 range

Open complete file

88 entry, ok = c.clients[clientKey]
89 if ok {
90 > return entry.client, nil client_cache.go
91 > }
92
93 client, release, err := c.clientProvider(clientKey)
go.temporal.io/server/common/namespace/namespace.go 2 introduced LOC · 2 ranges

Open complete file

360 )
361 }
362 > return alias, nil namespace.go
363 }
364
370 )
371 }
372 > return fieldName, nil namespace.go
373 }
374
go.temporal.io/server/common/searchattribute/mapper.go 2 introduced LOC · 2 ranges

Open complete file

68 return fieldName, nil
69 }
70 > return alias, nil mapper.go
71 }
72
80 return alias, nil
81 }
82 > return fieldName, nil mapper.go
83 }
84
go.temporal.io/server/service/matching/task.go 2 introduced LOC · 1 range

Open complete file

125 func (res taskResponse) err() error {
126 if res.forwarded {
127 > return res.forwardErr task.go
128 > }
129 return res.startErr
130 }

Direct hierarchy

Children