def _generate_monthly_uniques(self, table, tables):
        idx = tables.index(table)

        # Join them all
        at = union_all(*[
            select([tbl]).where(tbl.c.message == "Ack")
            for tbl in tables[idx-29:idx+1]
        ])

        # Get uniques
        s = select([
            func.count(func.distinct(at.c.uaid_hash)).
            label("unique_count"),
            at.c.browser_os,
            at.c.browser_version,
        ]).\
            group_by(at.c.browser_os, at.c.browser_version)
        results = self._conn.execute(s).fetchall()
        if not results:
            return

        # Determine the date for this entry
        tname = table.name
        date_parts = [tname[-8:-4], tname[-4:-2], tname[-2:]]
        insert_date = "-".join(date_parts)

        self._conn.execute(monthly_rollup.insert(), [
            dict(date=insert_date,
                 count=x.unique_count,
                 browser_os=x.browser_os,
                 browser_version=x.browser_version)
            for x in results
        ])

def polymorphic_union(table_map, typecolname, aliasname='p_union'):
    """create a UNION statement used by a polymorphic mapper.
    
    See the SQLAlchemy advanced mapping docs for an example of how this is used."""
    colnames = util.Set()
    colnamemaps = {}
    types = {}
    for key in table_map.keys():
        table = table_map[key]

        # mysql doesnt like selecting from a select; make it an alias of the select
        if isinstance(table, sql.Select):
            table = table.alias()
            table_map[key] = table

        m = {}
        for c in table.c:
            colnames.add(c.name)
            m[c.name] = c
            types[c.name] = c.type
        colnamemaps[table] = m

    def col(name, table):
        try:
            return colnamemaps[table][name]
        except KeyError:
            return sql.cast(sql.null(), types[name]).label(name)

    result = []
    for type, table in table_map.iteritems():
        if typecolname is not None:
            result.append(sql.select([col(name, table) for name in colnames] + [sql.literal_column("'%s'" % type).label(typecolname)], from_obj=[table]))
        else:
            result.append(sql.select([col(name, table) for name in colnames], from_obj=[table]))
    return sql.union_all(*result).alias(aliasname)

def get_activity_query(user_id=None, session_id=None, test_id=None):
    # pylint: disable=no-member
    from .models import Activity, Comment, User

    _filter = functools.partial(_apply_filters, user_id=user_id, session_id=session_id, test_id=test_id)

    comments = select([
            literal_column("('comment:' || comment.id)").label('id'),
            literal_column(str(ACTION_COMMENTED)).label('action'),
            Comment.user_id.label('user_id'),
            Comment.session_id.label('session_id'),
            Comment.test_id.label('test_id'),
            Comment.timestamp.label('timestamp'),
            Comment.comment.label('text'),
            User.email.label('user_email'),
        ]).select_from(Comment.__table__.join(User, User.id == Comment.user_id))

    comments = _filter(Comment, comments)

    activity = select([
            literal_column("('activity:' || activity.id)").label('id'),
            Activity.action.label('action'),
            Activity.user_id.label('user_id'),
            Activity.session_id.label('session_id'),
            Activity.test_id.label('test_id'),
            Activity.timestamp.label('timestamp'),
            literal_column("NULL").label('text'),
            User.email.label('user_email'),
        ]).select_from(Activity.__table__.join(User, User.id == Activity.user_id))

    activity = _filter(Activity, activity)

    u = union_all(comments, activity).alias('u')

    return select([u]).order_by(u.c.timestamp)

def polymorphic_union(table_map, typecolname, aliasname="p_union", cast_nulls=True):
    """Create a ``UNION`` statement used by a polymorphic mapper.

    See  :ref:`concrete_inheritance` for an example of how
    this is used.

    :param table_map: mapping of polymorphic identities to
     :class:`.Table` objects.
    :param typecolname: string name of a "discriminator" column, which will be
     derived from the query, producing the polymorphic identity for each row.  If
     ``None``, no polymorphic discriminator is generated.
    :param aliasname: name of the :func:`~sqlalchemy.sql.expression.alias()`
     construct generated.
    :param cast_nulls: if True, non-existent columns, which are represented as labeled
     NULLs, will be passed into CAST.   This is a legacy behavior that is problematic
     on some backends such as Oracle - in which case it can be set to False.

    """

    colnames = util.OrderedSet()
    colnamemaps = {}
    types = {}
    for key in table_map.keys():
        table = table_map[key]

        # mysql doesnt like selecting from a select;
        # make it an alias of the select
        if isinstance(table, sql.Select):
            table = table.alias()
            table_map[key] = table

        m = {}
        for c in table.c:
            colnames.add(c.key)
            m[c.key] = c
            types[c.key] = c.type
        colnamemaps[table] = m

    def col(name, table):
        try:
            return colnamemaps[table][name]
        except KeyError:
            if cast_nulls:
                return sql.cast(sql.null(), types[name]).label(name)
            else:
                return sql.type_coerce(sql.null(), types[name]).label(name)

    result = []
    for type, table in table_map.iteritems():
        if typecolname is not None:
            result.append(
                sql.select(
                    [col(name, table) for name in colnames]
                    + [sql.literal_column(sql_util._quote_ddl_expr(type)).label(typecolname)],
                    from_obj=[table],
                )
            )
        else:
            result.append(sql.select([col(name, table) for name in colnames], from_obj=[table]))
    return sql.union_all(*result).alias(aliasname)

