services/chart_svc.py

# Copyright 2018 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

"""A service for querying data for charts.

Functions for querying the IssueSnapshot table and associated join tables.
"""
from __future__ import print_function
from __future__ import division
from __future__ import absolute_import

import logging
import settings
import time

from features import hotlist_helpers
from framework import framework_helpers
from framework import sql
from search import search_helpers
from tracker import tracker_bizobj
from tracker import tracker_helpers
from search import query2ast
from search import ast2select
from search import ast2ast


ISSUESNAPSHOT_TABLE_NAME = 'IssueSnapshot'
ISSUESNAPSHOT2CC_TABLE_NAME = 'IssueSnapshot2Cc'
ISSUESNAPSHOT2COMPONENT_TABLE_NAME = 'IssueSnapshot2Component'
ISSUESNAPSHOT2LABEL_TABLE_NAME = 'IssueSnapshot2Label'

ISSUESNAPSHOT_COLS = ['id', 'issue_id', 'shard', 'project_id', 'local_id',
    'reporter_id', 'owner_id', 'status_id', 'period_start', 'period_end',
    'is_open']
ISSUESNAPSHOT2CC_COLS = ['issuesnapshot_id', 'cc_id']
ISSUESNAPSHOT2COMPONENT_COLS = ['issuesnapshot_id', 'component_id']
ISSUESNAPSHOT2LABEL_COLS = ['issuesnapshot_id', 'label_id']


class ChartService(object):
  """Class for querying chart data."""

  def __init__(self, config_service):
    """Constructor for ChartService.

    Args:
      config_service (ConfigService): An instance of ConfigService.
    """
    self.config_service = config_service

    # Set up SQL table objects.
    self.issuesnapshot_tbl = sql.SQLTableManager(ISSUESNAPSHOT_TABLE_NAME)
    self.issuesnapshot2cc_tbl = sql.SQLTableManager(
        ISSUESNAPSHOT2CC_TABLE_NAME)
    self.issuesnapshot2component_tbl = sql.SQLTableManager(
        ISSUESNAPSHOT2COMPONENT_TABLE_NAME)
    self.issuesnapshot2label_tbl = sql.SQLTableManager(
        ISSUESNAPSHOT2LABEL_TABLE_NAME)

  def QueryIssueSnapshots(self, cnxn, services, unixtime, effective_ids,
                          project, perms, group_by=None, label_prefix=None,
                          query=None, canned_query=None, hotlist=None):
    """Queries historical issue counts grouped by label or component.

    Args:
      cnxn: A MonorailConnection instance.
      services: A Services instance.
      unixtime: An integer representing the Unix time in seconds.
      effective_ids: The effective User IDs associated with the current user.
      project: A project object representing the current project.
      perms: A permissions object associated with the current user.
      group_by (str, optional): Which dimension to group by. Values can
        be 'label', 'component', or None, in which case no grouping will
        be applied.
      label_prefix: Required when group_by is 'label.' Will limit the query to
        only labels with the specified prefix (for example 'Pri').
      query (str, optional): A query string from the request to apply to
        the snapshot query.
      canned_query (str, optional): Parsed canned query applied to the query
        scope.
      hotlist (Hotlist, optional): Hotlist to search under (in lieu of project).

    Returns:
      1. A dict of {'2nd dimension or "total"': number of occurences}.
      2. A list of any unsupported query conditions in query.
      3. A boolean that is true if any results were capped.
    """
    if hotlist:
      # TODO(jeffcarp): Get project_ids in a more efficient manner. We can
      #   query for "SELECT DISTINCT(project_id)" for all issues in hotlist.
      issues_list = services.issue.GetIssues(cnxn,
          [hotlist_issue.issue_id for hotlist_issue in hotlist.items])
      hotlist_issues_project_ids = hotlist_helpers.GetAllProjectsOfIssues(
          [issue for issue in issues_list])
      config_list = hotlist_helpers.GetAllConfigsOfProjects(
          cnxn, hotlist_issues_project_ids, services)
      project_config = tracker_bizobj.HarmonizeConfigs(config_list)
    else:
      project_config = services.config.GetProjectConfig(cnxn,
          project.project_id)

    if project:
      project_ids = [project.project_id]
    else:
      project_ids = hotlist_issues_project_ids

    try:
      query_left_joins, query_where, unsupported_conds = self._QueryToWhere(
          cnxn, services, project_config, query, canned_query, project_ids)
    except ast2select.NoPossibleResults:
      return {}, ['Invalid query.'], False

    restricted_label_ids = search_helpers.GetPersonalAtRiskLabelIDs(
      cnxn, None, self.config_service, effective_ids, project, perms)

    left_joins = [
      ('Issue ON IssueSnapshot.issue_id = Issue.id', []),
    ]

    if restricted_label_ids:
      left_joins.append(
        (('Issue2Label AS Forbidden_label'
          ' ON Issue.id = Forbidden_label.issue_id'
          ' AND Forbidden_label.label_id IN (%s)' % (
            sql.PlaceHolders(restricted_label_ids)
        )), restricted_label_ids))

    if effective_ids:
      left_joins.append(
        ('Issue2Cc AS I2cc'
         ' ON Issue.id = I2cc.issue_id'
         ' AND I2cc.cc_id IN (%s)' % sql.PlaceHolders(effective_ids),
         effective_ids))

    # TODO(jeffcarp): Handle case where there are issues with no labels.
    where = [
      ('IssueSnapshot.period_start <= %s', [unixtime]),
      ('IssueSnapshot.period_end > %s', [unixtime]),
      ('Issue.is_spam = %s', [False]),
      ('Issue.deleted = %s', [False]),
    ]
    if project_ids:
      where.append(
        ('IssueSnapshot.project_id IN (%s)' % sql.PlaceHolders(project_ids),
          project_ids))

    forbidden_label_clause = 'Forbidden_label.label_id IS NULL'
    if effective_ids:
      if restricted_label_ids:
        forbidden_label_clause = ' OR %s' % forbidden_label_clause
      else:
        forbidden_label_clause =  ''

      where.append(
        ((
          '(Issue.reporter_id IN (%s)'
          ' OR Issue.owner_id IN (%s)'
          ' OR I2cc.cc_id IS NOT NULL'
          '%s)'
        ) % (
          sql.PlaceHolders(effective_ids), sql.PlaceHolders(effective_ids),
          forbidden_label_clause
        ),
          list(effective_ids) + list(effective_ids)
        ))
    else:
      where.append((forbidden_label_clause, []))

    if group_by == 'component':
      cols = ['Comp.path', 'COUNT(IssueSnapshot.issue_id)']
      left_joins.extend([
        (('IssueSnapshot2Component AS Is2c ON'
          ' Is2c.issuesnapshot_id = IssueSnapshot.id'), []),
        ('ComponentDef AS Comp ON Comp.id = Is2c.component_id', []),
      ])
      group_by = ['Comp.path']
    elif group_by == 'label':
      cols = ['Lab.label', 'COUNT(IssueSnapshot.issue_id)']
      left_joins.extend([
        (('IssueSnapshot2Label AS Is2l'
          ' ON Is2l.issuesnapshot_id = IssueSnapshot.id'), []),
        ('LabelDef AS Lab ON Lab.id = Is2l.label_id', []),
      ])

      if not label_prefix:
        raise ValueError('`label_prefix` required when grouping by label.')

      # TODO(jeffcarp): If LookupIDsOfLabelsMatching() is called on output,
      # ensure regex is case-insensitive.
      where.append(('LOWER(Lab.label) LIKE %s', [label_prefix.lower() + '-%']))
      group_by = ['Lab.label']
    elif group_by == 'open':
      cols = ['IssueSnapshot.is_open',
        'COUNT(IssueSnapshot.issue_id) AS issue_count']
      group_by = ['IssueSnapshot.is_open']
    elif group_by == 'status':
      left_joins.append(('StatusDef AS Stats ON ' \
        'Stats.id = IssueSnapshot.status_id', []))
      cols = ['Stats.status', 'COUNT(IssueSnapshot.issue_id)']
      group_by = ['Stats.status']
    elif group_by == 'owner':
      cols = ['IssueSnapshot.owner_id', 'COUNT(IssueSnapshot.issue_id)']
      group_by = ['IssueSnapshot.owner_id']
    elif not group_by:
      cols = ['IssueSnapshot.issue_id']
    else:
      raise ValueError('`group_by` must be label, component, ' \
        'open, status, owner or None.')

    if query_left_joins:
      left_joins.extend(query_left_joins)

    if query_where:
      where.extend(query_where)

    if hotlist:
      left_joins.extend([
        (('IssueSnapshot2Hotlist AS Is2h'
          ' ON Is2h.issuesnapshot_id = IssueSnapshot.id'
          ' AND Is2h.hotlist_id = %s'), [hotlist.hotlist_id]),
      ])
      where.append(
        ('Is2h.hotlist_id = %s', [hotlist.hotlist_id]))

    promises = []

    for shard_id in range(settings.num_logical_shards):
      count_stmt, stmt_args = self._BuildSnapshotQuery(cols=cols,
          where=where, joins=left_joins, group_by=group_by,
          shard_id=shard_id)
      promises.append(framework_helpers.Promise(cnxn.Execute,
          count_stmt, stmt_args, shard_id=shard_id))

    shard_values_dict = {}

    search_limit_reached = False

    for promise in promises:
      # Wait for each query to complete and add it to the dict.
      shard_values = list(promise.WaitAndGetValue())

      if not shard_values:
        continue
      if group_by:
        for name, count in shard_values:
          if count >= settings.chart_query_max_rows:
            search_limit_reached = True

          shard_values_dict.setdefault(name, 0)
          shard_values_dict[name] += count
      else:
        if shard_values[0][0] >= settings.chart_query_max_rows:
            search_limit_reached = True

        shard_values_dict.setdefault('total', 0)
        shard_values_dict['total'] += shard_values[0][0]

    unsupported_field_names = list(set([
        field.field_name
        for cond in unsupported_conds
        for field in cond.field_defs
    ]))

    return shard_values_dict, unsupported_field_names, search_limit_reached

  def StoreIssueSnapshots(self, cnxn, issues, commit=True):
    """Adds an IssueSnapshot and updates the previous one for each issue."""
    for issue in issues:
      right_now = self._currentTime()

      # Update previous snapshot of current issue's end time to right now.
      self.issuesnapshot_tbl.Update(cnxn,
          delta={'period_end': right_now},
          where=[('IssueSnapshot.issue_id = %s', [issue.issue_id]),
            ('IssueSnapshot.period_end = %s',
              [settings.maximum_snapshot_period_end])],
          commit=commit)

      config = self.config_service.GetProjectConfig(cnxn, issue.project_id)
      period_end = settings.maximum_snapshot_period_end
      is_open = tracker_helpers.MeansOpenInProject(
        tracker_bizobj.GetStatus(issue), config)
      shard = issue.issue_id % settings.num_logical_shards
      status = tracker_bizobj.GetStatus(issue)
      status_id = self.config_service.LookupStatusID(
          cnxn, issue.project_id, status) or None
      owner_id = tracker_bizobj.GetOwnerId(issue) or None

      issuesnapshot_rows = [(issue.issue_id, shard, issue.project_id,
        issue.local_id, issue.reporter_id, owner_id, status_id, right_now,
        period_end, is_open)]

      ids = self.issuesnapshot_tbl.InsertRows(
          cnxn, ISSUESNAPSHOT_COLS[1:],
          issuesnapshot_rows,
          replace=True, commit=commit,
          return_generated_ids=True)
      issuesnapshot_id = ids[0]

      # Add all labels to IssueSnapshot2Label.
      label_rows = [
          (issuesnapshot_id,
           self.config_service.LookupLabelID(cnxn, issue.project_id, label))
          for label in tracker_bizobj.GetLabels(issue)
      ]
      self.issuesnapshot2label_tbl.InsertRows(
          cnxn, ISSUESNAPSHOT2LABEL_COLS,
          label_rows, replace=True, commit=commit)

      # Add all CCs to IssueSnapshot2Cc.
      cc_rows = [
        (issuesnapshot_id, cc_id)
        for cc_id in tracker_bizobj.GetCcIds(issue)
      ]
      self.issuesnapshot2cc_tbl.InsertRows(
          cnxn, ISSUESNAPSHOT2CC_COLS,
          cc_rows,
          replace=True, commit=commit)

      # Add all components to IssueSnapshot2Component.
      component_rows = [
        (issuesnapshot_id, component_id)
        for component_id in issue.component_ids
      ]
      self.issuesnapshot2component_tbl.InsertRows(
          cnxn, ISSUESNAPSHOT2COMPONENT_COLS,
          component_rows,
          replace=True, commit=commit)

      # Add all components to IssueSnapshot2Hotlist.
      # This is raw SQL to obviate passing FeaturesService down through
      #   the call stack wherever this function is called.
      # TODO(jrobbins): sort out dependencies between service classes.
      cnxn.Execute('''
        INSERT INTO IssueSnapshot2Hotlist (issuesnapshot_id, hotlist_id)
        SELECT %s, hotlist_id FROM Hotlist2Issue WHERE issue_id = %s
      ''', [issuesnapshot_id, issue.issue_id])

  def ExpungeHotlistsFromIssueSnapshots(self, cnxn, hotlist_ids, commit=True):
    """Expunge the existence of hotlists from issue snapshots.

    Args:
      cnxn: connection to SQL database.
      hotlist_ids: list of hotlist_ids for hotlists we want to delete.
      commit: set to False to skip the DB commit and do it in a caller.
    """
    vals_ph = sql.PlaceHolders(hotlist_ids)
    cnxn.Execute(
        'DELETE FROM IssueSnapshot2Hotlist '
        'WHERE hotlist_id IN ({vals_ph})'.format(vals_ph=vals_ph),
        hotlist_ids,
        commit=commit)

  def _currentTime(self):
    """This is a separate method so it can be mocked by tests."""
    return time.time()

  def _QueryToWhere(self, cnxn, services, project_config, query, canned_query,
                    project_ids):
    """Parses a query string into LEFT JOIN and WHERE conditions.

    Args:
      cnxn: A MonorailConnection instance.
      services: A Services instance.
      project_config: The configuration for the given project.
      query (string): The query to parse.
      canned_query (string): The supplied canned query.
      project_ids: The current project ID(s).

    Returns:
      1. A list of LEFT JOIN clauses for the SQL query.
      2. A list of WHERE clases for the SQL query.
      3. A list of query conditions that are unsupported with snapshots.
    """
    if not (query or canned_query):
      return [], [], []

    query = query or ''
    scope = canned_query or ''

    query_ast = query2ast.ParseUserQuery(query, scope,
        query2ast.BUILTIN_ISSUE_FIELDS, project_config)
    query_ast = ast2ast.PreprocessAST(cnxn, query_ast, project_ids,
        services, project_config)
    left_joins, where, unsupported = ast2select.BuildSQLQuery(query_ast,
        snapshot_mode=True)

    return left_joins, where, unsupported

  def _BuildSnapshotQuery(self, cols, where, joins, group_by, shard_id):
    """Given SQL arguments, executes a snapshot COUNT query."""
    stmt = sql.Statement.MakeSelect('IssueSnapshot', cols, distinct=True)
    stmt.AddJoinClauses(joins, left=True)
    stmt.AddWhereTerms(where + [('IssueSnapshot.shard = %s', [shard_id])])
    if group_by:
      stmt.AddGroupByTerms(group_by)
    stmt.SetLimitAndOffset(limit=settings.chart_query_max_rows, offset=0)
    stmt_str, stmt_args = stmt.Generate()
    if group_by:
      if group_by[0] == 'IssueSnapshot.is_open':
        count_stmt = ('SELECT IF(results.is_open = 1, "Opened", "Closed") ' \
          'AS bool_open, results.issue_count ' \
          'FROM (%s) AS results' % stmt_str)
      else:
        count_stmt = stmt_str
    else:
      count_stmt = 'SELECT COUNT(results.issue_id) FROM (%s) AS results' % (
        stmt_str)
    return count_stmt, stmt_args