Project import generated by Copybara.
GitOrigin-RevId: d9e9e3fb4e31372ec1fb43b178994ca78fa8fe70
diff --git a/search/frontendsearchpipeline.py b/search/frontendsearchpipeline.py
new file mode 100644
index 0000000..367c52f
--- /dev/null
+++ b/search/frontendsearchpipeline.py
@@ -0,0 +1,1237 @@
+# Copyright 2016 The Chromium Authors. All rights reserved.
+# Use of this source code is governed by a BSD-style
+# license that can be found in the LICENSE file or at
+# https://developers.google.com/open-source/licenses/bsd
+
+"""The FrontendSearchPipeline class manages issue search and sorting.
+
+The frontend pipeline checks memcache for cached results in each shard. It
+then calls backend jobs to do any shards that had a cache miss. On cache hit,
+the cached results must be filtered by permissions, so the at-risk cache and
+backends are consulted. Next, the sharded results are combined into an overall
+list of IIDs. Then, that list is paginated and the issues on the current
+pagination page can be shown. Alternatively, this class can determine just the
+position the currently shown issue would occupy in the overall sorted list.
+"""
+
+from __future__ import division
+from __future__ import print_function
+from __future__ import absolute_import
+
+import json
+
+import collections
+import logging
+import math
+import random
+import time
+
+from google.appengine.api import apiproxy_stub_map
+from google.appengine.api import memcache
+from google.appengine.api import modules
+from google.appengine.api import urlfetch
+
+import settings
+from features import savedqueries_helpers
+from framework import framework_bizobj
+from framework import framework_constants
+from framework import framework_helpers
+from framework import paginate
+from framework import permissions
+from framework import sorting
+from framework import urls
+from search import ast2ast
+from search import query2ast
+from search import searchpipeline
+from services import fulltext_helpers
+from tracker import tracker_bizobj
+from tracker import tracker_constants
+from tracker import tracker_helpers
+
+
+# Fail-fast responses usually finish in less than 50ms. If we see a failure
+# in under that amount of time, we don't bother logging it.
+FAIL_FAST_LIMIT_SEC = 0.1
+
+DELAY_BETWEEN_RPC_COMPLETION_POLLS = 0.04 # 40 milliseconds
+
+# The choices help balance the cost of choosing samples vs. the cost of
+# selecting issues that are in a range bounded by neighboring samples.
+# Preferred chunk size parameters were determined by experimentation.
+MIN_SAMPLE_CHUNK_SIZE = int(
+ math.sqrt(tracker_constants.DEFAULT_RESULTS_PER_PAGE))
+MAX_SAMPLE_CHUNK_SIZE = int(math.sqrt(settings.search_limit_per_shard))
+PREFERRED_NUM_CHUNKS = 50
+
+
+# TODO(jojwang): monorail:4127: combine some url parameters info or
+# query info into dicts or tuples to make argument manager easier.
+class FrontendSearchPipeline(object):
+ """Manage the process of issue search, including backends and caching.
+
+ Even though the code is divided into several methods, the public
+ methods should be called in sequence, so the execution of the code
+ is pretty much in the order of the source code lines here.
+ """
+
+ def __init__(
+ self,
+ cnxn,
+ services,
+ auth,
+ me_user_ids,
+ query,
+ query_project_names,
+ items_per_page,
+ paginate_start,
+ can,
+ group_by_spec,
+ sort_spec,
+ warnings,
+ errors,
+ use_cached_searches,
+ profiler,
+ project=None):
+ self.cnxn = cnxn
+ self.me_user_ids = me_user_ids
+ self.auth = auth
+ self.logged_in_user_id = auth.user_id or 0
+ self.can = can
+ self.items_per_page = items_per_page
+ self.paginate_start = paginate_start
+ self.group_by_spec = group_by_spec
+ self.sort_spec = sort_spec
+ self.warnings = warnings
+ self.use_cached_searches = use_cached_searches
+ self.profiler = profiler
+
+ self.services = services
+ self.pagination = None
+ self.num_skipped_at_start = 0
+ self.total_count = 0
+ self.errors = errors
+
+ self.project_name = ''
+ if project:
+ self.project_name = project.project_name
+ self.query_projects = []
+ if query_project_names:
+ consider_projects = list(services.project.GetProjectsByName(
+ self.cnxn, query_project_names).values())
+ self.query_projects = [
+ p for p in consider_projects
+ if permissions.UserCanViewProject(
+ self.auth.user_pb, self.auth.effective_ids, p)]
+ if project:
+ self.query_projects.append(project)
+ member_of_all_projects = self.auth.user_pb.is_site_admin or all(
+ framework_bizobj.UserIsInProject(p, self.auth.effective_ids)
+ for p in self.query_projects)
+ self.query_project_ids = sorted([
+ p.project_id for p in self.query_projects])
+ self.query_project_names = sorted([
+ p.project_name for p in self.query_projects])
+
+ config_dict = self.services.config.GetProjectConfigs(
+ self.cnxn, self.query_project_ids)
+ self.harmonized_config = tracker_bizobj.HarmonizeConfigs(
+ list(config_dict.values()))
+
+ # The following fields are filled in as the pipeline progresses.
+ # The value None means that we still need to compute that value.
+ # A shard_key is a tuple (shard_id, subquery).
+ self.users_by_id = {}
+ self.nonviewable_iids = {} # {shard_id: set(iid)}
+ self.unfiltered_iids = {} # {shard_key: [iid, ...]} needing perm checks.
+ self.filtered_iids = {} # {shard_key: [iid, ...]} already perm checked.
+ self.search_limit_reached = {} # {shard_key: [bool, ...]}.
+ self.allowed_iids = [] # Matching iids that user is permitted to view.
+ self.allowed_results = None # results that the user is permitted to view.
+ self.visible_results = None # allowed_results on current pagination page.
+ self.error_responses = set()
+
+ error_msg = _CheckQuery(
+ self.cnxn, self.services, query, self.harmonized_config,
+ self.query_project_ids, member_of_all_projects,
+ warnings=self.warnings)
+ if error_msg:
+ self.errors.query = error_msg
+
+ # Split up query into smaller subqueries that would get the same results
+ # to improve performance. Smaller queries are more likely to get cache
+ # hits and subqueries can be parallelized by querying for them across
+ # multiple shards.
+ self.subqueries = []
+ try:
+ self.subqueries = query2ast.QueryToSubqueries(query)
+ except query2ast.InvalidQueryError:
+ # Ignore errors because they've already been recorded in
+ # self.errors.query.
+ pass
+
+ def SearchForIIDs(self):
+ """Use backends to search each shard and store their results."""
+ with self.profiler.Phase('Checking cache and calling Backends'):
+ rpc_tuples = _StartBackendSearch(
+ self.cnxn, self.query_project_names, self.query_project_ids,
+ self.harmonized_config, self.unfiltered_iids,
+ self.search_limit_reached, self.nonviewable_iids,
+ self.error_responses, self.services, self.me_user_ids,
+ self.logged_in_user_id, self.items_per_page + self.paginate_start,
+ self.subqueries, self.can, self.group_by_spec, self.sort_spec,
+ self.warnings, self.use_cached_searches)
+
+ with self.profiler.Phase('Waiting for Backends'):
+ try:
+ _FinishBackendSearch(rpc_tuples)
+ except Exception as e:
+ logging.exception(e)
+ raise
+
+ if self.error_responses:
+ logging.error('%r error responses. Incomplete search results.',
+ self.error_responses)
+
+ with self.profiler.Phase('Filtering cached results'):
+ for shard_key in self.unfiltered_iids:
+ shard_id, _subquery = shard_key
+ if shard_id not in self.nonviewable_iids:
+ logging.error(
+ 'Not displaying shard %r because of no nonviewable_iids', shard_id)
+ self.error_responses.add(shard_id)
+ filtered_shard_iids = []
+ else:
+ unfiltered_shard_iids = self.unfiltered_iids[shard_key]
+ nonviewable_shard_iids = self.nonviewable_iids[shard_id]
+ # TODO(jrobbins): avoid creating large temporary lists.
+ filtered_shard_iids = [iid for iid in unfiltered_shard_iids
+ if iid not in nonviewable_shard_iids]
+ self.filtered_iids[shard_key] = filtered_shard_iids
+
+ seen_iids_by_shard_id = collections.defaultdict(set)
+ with self.profiler.Phase('Dedupping result IIDs across shards'):
+ for shard_key in self.filtered_iids:
+ shard_id, _subquery = shard_key
+ deduped = [iid for iid in self.filtered_iids[shard_key]
+ if iid not in seen_iids_by_shard_id[shard_id]]
+ self.filtered_iids[shard_key] = deduped
+ seen_iids_by_shard_id[shard_id].update(deduped)
+
+ with self.profiler.Phase('Counting all filtered results'):
+ for shard_key in self.filtered_iids:
+ self.total_count += len(self.filtered_iids[shard_key])
+
+ with self.profiler.Phase('Trimming results beyond pagination page'):
+ for shard_key in self.filtered_iids:
+ self.filtered_iids[shard_key] = self.filtered_iids[
+ shard_key][:self.paginate_start + self.items_per_page]
+
+ def MergeAndSortIssues(self):
+ """Merge and sort results from all shards into one combined list."""
+ with self.profiler.Phase('selecting issues to merge and sort'):
+ self._NarrowFilteredIIDs()
+ self.allowed_iids = []
+ for filtered_shard_iids in self.filtered_iids.values():
+ self.allowed_iids.extend(filtered_shard_iids)
+
+ with self.profiler.Phase('getting allowed results'):
+ self.allowed_results = self.services.issue.GetIssues(
+ self.cnxn, self.allowed_iids)
+
+ # Note: At this point, we have results that are only sorted within
+ # each backend's shard. We still need to sort the merged result.
+ self._LookupNeededUsers(self.allowed_results)
+ with self.profiler.Phase('merging and sorting issues'):
+ self.allowed_results = _SortIssues(
+ self.allowed_results, self.harmonized_config, self.users_by_id,
+ self.group_by_spec, self.sort_spec)
+
+ def _NarrowFilteredIIDs(self):
+ """Combine filtered shards into a range of IIDs for issues to sort.
+
+ The niave way is to concatenate shard_iids[:start + num] for all
+ shards then select [start:start + num]. We do better by sampling
+ issues and then determining which of those samples are known to
+ come before start or after start+num. We then trim off all those IIDs
+ and sort a smaller range of IIDs that might actuall be displayed.
+ See the design doc at go/monorail-sorting.
+
+ This method modifies self.fitered_iids and self.num_skipped_at_start.
+ """
+ # Sample issues and skip those that are known to come before start.
+ # See the "Sorting in Monorail" design doc.
+
+ # If the result set is small, don't bother optimizing it.
+ orig_length = _TotalLength(self.filtered_iids)
+ if orig_length < self.items_per_page * 4:
+ return
+
+ # 1. Get sample issues in each shard and sort them all together.
+ last = self.paginate_start + self.items_per_page
+
+ samples_by_shard, sample_iids_to_shard = self._FetchAllSamples(
+ self.filtered_iids)
+ sample_issues = []
+ for issue_dict in samples_by_shard.values():
+ sample_issues.extend(list(issue_dict.values()))
+
+ self._LookupNeededUsers(sample_issues)
+ sample_issues = _SortIssues(
+ sample_issues, self.harmonized_config, self.users_by_id,
+ self.group_by_spec, self.sort_spec)
+ sample_iid_tuples = [
+ (issue.issue_id, sample_iids_to_shard[issue.issue_id])
+ for issue in sample_issues]
+
+ # 2. Trim off some IIDs that are sure to be positioned after last.
+ num_trimmed_end = _TrimEndShardedIIDs(
+ self.filtered_iids, sample_iid_tuples, last)
+ logging.info('Trimmed %r issues from the end of shards', num_trimmed_end)
+
+ # 3. Trim off some IIDs that are sure to be posiitoned before start.
+ keep = _TotalLength(self.filtered_iids) - self.paginate_start
+ # Reverse the sharded lists.
+ _ReverseShards(self.filtered_iids)
+ sample_iid_tuples.reverse()
+ self.num_skipped_at_start = _TrimEndShardedIIDs(
+ self.filtered_iids, sample_iid_tuples, keep)
+ logging.info('Trimmed %r issues from the start of shards',
+ self.num_skipped_at_start)
+ # Reverse sharded lists again to get back into forward order.
+ _ReverseShards(self.filtered_iids)
+
+ def DetermineIssuePosition(self, issue):
+ """Calculate info needed to show the issue flipper.
+
+ Args:
+ issue: The issue currently being viewed.
+
+ Returns:
+ A 3-tuple (prev_iid, index, next_iid) were prev_iid is the
+ IID of the previous issue in the total ordering (or None),
+ index is the index that the current issue has in the total
+ ordering, and next_iid is the next issue (or None). If the current
+ issue is not in the list of results at all, returns None, None, None.
+ """
+ # 1. If the current issue is not in the results at all, then exit.
+ if not any(issue.issue_id in filtered_shard_iids
+ for filtered_shard_iids in self.filtered_iids.values()):
+ return None, None, None
+
+ # 2. Choose and retrieve sample issues in each shard.
+ samples_by_shard, _ = self._FetchAllSamples(self.filtered_iids)
+
+ # 3. Build up partial results for each shard.
+ preceeding_counts = {} # dict {shard_key: num_issues_preceeding_current}
+ prev_candidates, next_candidates = [], []
+ for shard_key in self.filtered_iids:
+ prev_candidate, index_in_shard, next_candidate = (
+ self._DetermineIssuePositionInShard(
+ shard_key, issue, samples_by_shard[shard_key]))
+ preceeding_counts[shard_key] = index_in_shard
+ if prev_candidate:
+ prev_candidates.append(prev_candidate)
+ if next_candidate:
+ next_candidates.append(next_candidate)
+
+ # 4. Combine the results.
+ index = sum(preceeding_counts.values())
+ prev_candidates = _SortIssues(
+ prev_candidates, self.harmonized_config, self.users_by_id,
+ self.group_by_spec, self.sort_spec)
+ prev_iid = prev_candidates[-1].issue_id if prev_candidates else None
+ next_candidates = _SortIssues(
+ next_candidates, self.harmonized_config, self.users_by_id,
+ self.group_by_spec, self.sort_spec)
+ next_iid = next_candidates[0].issue_id if next_candidates else None
+
+ return prev_iid, index, next_iid
+
+ def _DetermineIssuePositionInShard(self, shard_key, issue, sample_dict):
+ """Determine where the given issue would fit into results from a shard."""
+ # See the design doc for details. Basically, it first surveys the results
+ # to bound a range where the given issue would belong, then it fetches the
+ # issues in that range and sorts them.
+
+ filtered_shard_iids = self.filtered_iids[shard_key]
+
+ # 1. Select a sample of issues, leveraging ones we have in RAM already.
+ issues_on_hand = list(sample_dict.values())
+ if issue.issue_id not in sample_dict:
+ issues_on_hand.append(issue)
+
+ self._LookupNeededUsers(issues_on_hand)
+ sorted_on_hand = _SortIssues(
+ issues_on_hand, self.harmonized_config, self.users_by_id,
+ self.group_by_spec, self.sort_spec)
+ sorted_on_hand_iids = [soh.issue_id for soh in sorted_on_hand]
+ index_in_on_hand = sorted_on_hand_iids.index(issue.issue_id)
+
+ # 2. Bound the gap around where issue belongs.
+ if index_in_on_hand == 0:
+ fetch_start = 0
+ else:
+ prev_on_hand_iid = sorted_on_hand_iids[index_in_on_hand - 1]
+ fetch_start = filtered_shard_iids.index(prev_on_hand_iid) + 1
+
+ if index_in_on_hand == len(sorted_on_hand) - 1:
+ fetch_end = len(filtered_shard_iids)
+ else:
+ next_on_hand_iid = sorted_on_hand_iids[index_in_on_hand + 1]
+ fetch_end = filtered_shard_iids.index(next_on_hand_iid)
+
+ # 3. Retrieve all the issues in that gap to get an exact answer.
+ fetched_issues = self.services.issue.GetIssues(
+ self.cnxn, filtered_shard_iids[fetch_start:fetch_end])
+ if issue.issue_id not in filtered_shard_iids[fetch_start:fetch_end]:
+ fetched_issues.append(issue)
+ self._LookupNeededUsers(fetched_issues)
+ sorted_fetched = _SortIssues(
+ fetched_issues, self.harmonized_config, self.users_by_id,
+ self.group_by_spec, self.sort_spec)
+ sorted_fetched_iids = [sf.issue_id for sf in sorted_fetched]
+ index_in_fetched = sorted_fetched_iids.index(issue.issue_id)
+
+ # 4. Find the issues that come immediately before and after the place where
+ # the given issue would belong in this shard.
+ if index_in_fetched > 0:
+ prev_candidate = sorted_fetched[index_in_fetched - 1]
+ elif index_in_on_hand > 0:
+ prev_candidate = sorted_on_hand[index_in_on_hand - 1]
+ else:
+ prev_candidate = None
+
+ if index_in_fetched < len(sorted_fetched) - 1:
+ next_candidate = sorted_fetched[index_in_fetched + 1]
+ elif index_in_on_hand < len(sorted_on_hand) - 1:
+ next_candidate = sorted_on_hand[index_in_on_hand + 1]
+ else:
+ next_candidate = None
+
+ return prev_candidate, fetch_start + index_in_fetched, next_candidate
+
+ def _FetchAllSamples(self, filtered_iids):
+ """Return a dict {shard_key: {iid: sample_issue}}."""
+ samples_by_shard = {} # {shard_key: {iid: sample_issue}}
+ sample_iids_to_shard = {} # {iid: shard_key}
+ all_needed_iids = [] # List of iids to retrieve.
+
+ for shard_key in filtered_iids:
+ on_hand_issues, shard_needed_iids = self._ChooseSampleIssues(
+ filtered_iids[shard_key])
+ samples_by_shard[shard_key] = on_hand_issues
+ for iid in on_hand_issues:
+ sample_iids_to_shard[iid] = shard_key
+ for iid in shard_needed_iids:
+ sample_iids_to_shard[iid] = shard_key
+ all_needed_iids.extend(shard_needed_iids)
+
+ retrieved_samples, _misses = self.services.issue.GetIssuesDict(
+ self.cnxn, all_needed_iids)
+ for retrieved_iid, retrieved_issue in retrieved_samples.items():
+ retr_shard_key = sample_iids_to_shard[retrieved_iid]
+ samples_by_shard[retr_shard_key][retrieved_iid] = retrieved_issue
+
+ return samples_by_shard, sample_iids_to_shard
+
+ def _ChooseSampleIssues(self, issue_ids):
+ """Select a scattering of issues from the list, leveraging RAM cache.
+
+ Args:
+ issue_ids: A list of issue IDs that comprise the results in a shard.
+
+ Returns:
+ A pair (on_hand_issues, needed_iids) where on_hand_issues is
+ an issue dict {iid: issue} of issues already in RAM, and
+ shard_needed_iids is a list of iids of issues that need to be retrieved.
+ """
+ on_hand_issues = {} # {iid: issue} of sample issues already in RAM.
+ needed_iids = [] # [iid, ...] of sample issues not in RAM yet.
+ chunk_size = max(MIN_SAMPLE_CHUNK_SIZE, min(MAX_SAMPLE_CHUNK_SIZE,
+ int(len(issue_ids) // PREFERRED_NUM_CHUNKS)))
+ for i in range(chunk_size, len(issue_ids), chunk_size):
+ issue = self.services.issue.GetAnyOnHandIssue(
+ issue_ids, start=i, end=min(i + chunk_size, len(issue_ids)))
+ if issue:
+ on_hand_issues[issue.issue_id] = issue
+ else:
+ needed_iids.append(issue_ids[i])
+
+ return on_hand_issues, needed_iids
+
+ def _LookupNeededUsers(self, issues):
+ """Look up user info needed to sort issues, if any."""
+ with self.profiler.Phase('lookup of owner, reporter, and cc'):
+ additional_user_views_by_id = (
+ tracker_helpers.MakeViewsForUsersInIssues(
+ self.cnxn, issues, self.services.user,
+ omit_ids=list(self.users_by_id.keys())))
+ self.users_by_id.update(additional_user_views_by_id)
+
+ def Paginate(self):
+ """Fetch matching issues and paginate the search results.
+
+ These two actions are intertwined because we try to only
+ retrieve the Issues on the current pagination page.
+ """
+ # We already got the issues, just display a slice of the visible ones.
+ limit_reached = False
+ for shard_limit_reached in self.search_limit_reached.values():
+ limit_reached |= shard_limit_reached
+ self.pagination = paginate.ArtifactPagination(
+ self.allowed_results,
+ self.items_per_page,
+ self.paginate_start,
+ self.project_name,
+ urls.ISSUE_LIST,
+ total_count=self.total_count,
+ limit_reached=limit_reached,
+ skipped=self.num_skipped_at_start)
+ self.visible_results = self.pagination.visible_results
+
+ # If we were not forced to look up visible users already, do it now.
+ self._LookupNeededUsers(self.visible_results)
+
+ def __repr__(self):
+ """Return a string that shows the internal state of this pipeline."""
+ if self.allowed_iids:
+ shown_allowed_iids = self.allowed_iids[:200]
+ else:
+ shown_allowed_iids = self.allowed_iids
+
+ if self.allowed_results:
+ shown_allowed_results = self.allowed_results[:200]
+ else:
+ shown_allowed_results = self.allowed_results
+
+ parts = [
+ 'allowed_iids: %r' % shown_allowed_iids,
+ 'allowed_results: %r' % shown_allowed_results,
+ 'len(visible_results): %r' % (
+ self.visible_results and len(self.visible_results))]
+ return '%s(%s)' % (self.__class__.__name__, '\n'.join(parts))
+
+
+def _CheckQuery(
+ cnxn, services, query, harmonized_config, project_ids,
+ member_of_all_projects, warnings=None):
+ """Parse the given query and report the first error or None."""
+ try:
+ query_ast = query2ast.ParseUserQuery(
+ query, '', query2ast.BUILTIN_ISSUE_FIELDS, harmonized_config,
+ warnings=warnings)
+ query_ast = ast2ast.PreprocessAST(
+ cnxn, query_ast, project_ids, services, harmonized_config,
+ is_member=member_of_all_projects)
+ except query2ast.InvalidQueryError as e:
+ return e.message
+ except ast2ast.MalformedQuery as e:
+ return e.message
+
+ return None
+
+
+def _MakeBackendCallback(func, *args):
+ # type: (Callable[[*Any], Any], *Any) -> Callable[[*Any], Any]
+ """Helper to store a particular function and argument set into a callback.
+
+ Args:
+ func: Function to callback.
+ *args: The arguments to pass into the function.
+
+ Returns:
+ Callback function based on specified arguments.
+ """
+ return lambda: func(*args)
+
+
+def _StartBackendSearch(
+ cnxn, query_project_names, query_project_ids, harmonized_config,
+ unfiltered_iids_dict, search_limit_reached_dict, nonviewable_iids,
+ error_responses, services, me_user_ids, logged_in_user_id, new_url_num,
+ subqueries, can, group_by_spec, sort_spec, warnings, use_cached_searches):
+ # type: (MonorailConnection, Sequence[str], Sequence[int],
+ # proto.tracker_pb2.ProjectIssueConfig,
+ # Mapping[Tuple(int, str), Sequence[int]],
+ # Mapping[Tuple(int, str), Sequence[bool]],
+ # Mapping[Tuple(int, str), Collection[int]], Sequence[Tuple(int, str)],
+ # Services, Sequence[int], int, int, Sequence[str], int, str, str,
+ # Sequence[Tuple(str, Sequence[str])], bool) ->
+ # Sequence[Tuple(int, Tuple(int, str),
+ # google.appengine.api.apiproxy_stub_map.UserRPC)]
+ """Request that our backends search and return a list of matching issue IDs.
+
+ Args:
+ cnxn: monorail connection to the database.
+ query_project_names: set of project names to search.
+ query_project_ids: list of project IDs to search.
+ harmonized_config: combined ProjectIssueConfig for all projects being
+ searched.
+ unfiltered_iids_dict: dict {shard_key: [iid, ...]} of unfiltered search
+ results to accumulate into. They need to be later filtered by
+ permissions and merged into filtered_iids_dict.
+ search_limit_reached_dict: dict {shard_key: [bool, ...]} to determine if
+ the search limit of any shard was reached.
+ nonviewable_iids: dict {shard_id: set(iid)} of restricted issues in the
+ projects being searched that the signed in user cannot view.
+ error_responses: shard_iids of shards that encountered errors.
+ services: connections to backends.
+ me_user_ids: Empty list when no user is logged in, or user ID of the logged
+ in user when doing an interactive search, or the viewed user ID when
+ viewing someone else's dashboard, or the subscribing user's ID when
+ evaluating subscriptions. And, any linked accounts.
+ logged_in_user_id: user_id of the logged in user, 0 otherwise
+ new_url_num: the number of issues for BackendSearchPipeline to query.
+ Computed based on pagination offset + number of items per page.
+ subqueries: split up list of query string segments.
+ can: "canned query" number to scope the user's search.
+ group_by_spec: string that lists the grouping order.
+ sort_spec: string that lists the sort order.
+ warnings: list to accumulate warning messages.
+ use_cached_searches: Bool for whether to use cached searches.
+
+ Returns:
+ A list of rpc_tuples that can be passed to _FinishBackendSearch to wait
+ on any remaining backend calls.
+
+ SIDE-EFFECTS:
+ Any data found in memcache is immediately put into unfiltered_iids_dict.
+ As the backends finish their work, _HandleBackendSearchResponse will update
+ unfiltered_iids_dict for those shards.
+
+ Any warnings produced throughout this process will be added to the list
+ warnings.
+ """
+ rpc_tuples = []
+ needed_shard_keys = set()
+ for subquery in subqueries:
+ subquery, warnings = searchpipeline.ReplaceKeywordsWithUserIDs(
+ me_user_ids, subquery)
+ warnings.extend(warnings)
+ for shard_id in range(settings.num_logical_shards):
+ needed_shard_keys.add((shard_id, subquery))
+
+ # 1. Get whatever we can from memcache. Cache hits are only kept if they are
+ # not already expired.
+ project_shard_timestamps = _GetProjectTimestamps(
+ query_project_ids, needed_shard_keys)
+
+ if use_cached_searches:
+ cached_unfiltered_iids_dict, cached_search_limit_reached_dict = (
+ _GetCachedSearchResults(
+ cnxn, query_project_ids, needed_shard_keys,
+ harmonized_config, project_shard_timestamps, services, me_user_ids,
+ can, group_by_spec, sort_spec, warnings))
+ unfiltered_iids_dict.update(cached_unfiltered_iids_dict)
+ search_limit_reached_dict.update(cached_search_limit_reached_dict)
+ for cache_hit_shard_key in unfiltered_iids_dict:
+ needed_shard_keys.remove(cache_hit_shard_key)
+
+ # 2. Each kept cache hit will have unfiltered IIDs, so we filter them by
+ # removing non-viewable IDs.
+ _GetNonviewableIIDs(
+ query_project_ids, logged_in_user_id,
+ set(range(settings.num_logical_shards)),
+ rpc_tuples, nonviewable_iids, project_shard_timestamps,
+ services.cache_manager.processed_invalidations_up_to,
+ use_cached_searches)
+
+ # 3. Hit backends for any shards that are still needed. When these results
+ # come back, they are also put into unfiltered_iids_dict.
+ for shard_key in needed_shard_keys:
+ rpc = _StartBackendSearchCall(
+ query_project_names,
+ shard_key,
+ services.cache_manager.processed_invalidations_up_to,
+ me_user_ids,
+ logged_in_user_id,
+ new_url_num,
+ can=can,
+ sort_spec=sort_spec,
+ group_by_spec=group_by_spec)
+ rpc_tuple = (time.time(), shard_key, rpc)
+ rpc.callback = _MakeBackendCallback(
+ _HandleBackendSearchResponse, query_project_names, rpc_tuple,
+ rpc_tuples, settings.backend_retries, unfiltered_iids_dict,
+ search_limit_reached_dict,
+ services.cache_manager.processed_invalidations_up_to, error_responses,
+ me_user_ids, logged_in_user_id, new_url_num, can, sort_spec,
+ group_by_spec)
+ rpc_tuples.append(rpc_tuple)
+
+ return rpc_tuples
+
+
+def _FinishBackendSearch(rpc_tuples):
+ """Wait for all backend calls to complete, including any retries."""
+ while rpc_tuples:
+ active_rpcs = [rpc for (_time, _shard_key, rpc) in rpc_tuples]
+ # Wait for any active RPC to complete. It's callback function will
+ # automatically be called.
+ finished_rpc = real_wait_any(active_rpcs)
+ # Figure out which rpc_tuple finished and remove it from our list.
+ for rpc_tuple in rpc_tuples:
+ _time, _shard_key, rpc = rpc_tuple
+ if rpc == finished_rpc:
+ rpc_tuples.remove(rpc_tuple)
+ break
+ else:
+ raise ValueError('We somehow finished an RPC that is not in rpc_tuples')
+
+
+def real_wait_any(active_rpcs):
+ """Work around the blocking nature of wait_any().
+
+ wait_any() checks for any finished RPCs, and returns one if found.
+ If no RPC is finished, it simply blocks on the last RPC in the list.
+ This is not the desired behavior because we are not able to detect
+ FAST-FAIL RPC results and retry them if wait_any() is blocked on a
+ request that is taking a long time to do actual work.
+
+ Instead, we do the same check, without blocking on any individual RPC.
+ """
+ if settings.local_mode:
+ # The development server has very different code for RPCs than the
+ # code used in the hosted environment.
+ return apiproxy_stub_map.UserRPC.wait_any(active_rpcs)
+ while True:
+ finished, _ = apiproxy_stub_map.UserRPC._UserRPC__check_one(active_rpcs)
+ if finished:
+ return finished
+ time.sleep(DELAY_BETWEEN_RPC_COMPLETION_POLLS)
+
+def _GetProjectTimestamps(query_project_ids, needed_shard_keys):
+ """Get a dict of modified_ts values for all specified project-shards."""
+ project_shard_timestamps = {}
+ if query_project_ids:
+ keys = []
+ for pid in query_project_ids:
+ for sid, _subquery in needed_shard_keys:
+ keys.append('%d;%d' % (pid, sid))
+ else:
+ keys = [('all;%d' % sid)
+ for sid, _subquery in needed_shard_keys]
+
+ timestamps_for_project = memcache.get_multi(
+ keys=keys, namespace=settings.memcache_namespace)
+ for key, timestamp in timestamps_for_project.items():
+ pid_str, sid_str = key.split(';')
+ if pid_str == 'all':
+ project_shard_timestamps['all', int(sid_str)] = timestamp
+ else:
+ project_shard_timestamps[int(pid_str), int(sid_str)] = timestamp
+
+ return project_shard_timestamps
+
+
+def _GetNonviewableIIDs(
+ query_project_ids, logged_in_user_id, needed_shard_ids, rpc_tuples,
+ nonviewable_iids, project_shard_timestamps, invalidation_timestep,
+ use_cached_searches):
+ """Build a set of at-risk IIDs, and accumulate RPCs to get uncached ones."""
+ if query_project_ids:
+ keys = []
+ for pid in query_project_ids:
+ for sid in needed_shard_ids:
+ keys.append('%d;%d;%d' % (pid, logged_in_user_id, sid))
+ else:
+ keys = [
+ ('all;%d;%d' % (logged_in_user_id, sid)) for sid in needed_shard_ids
+ ]
+
+ if use_cached_searches:
+ cached_dict = memcache.get_multi(
+ keys, key_prefix='nonviewable:', namespace=settings.memcache_namespace)
+ else:
+ cached_dict = {}
+
+ for sid in needed_shard_ids:
+ if query_project_ids:
+ for pid in query_project_ids:
+ _AccumulateNonviewableIIDs(
+ pid, logged_in_user_id, sid, cached_dict, nonviewable_iids,
+ project_shard_timestamps, rpc_tuples, invalidation_timestep)
+ else:
+ _AccumulateNonviewableIIDs(
+ None, logged_in_user_id, sid, cached_dict, nonviewable_iids,
+ project_shard_timestamps, rpc_tuples, invalidation_timestep)
+
+
+def _AccumulateNonviewableIIDs(
+ pid, logged_in_user_id, sid, cached_dict, nonviewable_iids,
+ project_shard_timestamps, rpc_tuples, invalidation_timestep):
+ """Use one of the retrieved cache entries or call a backend if needed."""
+ if pid is None:
+ key = 'all;%d;%d' % (logged_in_user_id, sid)
+ else:
+ key = '%d;%d;%d' % (pid, logged_in_user_id, sid)
+
+ if key in cached_dict:
+ issue_ids, cached_ts = cached_dict.get(key)
+ modified_ts = project_shard_timestamps.get((pid, sid))
+ if modified_ts is None or modified_ts > cached_ts:
+ logging.info('nonviewable too stale on (project %r, shard %r)',
+ pid, sid)
+ else:
+ logging.info('adding %d nonviewable issue_ids', len(issue_ids))
+ nonviewable_iids[sid] = set(issue_ids)
+
+ if sid not in nonviewable_iids:
+ logging.info('nonviewable for %r not found', key)
+ logging.info('starting backend call for nonviewable iids %r', key)
+ rpc = _StartBackendNonviewableCall(
+ pid, logged_in_user_id, sid, invalidation_timestep)
+ rpc_tuple = (time.time(), sid, rpc)
+ rpc.callback = _MakeBackendCallback(
+ _HandleBackendNonviewableResponse, pid, logged_in_user_id, sid,
+ rpc_tuple, rpc_tuples, settings.backend_retries, nonviewable_iids,
+ invalidation_timestep)
+ rpc_tuples.append(rpc_tuple)
+
+
+def _GetCachedSearchResults(
+ cnxn, query_project_ids, needed_shard_keys, harmonized_config,
+ project_shard_timestamps, services, me_user_ids, can, group_by_spec,
+ sort_spec, warnings):
+ """Return a dict of cached search results that are not already stale.
+
+ If it were not for cross-project search, we would simply cache when we do a
+ search and then invalidate when an issue is modified. But, with
+ cross-project search we don't know all the memcache entries that would
+ need to be invalidated. So, instead, we write the search result cache
+ entries and then an initial modified_ts value for each project if it was
+ not already there. And, when we update an issue we write a new
+ modified_ts entry, which implicitly invalidate all search result
+ cache entries that were written earlier because they are now stale. When
+ reading from the cache, we ignore any query project with modified_ts
+ after its search result cache timestamp, because it is stale.
+
+ Args:
+ cnxn: monorail connection to the database.
+ query_project_ids: list of project ID numbers for all projects being
+ searched.
+ needed_shard_keys: set of shard keys that need to be checked.
+ harmonized_config: ProjectIsueConfig with combined information for all
+ projects involved in this search.
+ project_shard_timestamps: a dict {(project_id, shard_id): timestamp, ...}
+ that tells when each shard was last invalidated.
+ services: connections to backends.
+ me_user_ids: Empty list when no user is logged in, or user ID of the logged
+ in user when doing an interactive search, or the viewed user ID when
+ viewing someone else's dashboard, or the subscribing user's ID when
+ evaluating subscriptions. And, any linked accounts.
+ can: "canned query" number to scope the user's search.
+ group_by_spec: string that lists the grouping order.
+ sort_spec: string that lists the sort order.
+ warnings: list to accumulate warning messages.
+
+
+ Returns:
+ Tuple consisting of:
+ A dictionary {shard_id: [issue_id, ...], ...} of unfiltered search result
+ issue IDs. Only shard_ids found in memcache will be in that dictionary.
+ The result issue IDs must be permission checked before they can be
+ considered to be part of the user's result set.
+ A dictionary {shard_id: bool, ...}. The boolean is set to True if
+ the search results limit of the shard is hit.
+ """
+ projects_str = ','.join(str(pid) for pid in sorted(query_project_ids))
+ projects_str = projects_str or 'all'
+ canned_query = savedqueries_helpers.SavedQueryIDToCond(
+ cnxn, services.features, can)
+ canned_query, warnings = searchpipeline.ReplaceKeywordsWithUserIDs(
+ me_user_ids, canned_query)
+ warnings.extend(warnings)
+
+ sd = sorting.ComputeSortDirectives(
+ harmonized_config, group_by_spec, sort_spec)
+ sd_str = ' '.join(sd)
+ memcache_key_prefix = '%s;%s' % (projects_str, canned_query)
+ limit_reached_key_prefix = '%s;%s' % (projects_str, canned_query)
+
+ cached_dict = memcache.get_multi(
+ ['%s;%s;%s;%d' % (memcache_key_prefix, subquery, sd_str, sid)
+ for sid, subquery in needed_shard_keys],
+ namespace=settings.memcache_namespace)
+ cached_search_limit_reached_dict = memcache.get_multi(
+ ['%s;%s;%s;search_limit_reached;%d' % (
+ limit_reached_key_prefix, subquery, sd_str, sid)
+ for sid, subquery in needed_shard_keys],
+ namespace=settings.memcache_namespace)
+
+ unfiltered_dict = {}
+ search_limit_reached_dict = {}
+ for shard_key in needed_shard_keys:
+ shard_id, subquery = shard_key
+ memcache_key = '%s;%s;%s;%d' % (
+ memcache_key_prefix, subquery, sd_str, shard_id)
+ limit_reached_key = '%s;%s;%s;search_limit_reached;%d' % (
+ limit_reached_key_prefix, subquery, sd_str, shard_id)
+ if memcache_key not in cached_dict:
+ logging.info('memcache miss on shard %r', shard_key)
+ continue
+
+ cached_iids, cached_ts = cached_dict[memcache_key]
+ if cached_search_limit_reached_dict.get(limit_reached_key):
+ search_limit_reached, _ = cached_search_limit_reached_dict[
+ limit_reached_key]
+ else:
+ search_limit_reached = False
+
+ stale = False
+ if query_project_ids:
+ for project_id in query_project_ids:
+ modified_ts = project_shard_timestamps.get((project_id, shard_id))
+ if modified_ts is None or modified_ts > cached_ts:
+ stale = True
+ logging.info('memcache too stale on shard %r because of %r',
+ shard_id, project_id)
+ break
+ else:
+ modified_ts = project_shard_timestamps.get(('all', shard_id))
+ if modified_ts is None or modified_ts > cached_ts:
+ stale = True
+ logging.info('memcache too stale on shard %r because of all',
+ shard_id)
+
+ if not stale:
+ unfiltered_dict[shard_key] = cached_iids
+ search_limit_reached_dict[shard_key] = search_limit_reached
+
+ return unfiltered_dict, search_limit_reached_dict
+
+
+def _MakeBackendRequestHeaders(failfast):
+ headers = {
+ # This is needed to allow frontends to talk to backends without going
+ # through a login screen on googleplex.com.
+ # http://wiki/Main/PrometheusInternal#Internal_Applications_and_APIs
+ 'X-URLFetch-Service-Id': 'GOOGLEPLEX',
+ }
+ if failfast:
+ headers['X-AppEngine-FailFast'] = 'Yes'
+ return headers
+
+
+def _StartBackendSearchCall(
+ query_project_names,
+ shard_key,
+ invalidation_timestep,
+ me_user_ids,
+ logged_in_user_id,
+ new_url_num,
+ can=None,
+ sort_spec=None,
+ group_by_spec=None,
+ deadline=None,
+ failfast=True):
+ # type: (Sequence[str], Tuple(int, str), int, Sequence[int], int,
+ # int, str, str, int, bool) ->
+ # google.appengine.api.apiproxy_stub_map.UserRPC
+ """Ask a backend to query one shard of the database.
+
+ Args:
+ query_project_names: List of project names queried.
+ shard_key: Tuple specifying which DB shard to query.
+ invalidation_timestep: int timestep to use keep cached items fresh.
+ me_user_ids: Empty list when no user is logged in, or user ID of the logged
+ in user when doing an interactive search, or the viewed user ID when
+ viewing someone else's dashboard, or the subscribing user's ID when
+ evaluating subscriptions. And, any linked accounts.
+ logged_in_user_id: Id of the logged in user.
+ new_url_num: the number of issues for BackendSearchPipeline to query.
+ Computed based on pagination offset + number of items per page.
+ can: Id of th canned query to use.
+ sort_spec: Str specifying how issues should be sorted.
+ group_by_spec: Str specifying how issues should be grouped.
+ deadline: Max time for the RPC to take before failing.
+ failfast: Whether to set the X-AppEngine-FailFast request header.
+
+ Returns:
+ UserRPC for the created RPC call.
+ """
+ shard_id, subquery = shard_key
+ backend_host = modules.get_hostname(module='besearch')
+ url = 'http://%s%s' % (
+ backend_host,
+ framework_helpers.FormatURL(
+ [],
+ urls.BACKEND_SEARCH,
+ projects=','.join(query_project_names),
+ q=subquery,
+ start=0,
+ num=new_url_num,
+ can=can,
+ sort=sort_spec,
+ groupby=group_by_spec,
+ logged_in_user_id=logged_in_user_id,
+ me_user_ids=','.join(str(uid) for uid in me_user_ids),
+ shard_id=shard_id,
+ invalidation_timestep=invalidation_timestep))
+ logging.info('\n\nCalling backend: %s', url)
+ rpc = urlfetch.create_rpc(
+ deadline=deadline or settings.backend_deadline)
+ headers = _MakeBackendRequestHeaders(failfast)
+ # follow_redirects=False is needed to avoid a login screen on googleplex.
+ urlfetch.make_fetch_call(rpc, url, follow_redirects=False, headers=headers)
+ return rpc
+
+
+def _StartBackendNonviewableCall(
+ project_id, logged_in_user_id, shard_id, invalidation_timestep,
+ deadline=None, failfast=True):
+ """Ask a backend to query one shard of the database."""
+ backend_host = modules.get_hostname(module='besearch')
+ url = 'http://%s%s' % (backend_host, framework_helpers.FormatURL(
+ None, urls.BACKEND_NONVIEWABLE,
+ project_id=project_id or '',
+ logged_in_user_id=logged_in_user_id or '',
+ shard_id=shard_id,
+ invalidation_timestep=invalidation_timestep))
+ logging.info('Calling backend nonviewable: %s', url)
+ rpc = urlfetch.create_rpc(deadline=deadline or settings.backend_deadline)
+ headers = _MakeBackendRequestHeaders(failfast)
+ # follow_redirects=False is needed to avoid a login screen on googleplex.
+ urlfetch.make_fetch_call(rpc, url, follow_redirects=False, headers=headers)
+ return rpc
+
+
+def _HandleBackendSearchResponse(
+ query_project_names, rpc_tuple, rpc_tuples, remaining_retries,
+ unfiltered_iids, search_limit_reached, invalidation_timestep,
+ error_responses, me_user_ids, logged_in_user_id, new_url_num, can,
+ sort_spec, group_by_spec):
+ # type: (Sequence[str], Tuple(int, Tuple(int, str),
+ # google.appengine.api.apiproxy_stub_map.UserRPC),
+ # Sequence[Tuple(int, Tuple(int, str),
+ # google.appengine.api.apiproxy_stub_map.UserRPC)],
+ # int, Mapping[Tuple(int, str), Sequence[int]],
+ # Mapping[Tuple(int, str), bool], int, Collection[Tuple(int, str)],
+ # Sequence[int], int, int, int, str, str) -> None
+ #
+ """Process one backend response and retry if there was an error.
+
+ SIDE EFFECTS: This function edits many of the passed in parameters in place.
+ For example, search_limit_reached and unfiltered_iids are updated with
+ response data from the RPC, keyed by shard_key.
+
+ Args:
+ query_project_names: List of projects to query.
+ rpc_tuple: Tuple containing an RPC response object, the time it happened,
+ and what shard the RPC was queried against.
+ rpc_tuples: List of RPC responses to mutate with any retry responses that
+ heppened.
+ remaining_retries: Number of times left to retry.
+ unfiltered_iids: Dict of Issue ids, before they've been filtered by
+ permissions.
+ search_limit_reached: Dict of whether the search limit for a particular
+ shard has been hit.
+ invalidation_timestep: int timestep to use keep cached items fresh.
+ error_responses:
+ me_user_ids: List of relevant user IDs. ie: the currently logged in user
+ and linked account IDs if applicable.
+ logged_in_user_id: Logged in user's ID.
+ new_url_num: the number of issues for BackendSearchPipeline to query.
+ Computed based on pagination offset + number of items per page.
+ can: Canned query ID to use.
+ sort_spec: str specifying how issues should be sorted.
+ group_by_spec: str specifying how issues should be grouped.
+ """
+ start_time, shard_key, rpc = rpc_tuple
+ duration_sec = time.time() - start_time
+
+ try:
+ response = rpc.get_result()
+ logging.info('call to backend took %d sec', duration_sec)
+ # Note that response.content has "})]'\n" prepended to it.
+ json_content = response.content[5:]
+ logging.info('got json text: %r length %r',
+ json_content[:framework_constants.LOGGING_MAX_LENGTH],
+ len(json_content))
+ if json_content == '':
+ raise Exception('Fast fail')
+ json_data = json.loads(json_content)
+ unfiltered_iids[shard_key] = json_data['unfiltered_iids']
+ search_limit_reached[shard_key] = json_data['search_limit_reached']
+ if json_data.get('error'):
+ # Don't raise an exception, just log, because these errors are more like
+ # 400s than 500s, and shouldn't be retried.
+ logging.error('Backend shard %r returned error "%r"' % (
+ shard_key, json_data.get('error')))
+ error_responses.add(shard_key)
+
+ except Exception as e:
+ if duration_sec > FAIL_FAST_LIMIT_SEC: # Don't log fail-fast exceptions.
+ logging.exception(e)
+ if not remaining_retries:
+ logging.error('backend search retries exceeded')
+ error_responses.add(shard_key)
+ return # Used all retries, so give up.
+
+ if duration_sec >= settings.backend_deadline:
+ logging.error('backend search on %r took too long', shard_key)
+ error_responses.add(shard_key)
+ return # That backend shard is overloaded, so give up.
+
+ logging.error('backend call for shard %r failed, retrying', shard_key)
+ retry_rpc = _StartBackendSearchCall(
+ query_project_names,
+ shard_key,
+ invalidation_timestep,
+ me_user_ids,
+ logged_in_user_id,
+ new_url_num,
+ can=can,
+ sort_spec=sort_spec,
+ group_by_spec=group_by_spec,
+ failfast=remaining_retries > 2)
+ retry_rpc_tuple = (time.time(), shard_key, retry_rpc)
+ retry_rpc.callback = _MakeBackendCallback(
+ _HandleBackendSearchResponse, query_project_names, retry_rpc_tuple,
+ rpc_tuples, remaining_retries - 1, unfiltered_iids,
+ search_limit_reached, invalidation_timestep, error_responses,
+ me_user_ids, logged_in_user_id, new_url_num, can, sort_spec,
+ group_by_spec)
+ rpc_tuples.append(retry_rpc_tuple)
+
+
+def _HandleBackendNonviewableResponse(
+ project_id, logged_in_user_id, shard_id, rpc_tuple, rpc_tuples,
+ remaining_retries, nonviewable_iids, invalidation_timestep):
+ """Process one backend response and retry if there was an error."""
+ start_time, shard_id, rpc = rpc_tuple
+ duration_sec = time.time() - start_time
+
+ try:
+ response = rpc.get_result()
+ logging.info('call to backend nonviewable took %d sec', duration_sec)
+ # Note that response.content has "})]'\n" prepended to it.
+ json_content = response.content[5:]
+ logging.info('got json text: %r length %r',
+ json_content[:framework_constants.LOGGING_MAX_LENGTH],
+ len(json_content))
+ if json_content == '':
+ raise Exception('Fast fail')
+ json_data = json.loads(json_content)
+ nonviewable_iids[shard_id] = set(json_data['nonviewable'])
+
+ except Exception as e:
+ if duration_sec > FAIL_FAST_LIMIT_SEC: # Don't log fail-fast exceptions.
+ logging.exception(e)
+
+ if not remaining_retries:
+ logging.warn('Used all retries, so give up on shard %r', shard_id)
+ return
+
+ if duration_sec >= settings.backend_deadline:
+ logging.error('nonviewable call on %r took too long', shard_id)
+ return # That backend shard is overloaded, so give up.
+
+ logging.error(
+ 'backend nonviewable call for shard %r;%r;%r failed, retrying',
+ project_id, logged_in_user_id, shard_id)
+ retry_rpc = _StartBackendNonviewableCall(
+ project_id, logged_in_user_id, shard_id, invalidation_timestep,
+ failfast=remaining_retries > 2)
+ retry_rpc_tuple = (time.time(), shard_id, retry_rpc)
+ retry_rpc.callback = _MakeBackendCallback(
+ _HandleBackendNonviewableResponse, project_id, logged_in_user_id,
+ shard_id, retry_rpc_tuple, rpc_tuples, remaining_retries - 1,
+ nonviewable_iids, invalidation_timestep)
+ rpc_tuples.append(retry_rpc_tuple)
+
+
+def _TotalLength(sharded_iids):
+ """Return the total length of all issue_iids lists."""
+ return sum(len(issue_iids) for issue_iids in sharded_iids.values())
+
+
+def _ReverseShards(sharded_iids):
+ """Reverse each issue_iids list in place."""
+ for shard_key in sharded_iids:
+ sharded_iids[shard_key].reverse()
+
+
+def _TrimEndShardedIIDs(sharded_iids, sample_iid_tuples, num_needed):
+ """Trim the IIDs to keep at least num_needed items.
+
+ Args:
+ sharded_iids: dict {shard_key: issue_id_list} for search results. This is
+ modified in place to remove some trailing issue IDs.
+ sample_iid_tuples: list of (iid, shard_key) from a sorted list of sample
+ issues.
+ num_needed: int minimum total number of items to keep. Some IIDs that are
+ known to belong in positions > num_needed will be trimmed off.
+
+ Returns:
+ The total number of IIDs removed from the IID lists.
+ """
+ # 1. Get (sample_iid, position_in_shard) for each sample.
+ sample_positions = _CalcSamplePositions(sharded_iids, sample_iid_tuples)
+
+ # 2. Walk through the samples, computing a combined lower bound at each
+ # step until we know that we have passed at least num_needed IIDs.
+ lower_bound_per_shard = {}
+ excess_samples = []
+ for i in range(len(sample_positions)):
+ _sample_iid, sample_shard_key, pos = sample_positions[i]
+ lower_bound_per_shard[sample_shard_key] = pos
+ overall_lower_bound = sum(lower_bound_per_shard.values())
+ if overall_lower_bound >= num_needed:
+ excess_samples = sample_positions[i + 1:]
+ break
+ else:
+ return 0 # We went through all samples and never reached num_needed.
+
+ # 3. Truncate each shard at the first excess sample in that shard.
+ already_trimmed = set()
+ num_trimmed = 0
+ for _sample_iid, sample_shard_key, pos in excess_samples:
+ if sample_shard_key not in already_trimmed:
+ num_trimmed += len(sharded_iids[sample_shard_key]) - pos
+ sharded_iids[sample_shard_key] = sharded_iids[sample_shard_key][:pos]
+ already_trimmed.add(sample_shard_key)
+
+ return num_trimmed
+
+
+# TODO(jrobbins): Convert this to a python generator.
+def _CalcSamplePositions(sharded_iids, sample_iids):
+ """Return [(iid, shard_key, position_in_shard), ...] for each sample."""
+ # We keep track of how far index() has scanned in each shard to avoid
+ # starting over at position 0 when looking for the next sample in
+ # the same shard.
+ scan_positions = collections.defaultdict(lambda: 0)
+ sample_positions = []
+ for sample_iid, sample_shard_key in sample_iids:
+ try:
+ pos = sharded_iids.get(sample_shard_key, []).index(
+ sample_iid, scan_positions[sample_shard_key])
+ scan_positions[sample_shard_key] = pos
+ sample_positions.append((sample_iid, sample_shard_key, pos))
+ except ValueError:
+ pass
+
+ return sample_positions
+
+
+def _SortIssues(issues, config, users_by_id, group_by_spec, sort_spec):
+ """Sort the found issues based on the request and config values.
+
+ Args:
+ issues: A list of issues to be sorted.
+ config: A ProjectIssueConfig that could impact sort order.
+ users_by_id: dictionary {user_id: user_view,...} for all users who
+ participate in any issue in the entire list.
+ group_by_spec: string that lists the grouping order
+ sort_spec: string that lists the sort order
+
+
+ Returns:
+ A sorted list of issues, based on parameters from mr and config.
+ """
+ issues = sorting.SortArtifacts(
+ issues, config, tracker_helpers.SORTABLE_FIELDS,
+ tracker_helpers.SORTABLE_FIELDS_POSTPROCESSORS, group_by_spec,
+ sort_spec, users_by_id=users_by_id)
+ return issues