def auto_describes(docid):
    rty = db.relationship_types.alias()
    ide1 = db.identifiers.alias()
    ide2 = db.identifiers.alias()
    doc = db.documents.alias()
    pac = db.packages.alias()
    pfi = db.packages_files.alias()
    one = (
        select([
            doc.c.document_id,
            ide1.c.identifier_id        .label('left_identifier_id'),
            rty.c.relationship_type_id,
            ide2.c.identifier_id        .label('right_identifier_id'),
            ])
        .select_from(
            doc
            .join(pac, doc.c.package_id == pac.c.package_id)
            .join(ide1,
                (doc.c.document_id == ide1.c.document_id) &
                (doc.c.document_namespace_id == ide1.c.document_namespace_id)
                )
            .join(ide2,
                (pac.c.package_id == ide2.c.package_id) &
                (doc.c.document_namespace_id == ide2.c.document_namespace_id)
                )
            .join(rty, rty.c.name == 'DESCRIBES')
            )
        .where(doc.c.document_id == docid)
        )
    two = (
        select([
            doc.c.document_id,
            ide1.c.identifier_id        .label('left_identifier_id'),
            rty.c.relationship_type_id,
            ide2.c.identifier_id        .label('right_identifier_id'),
            ])
        .select_from(
            doc
            .join(pac, doc.c.package_id == pac.c.package_id)
            .join(pfi, pac.c.package_id == pfi.c.package_id, isouter=True)
            .join(ide1,
                (doc.c.document_id == ide1.c.document_id) &
                (doc.c.document_namespace_id == ide1.c.document_namespace_id)
                )
            .join(ide2,
                (pfi.c.package_file_id == ide2.c.package_file_id) &
                (doc.c.document_namespace_id == ide2.c.document_namespace_id)
                )
            .join(rty, rty.c.name == 'DESCRIBES')
            )
        .where(doc.c.document_id == docid)
        )
    return union_all(one, two)

    def _query_weekly_average(self, table, tables):
        # First see if we can find 6 days prior for a full week
        idx = tables.index(table)

        # For Python list math, 6 has 6 numbers before it as zero index
        # based, so 6 or larger is needed
        if idx < 6:
            return None

        # Get our weekly set together
        # Note that we add one to idx since list splicing needs one higher than
        # the index for right-side inclusive
        week_tables = union_all(*[
            select([tbl]).where(tbl.c.message == "Ack")
            for tbl in tables[idx-6:idx+1]
        ])

        # Calculate channels per user for the past week
        chans_per_user = select([
            week_tables.c.uaid_hash,
            func.count(func.distinct(week_tables.c.channel_id)).label("count")
        ]).\
            group_by(week_tables.c.uaid_hash)

        # Rank them into ntiles
        ranked = select([
            chans_per_user.c.uaid_hash,
            chans_per_user.c.count,
            func.ntile(100).over(order_by=text("count ASC")).label("rank"),
        ])

        # Remove the bottom/upper 5%, get sum/count for avg
        weekly_channels_stats = select([
            func.sum(ranked.c.count),
            func.count(ranked.c.uaid_hash),
        ]).\
            where(ranked.c.rank > 5).\
            where(ranked.c.rank <= 95)
        sums, count = self._conn.execute(weekly_channels_stats).fetchone()
        weekly_avg = Decimal(sums) / Decimal(count)
        return weekly_avg

#.........这里部分代码省略.........
        # We exclude messages on muted topics when finding the first unread
        # message in this narrow
        muting_conditions = exclude_muting_conditions(user_profile, narrow)
        if muting_conditions:
            condition = and_(condition, *muting_conditions)

        first_unread_query = query.where(condition)
        first_unread_query = first_unread_query.order_by(inner_msg_id_col.asc()).limit(1)
        first_unread_result = list(sa_conn.execute(first_unread_query).fetchall())
        if len(first_unread_result) > 0:
            anchor = first_unread_result[0][0]
        else:
            anchor = LARGER_THAN_MAX_MESSAGE_ID

    before_query = None
    after_query = None
    if num_before != 0:
        before_anchor = anchor
        if num_after != 0:
            # Don't include the anchor in both the before query and the after query
            before_anchor = anchor - 1
        before_query = query.where(inner_msg_id_col <= before_anchor) \
                            .order_by(inner_msg_id_col.desc()).limit(num_before)
    if num_after != 0:
        after_query = query.where(inner_msg_id_col >= anchor) \
                           .order_by(inner_msg_id_col.asc()).limit(num_after)

    if anchor == LARGER_THAN_MAX_MESSAGE_ID:
        # There's no need for an after_query if we're targeting just the target message.
        after_query = None

    if before_query is not None:
        if after_query is not None:
            query = union_all(before_query.self_group(), after_query.self_group())
        else:
            query = before_query
    elif after_query is not None:
        query = after_query
    else:
        # This can happen when a narrow is specified.
        query = query.where(inner_msg_id_col == anchor)

    main_query = alias(query)
    query = select(main_query.c, None, main_query).order_by(column("message_id").asc())
    # This is a hack to tag the query we use for testing
    query = query.prefix_with("/* get_old_messages */")
    query_result = list(sa_conn.execute(query).fetchall())

    # The following is a little messy, but ensures that the code paths
    # are similar regardless of the value of include_history.  The
    # 'user_messages' dictionary maps each message to the user's
    # UserMessage object for that message, which we will attach to the
    # rendered message dict before returning it.  We attempt to
    # bulk-fetch rendered message dicts from remote cache using the
    # 'messages' list.
    search_fields = dict() # type: Dict[int, Dict[str, Text]]
    message_ids = [] # type: List[int]
    user_message_flags = {} # type: Dict[int, List[str]]
    if include_history:
        message_ids = [row[0] for row in query_result]

        # TODO: This could be done with an outer join instead of two queries
        user_message_flags = dict((user_message.message_id, user_message.flags_list()) for user_message in
                                  UserMessage.objects.filter(user_profile=user_profile,
                                                             message__id__in=message_ids))
        for row in query_result: