192
request *archiver.QueryVisibilityRequest,
193
saTypeMap searchattribute.NameTypeMap,
195
>
// remaining is the number of workflow executions left to return before we reach pageSize.
196
>
remaining := request.PageSize
197
>
nextPageToken := request.NextPageToken
198
>
var executions []*workflowpb.WorkflowExecutionInfo
199
>
// We need to loop because the number of workflow executions returned by each call to query may be fewer than
200
>
// pageSize. This is because we may have to skip some workflow executions after querying S3 (client-side filtering)
201
>
// because there are 2 entries in S3 for each workflow execution indexed by workflowTypeName (one for closeTimeout
202
>
// and one for startTimeout), and we only want to return one entry per workflow execution. See
203
>
// createIndexesToArchive for a list of all indexes.
204
>
for {
205
>
searchPrefix := constructVisibilitySearchPrefix(uri.Path(), request.NamespaceID)
206
>
// We suffix searchPrefix with workflowTypeName because the data in S3 is duplicated across combinations of 2
207
>
// different primary indices (workflowID and workflowTypeName) and 2 different secondary indices (closeTimeout
208
>
// and startTimeout). We only want to return one entry per workflow execution, but the full path to the S3 key
209
>
// is <primaryIndexKey>/<primaryIndexValue>/<secondaryIndexKey>/<secondaryIndexValue>/<runID>, and we don't have
210
>
// the primaryIndexValue when we make the call to query, so we can only specify the primaryIndexKey.
211
>
searchPrefix += "/" + primaryIndexKeyWorkflowTypeName
212
>
// The pageSize we supply here is actually the maximum number of keys to fetch from S3. For each execution,
213
>
// there should be 2 keys in S3 for this prefix, so you might think that we should multiply the pageSize by 2.
214
>
// However, if we do that, we may end up returning more than pageSize workflow executions to the end user of
215
>
// this API. This is because we aren't guaranteed that both keys for a given workflow execution will be returned
216
>
// in the same call. For example, if the user supplies a pageSize of 1, and we specify a maximum number of keys
217
>
// of 2 to S3, we may get back entries from S3 for 2 different workflow executions. You might think that we can
218
>
// just truncate this result to 1 workflow execution, but then the nextPageToken would be incorrect. So, we may
219
>
// need to make multiple calls to S3 to get the correct number of workflow executions, which will probably make
220
>
// this API call slower.
221
>
res, err := v.queryPrefix(ctx, uri, &queryVisibilityRequest{
222
>
namespaceID: request.NamespaceID,
223
>
pageSize: remaining,
224
>
nextPageToken: nextPageToken,
225
>
parsedQuery: &parsedQuery{},
226
>
}, saTypeMap, searchPrefix, func(key string) bool {
227
>
// We only want to return entries for the closeTimeout secondary index. Keys for this
228
>
// index are always of the form:
229
>
// .../closeTimeout/<timestamp>/<runID>
230
>
// Walk from the end instead of splitting the whole string to avoid unnecessary
231
>
// allocations and to keep the logic clear:
232
>
// - drop <runID>
233
>
// - drop <timestamp>
234
>
// - check the remaining last segment equals "closeTimeout".
235
>
dir := path.Dir(key) // drop runID
236
>
dir = path.Dir(dir) // drop <timestamp>
237
>
return path.Base(dir) == secondaryIndexKeyCloseTimeout
238
>
})
239
>
if err != nil {
240
return nil, err
241
}
243
>
executions = append(executions, res.Executions...)
244
>
remaining -= len(res.Executions)
245
>
if len(nextPageToken) == 0 || remaining <= 0 {
246
>
break
247
}
248
}
250
>
Executions: executions,
251
>
NextPageToken: nextPageToken,
252
>
}, nil
253
}
254